67 lines
3.5 KiB
Diff
67 lines
3.5 KiB
Diff
# UNDF: UNDF-2026-000000617
|
|
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java
|
|
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java
|
|
@@ -113,6 +113,7 @@ public class Queue extends BaseDestination implements Task, UsageListener, Index
|
|
protected final List<Subscription> consumers = new ArrayList<Subscription>(50);
|
|
+ private int consumerRotationIndex = 0; /* round-robin cursor; replaces O(C) remove+add per message */
|
|
private final ReentrantReadWriteLock consumersLock = new ReentrantReadWriteLock();
|
|
|
|
@@ -2282,13 +2282,16 @@ public class Queue extends BaseDestination implements Task, UsageListener, Index
|
|
// If it got dispatched, rotate the consumer list to get round robin
|
|
// distribution.
|
|
if (target != null && !strictOrderDispatch && consumers.size() > 1
|
|
&& !dispatchSelector.isExclusiveConsumer(target)) {
|
|
- consumersLock.writeLock().lock();
|
|
- try {
|
|
- if (removeFromConsumerList(target)) {
|
|
- addToConsumerList(target);
|
|
- consumers = new ArrayList<Subscription>(this.consumers);
|
|
- }
|
|
- } finally {
|
|
- consumersLock.writeLock().unlock();
|
|
+ if (!useConsumerPriority) {
|
|
+ /* O(1) rotation: advance cursor instead of O(C) remove + O(C log C) sort.
|
|
+ * The local snapshot in consumers[] is already ordered; on next dispatch
|
|
+ * we start from the next position via the cursor on this.consumers. */
|
|
+ consumersLock.writeLock().lock();
|
|
+ try {
|
|
+ int idx = this.consumers.indexOf(target);
|
|
+ if (idx >= 0) {
|
|
+ consumerRotationIndex = (idx + 1) % this.consumers.size();
|
|
+ }
|
|
+ } finally {
|
|
+ consumersLock.writeLock().unlock();
|
|
+ }
|
|
+ } else {
|
|
+ /* Priority mode: re-sort needed when priorities differ; keep existing path. */
|
|
+ consumersLock.writeLock().lock();
|
|
+ try {
|
|
+ if (removeFromConsumerList(target)) {
|
|
+ addToConsumerList(target);
|
|
+ consumers = new ArrayList<Subscription>(this.consumers);
|
|
+ }
|
|
+ } finally {
|
|
+ consumersLock.writeLock().unlock();
|
|
+ }
|
|
}
|
|
}
|
|
|
|
@@ -2229,7 +2229,12 @@ public class Queue extends BaseDestination implements Task, UsageListener, Index
|
|
consumersLock.readLock().lock();
|
|
try {
|
|
if (this.consumers.isEmpty()) {
|
|
return list;
|
|
}
|
|
- consumers = new ArrayList<Subscription>(this.consumers);
|
|
+ if (!useConsumerPriority && consumerRotationIndex > 0 && consumerRotationIndex < this.consumers.size()) {
|
|
+ /* Start snapshot from rotation cursor for O(1) round-robin positioning. */
|
|
+ List<Subscription> rotated = new ArrayList<Subscription>(this.consumers.size());
|
|
+ rotated.addAll(this.consumers.subList(consumerRotationIndex, this.consumers.size()));
|
|
+ rotated.addAll(this.consumers.subList(0, consumerRotationIndex));
|
|
+ consumers = rotated;
|
|
+ } else {
|
|
+ consumers = new ArrayList<Subscription>(this.consumers);
|
|
+ }
|
|
} finally {
|
|
consumersLock.readLock().unlock();
|
|
}
|