Apache
/
activemq-artemis
mirror of https://github.com/apache/activemq-artemis.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

ARTEMIS-2664 Simplify the credits acquiring. 

Replace the AtomicInteger with an int. Indeed deliveredAcks is used only by
the acknowledge method and it is only executed by the EpollEventLoop thread
bounded with the relative connection channel.
This commit is contained in:
brusdev 2020-03-19 12:11:28 +01:00 committed by Clebert Suconic
parent 30d521152d
commit fc6402613d
1 changed files with 8 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQConsumer.java
View File

@ -68,7 +68,7 @@ public class AMQConsumer {
private int prefetchSize;
private final AtomicInteger currentWindow;
private final AtomicInteger deliveredAcks;
private int deliveredAcks;
private long messagePullSequence = 0;
private final AtomicReference<MessagePullHandler> messagePullHandler = new AtomicReference<>(null);
//internal means we don't expose
@ -88,7 +88,7 @@ public class AMQConsumer {
this.scheduledPool = scheduledPool;
this.prefetchSize = info.getPrefetchSize();
this.currentWindow = new AtomicInteger(prefetchSize);
this.deliveredAcks = new AtomicInteger(0);
this.deliveredAcks = 0;
if (prefetchSize == 0) {
messagePullHandler.set(new MessagePullHandler());
}
@ -300,18 +300,15 @@ public class AMQConsumer {
List<MessageReference> ackList = serverConsumer.getDeliveringReferencesBasedOnProtocol(removeReferences, first, last);
if (removeReferences && (ack.isIndividualAck() || ack.isStandardAck() || ack.isPoisonAck())) {
this.deliveredAcks.getAndUpdate(deliveredAcks -> {
if (deliveredAcks >= ackList.size()) {
return deliveredAcks - ackList.size();
}
if (deliveredAcks < ackList.size()) {
acquireCredit(ackList.size() - deliveredAcks);
return 0;
});
deliveredAcks = 0;
} else {
deliveredAcks -= ackList.size();
}
} else {
if (ack.isDeliveredAck()) {
this.deliveredAcks.addAndGet(ack.getMessageCount());
this.deliveredAcks += ack.getMessageCount();
}
acquireCredit(ack.getMessageCount());