opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-21 05:44:03 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #3498] Make messages in reviveTopic more evenly written to different queues #3499

Browse Source
This commit is contained in:
cserwen
2022-02-08 10:38:41 +08:00
committed by RongtongJin
parent 78a3ed7b0e
commit fe88fb36a3
1 changed files with 3 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java
+3 -1
View File
@@ -92,6 +92,7 @@ public class PopMessageProcessor implements NettyRequestProcessor {
private PopLongPollingService popLongPollingService;
private PopBufferMergeService popBufferMergeService;
private QueueLockManager queueLockManager;
private AtomicLong ckMessageNumber;
public PopMessageProcessor(final BrokerController brokerController) {
this.brokerController = brokerController;
@@ -104,6 +105,7 @@ public class PopMessageProcessor implements NettyRequestProcessor {
this.popLongPollingService = new PopLongPollingService();
this.queueLockManager = new QueueLockManager();
this.popBufferMergeService = new PopBufferMergeService(this.brokerController, this);
this.ckMessageNumber = new AtomicLong();
}
public PopLongPollingService getPopLongPollingService() {
@@ -350,7 +352,7 @@ public class PopMessageProcessor implements NettyRequestProcessor {
if (requestHeader.isOrder()) {
reviveQid = KeyBuilder.POP_ORDER_REVIVE_QUEUE;
} else {
reviveQid = randomQ % this.brokerController.getBrokerConfig().getReviveQueueNum();
reviveQid = (int) Math.abs(ckMessageNumber.getAndIncrement() % this.brokerController.getBrokerConfig().getReviveQueueNum());
}
GetMessageResult getMessageResult = new GetMessageResult();