专栏首页码匠的流水账聊聊rocketmq的AllocateMessageQueueAveragely
原创

聊聊rocketmq的AllocateMessageQueueAveragely

本文主要研究一下rocketmq的AllocateMessageQueueAveragely

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方法

AllocateMessageQueueAveragely

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

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

doc

原创声明,本文系作者授权云+社区发表,未经许可,不得转载。

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

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 聊聊rocketmq的AllocateMessageQueueAveragely

    本文主要研究一下rocketmq的AllocateMessageQueueAveragely

    codecraft
  • java9系列(四)Process API更新

    /Library/Java/JavaVirtualMachines/jdk-9.0.4.jdk/Contents/Home/lib/src.zip!/java....

    codecraft
  • 聊聊dubbo-go的apolloConfiguration

    dubbo-go-v1.4.2/config_center/apollo/impl.go

    codecraft
  • 欧里几德及扩展欧里几德算法

    欧几里德算法 欧几里德算法又称辗转相除法,用于计算两个整数a,b的最大公约数。 基本算法:设a=qb+r,其中a,b,q,r都是整数,则gcd(a,b)=gcd...

    Angel_Kitty
  • vue 使用 cli 工具构建项目

    章鱼喵
  • 易到13亿悬案始末 | “此刻,真正需要全力以赴的是,替司机、用户解决当下面临的困难和问题。”

    镁客网
  • 5G 网络将会怎样改变世界?

    近日,腾讯无线网络与物联网技术负责人李秋香与高校科研教授、产业链、运营商等各行业的嘉宾一起参与了知乎「 科技共振之 5G+ 」活动,除了专业的5G探讨,也聊了不...

    云加社区
  • 刘强东解读京东Q2财报(电话会议全文)

    大数据文摘
  • sublime txt 3 自定义设置

    平时用的最多文本工具就是sublime了,当前版本3126。 风格、编码、字符大小、换行、是否使用空格代替tab等都可以在此设置 我自己的设置如下:

    大数据工程师-公子
  • Vue.js动画在项目使用的两个示例

    vue.js在2016年大放异彩,我们 Qlippie 直播平台进来使用vue.js开发了一些页面,效果拔群,在这里跟大家分享在项目中用vue.js开发的两个动...

    李萌

扫码关注云+社区

领取腾讯云代金券