1+ /*
2+ * Licensed to the Apache Software Foundation (ASF) under one or more
3+ * contributor license agreements. See the NOTICE file distributed with
4+ * this work for additional information regarding copyright ownership.
5+ * The ASF licenses this file to You under the Apache License, Version 2.0
6+ * (the "License"); you may not use this file except in compliance with
7+ * the License. You may obtain a copy of the License at
8+ *
9+ * http://www.apache.org/licenses/LICENSE-2.0
10+ *
11+ * Unless required by applicable law or agreed to in writing, software
12+ * distributed under the License is distributed on an "AS IS" BASIS,
13+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+ * See the License for the specific language governing permissions and
15+ * limitations under the License.
16+ */
17+ package org .apache .rocketmq .proxy .service .route ;
18+
19+ import java .util .List ;
20+ import java .util .Objects ;
21+ import java .util .concurrent .atomic .AtomicInteger ;
22+ import org .apache .commons .lang3 .tuple .Pair ;
23+ import org .apache .rocketmq .common .message .MessageQueue ;
24+
25+ @ FunctionalInterface
26+ public interface MessageQueuePenalizer <Q extends MessageQueue > {
27+
28+ /**
29+ * Returns the penalty value for the given MessageQueue; lower is better.
30+ */
31+ int penaltyOf (Q messageQueue );
32+
33+ /**
34+ * Aggregates penalties from multiple penalizers for the same MessageQueue (by summing them up).
35+ */
36+ static <Q extends MessageQueue > int evaluatePenalty (Q messageQueue , List <MessageQueuePenalizer <Q >> penalizers ) {
37+ Objects .requireNonNull (messageQueue , "messageQueue" );
38+ if (penalizers == null || penalizers .isEmpty ()) {
39+ return 0 ;
40+ }
41+ int sum = 0 ;
42+ for (MessageQueuePenalizer <Q > p : penalizers ) {
43+ sum += p .penaltyOf (messageQueue );
44+ }
45+ return sum ;
46+ }
47+
48+ /**
49+ * Selects the queue with the lowest evaluated penalty from the given queue list.
50+ *
51+ * <p>The method iterates through all queues exactly once, but starts from a rotating index
52+ * derived from {@code startIndex} (round-robin) to avoid always scanning from position 0 .</p>
53+ *
54+ * <p>For each queue, it computes a penalty via {@link #evaluatePenalty} using
55+ * the provided {@code penalizers}. The queue with the smallest penalty is selected.</p>
56+ *
57+ * <p>Short-circuit rule: if any queue has a {@code penalty<= 0}, it is returned immediately,
58+ * since no better result than 0 is expected.</p>
59+ *
60+ * @param queues candidate queues to select from
61+ * @param penalizers penalty evaluators applied to each queue
62+ * @param startIndex atomic counter used to determine the rotating start position (round-robin)
63+ * @param <Q> queue type
64+ * @return a {@code Pair} of (selected queue, penalty), or {@code null} if {@code queues} is null/empty
65+ */
66+ static <Q extends MessageQueue > Pair <Q , Integer > selectLeastPenalty (List <Q > queues ,
67+ List <MessageQueuePenalizer <Q >> penalizers , AtomicInteger startIndex ) {
68+ if (queues == null || queues .isEmpty ()) {
69+ return null ;
70+ }
71+ Q bestQueue = null ;
72+ int bestPenalty = Integer .MAX_VALUE ;
73+
74+ for (int i = 0 ; i < queues .size (); i ++) {
75+ int index = Math .floorMod (startIndex .getAndIncrement (), queues .size ());
76+ Q messageQueue = queues .get (index );
77+ int penalty = evaluatePenalty (messageQueue , penalizers );
78+
79+ // Short-circuit: cannot do better than 0
80+ if (penalty <= 0 ) {
81+ return Pair .of (messageQueue , penalty );
82+ }
83+
84+ if (penalty < bestPenalty ) {
85+ bestPenalty = penalty ;
86+ bestQueue = messageQueue ;
87+ }
88+ }
89+ return Pair .of (bestQueue , bestPenalty );
90+ }
91+
92+ /**
93+ * Selects a queue with the lowest computed penalty from multiple priority groups.
94+ *
95+ * <p>The input {@code queuesWithPriority} is a list of queue groups ordered by priority.
96+ * For each priority group, this method delegates to {@link #selectLeastPenalty} to pick the best queue
97+ * within that group and obtain its penalty.</p>
98+ *
99+ * <p>Short-circuit rule: if any priority group yields a queue whose {@code penalty <= 0},
100+ * that result is returned immediately.</p>
101+ *
102+ * <p>Otherwise, it returns the queue with the smallest positive penalty among all groups.
103+ * If multiple groups produce the same minimum penalty, the first encountered one wins.</p>
104+ *
105+ * @param queuesWithPriority priority-ordered groups of queues; each inner list represents one priority level
106+ * @param penalizers penalty calculators used by {@code selectLeastPenalty} to score queues
107+ * @param startIndex round-robin start index forwarded to {@code selectLeastPenalty} to reduce contention/hotspots
108+ * @param <Q> queue type
109+ * @return a {@code Pair} of (selected queue, penalty), or {@code null} if {@code queuesWithPriority} is null/empty
110+ */
111+ static <Q extends MessageQueue > Pair <Q , Integer > selectLeastPenaltyWithPriority (List <List <Q >> queuesWithPriority ,
112+ List <MessageQueuePenalizer <Q >> penalizers , AtomicInteger startIndex ) {
113+ if (queuesWithPriority == null || queuesWithPriority .isEmpty ()) {
114+ return null ;
115+ }
116+ if (queuesWithPriority .size () == 1 ) {
117+ return selectLeastPenalty (queuesWithPriority .get (0 ), penalizers , startIndex );
118+ }
119+ Q bestQueue = null ;
120+ int bestPenalty = Integer .MAX_VALUE ;
121+ for (List <Q > queues : queuesWithPriority ) {
122+ Pair <Q , Integer > queueAndPenalty = selectLeastPenalty (queues , penalizers , startIndex );
123+ int penalty = queueAndPenalty .getRight ();
124+ if (queueAndPenalty .getRight () <= 0 ) {
125+ return queueAndPenalty ;
126+ }
127+ if (penalty < bestPenalty ) {
128+ bestPenalty = penalty ;
129+ bestQueue = queueAndPenalty .getLeft ();
130+ }
131+ }
132+ return Pair .of (bestQueue , bestPenalty );
133+ }
134+ }
0 commit comments