聊聊rocketmq的AllocateMessageQueueAveragelyByCircle

序

本文主要研究一下rocketmq的AllocateMessageQueueAveragelyByCircle

AllocateMessageQueueStrategy

rocketmq-client-4.5.2-sources.jar!/org/apache/rocketmq/client/consumer/AllocateMessageQueueStrategy.java

public interface AllocateMessageQueueStrategy {

    /**
     * Allocating by consumer id
     *
     * @param consumerGroup current consumer group
     * @param currentCID current consumer id
     * @param mqAll message queue set in current topic
     * @param cidAll consumer set in current consumer group
     * @return The allocate result of given strategy
     */
    List<MessageQueue> allocate(
        final String consumerGroup,
        final String currentCID,
        final List<MessageQueue> mqAll,
        final List<String> cidAll
    );

    /**
     * Algorithm name
     *
     * @return The strategy name
     */
    String getName();
}
複製代碼

• AllocateMessageQueueStrategy定義了allocate、getName方法

AllocateMessageQueueAveragelyByCircle

rocketmq-client-4.5.2-sources.jar!/org/apache/rocketmq/client/consumer/rebalance/AllocateMessageQueueAveragelyByCircle.java

public class AllocateMessageQueueAveragelyByCircle implements AllocateMessageQueueStrategy {
    private final InternalLogger log = ClientLogger.getLog();

    @Override
    public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll,
        List<String> cidAll) {
        if (currentCID == null || currentCID.length() < 1) {
            throw new IllegalArgumentException("currentCID is empty");
        }
        if (mqAll == null || mqAll.isEmpty()) {
            throw new IllegalArgumentException("mqAll is null or mqAll empty");
        }
        if (cidAll == null || cidAll.isEmpty()) {
            throw new IllegalArgumentException("cidAll is null or cidAll empty");
        }

        List<MessageQueue> result = new ArrayList<MessageQueue>();
        if (!cidAll.contains(currentCID)) {
            log.info("[BUG] ConsumerGroup: {} The consumerId: {} not in cidAll: {}",
                consumerGroup,
                currentCID,
                cidAll);
            return result;
        }

        int index = cidAll.indexOf(currentCID);
        for (int i = index; i < mqAll.size(); i++) {
            if (i % cidAll.size() == index) {
                result.add(mqAll.get(i));
            }
        }
        return result;
    }

    @Override
    public String getName() {
        return "AVG_BY_CIRCLE";
    }
}
複製代碼

• AllocateMessageQueueAveragelyByCircle實現了AllocateMessageQueueStrategy接口，其getName返還AVG_BY_CIRCLE，其allocate方法首先計算index(cidAll.indexOf(currentCID))，以後從index開始到mqAll.size()，針對知足i % cidAll.size() == index的下標的msg添加到result

小結

AllocateMessageQueueAveragelyByCircle實現了AllocateMessageQueueStrategy接口，其getName返還AVG_BY_CIRCLE，其allocate方法首先計算index(cidAll.indexOf(currentCID))，以後從index開始到mqAll.size()，針對知足i % cidAll.size() == index的下標的msg添加到result

doc

• AllocateMessageQueueAveragelyByCircle