前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >聊聊rocketmq的AllocateMessageQueueAveragely

聊聊rocketmq的AllocateMessageQueueAveragely

作者头像
code4it
发布2019-11-30 21:27:30
5900
发布2019-11-30 21:27:30
举报
文章被收录于专栏:码匠的流水账码匠的流水账

本文主要研究一下rocketmq的AllocateMessageQueueAveragely

AllocateMessageQueueStrategy

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

代码语言:javascript
复制
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方法

AllocateMessageQueueAveragely

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

代码语言:javascript
复制
public class AllocateMessageQueueAveragely 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);
        int mod = mqAll.size() % cidAll.size();
        int averageSize =
            mqAll.size() <= cidAll.size() ? 1 : (mod > 0 && index < mod ? mqAll.size() / cidAll.size()
                + 1 : mqAll.size() / cidAll.size());
        int startIndex = (mod > 0 && index < mod) ? index * averageSize : index * averageSize + mod;
        int range = Math.min(averageSize, mqAll.size() - startIndex);
        for (int i = 0; i < range; i++) {
            result.add(mqAll.get((startIndex + i) % mqAll.size()));
        }
        return result;
    }

    @Override
    public String getName() {
        return "AVG";
    }
}
  • AllocateMessageQueueAveragely实现了AllocateMessageQueueStrategy接口,其getName返还AVG,其allocate方法首先计算index(cidAll.indexOf(currentCID)),然后计算mod(mqAll.size() % cidAll.size())
  • 之后计算averageSize,若mqAll.size()小于等于cidAll.size()则取1,否则取mod > 0 && index < mod ? mqAll.size() / cidAll.size() + 1 : mqAll.size() / cidAll.size()
  • 最后计算startIndex((mod > 0 && index < mod) ? index * averageSize : index * averageSize + mod)及range(Math.min(averageSize, mqAll.size() - startIndex));之后按range进行循环,将下标为(startIndex + i) % mqAll.size()的消息添加到result

小结

AllocateMessageQueueAveragely实现了AllocateMessageQueueStrategy接口,其getName返还AVG,其allocate方法按range进行循环,将下标为(startIndex + i) % mqAll.size()的消息添加到result

本文参与 腾讯云自媒体分享计划,分享自微信公众号。
原始发表:2019-11-28,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 码匠的流水账 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体分享计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • AllocateMessageQueueStrategy
  • AllocateMessageQueueAveragely
  • 小结
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档