参考

有思考的大佬博客
https://www.jianshu.com/p/f071d5069059

PushConsumer限流：

Consumer启动的时候，独立的后台线程PullMessageService不断地从阻塞队列—pullRequestQueue中获取PullRequest请求
并通过网络通信模块发送Pull消息的RPC请求给Broker端。

public class PullMessageService extends ServiceThread {
    private final LinkedBlockingQueue<PullRequest> pullRequestQueue = new LinkedBlockingQueue<PullRequest>();
    @Override
    public void run() {
        while (!this.isStopped()) {
                PullRequest pullRequest = this.pullRequestQueue.take();
                this.pullMessage(pullRequest);
        }
    }
}

当前PullRequest对应的Queue（负载均衡分配好了Queue），如果该Queue的缓存消息数量超过1000，或者缓存Size超过，会暂缓Pull投递，50ms后再重试
DefaultMQPushConsumerImpl.class

    //默认一个MessageCacheQueue最多Cache1000个消息
    private int pullThresholdForQueue = 1000;
    //默认一个MessageCacheQueue最多Cache100M大小的消息
    private int pullThresholdSizeForQueue = 100;

    private static final long PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL = 50;


public void pullMessage(final PullRequest pullRequest) {
    final ProcessQueue processQueue = pullRequest.getProcessQueue();
    long cachedMessageCount = processQueue.getMsgCount().get();
    long cachedMessageSizeInMiB = processQueue.getMsgSize().get() / (1024 * 1024);

    if (cachedMessageCount > this.defaultMQPushConsumer.getPullThresholdForQueue()) {
        this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
        return;
    }
    
    if (cachedMessageSizeInMiB > this.defaultMQPushConsumer.getPullThresholdSizeForQueue()) {
        this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
        return;
    }
}

负载均衡代码详解

Consumer后台独立线程 RebalanceService，默认20s更新一次负载均衡策略（因为可能有新的Consumer加入到ConsumerGroup或者宕机了）

public class RebalanceService extends ServiceThread {
    private static long waitInterval =Long.parseLong(System.getProperty("rocketmq.client.rebalance.waitInterval", "20000"));
  
  @Override
    public void run() {
        while (!this.isStopped()) {
            this.waitForRunning(waitInterval);
            this.mqClientFactory.doRebalance();
        }
    }
}

然后为每个Consumer做负载均衡

public void doRebalance() {
        for (Map.Entry<String, MQConsumerInner> entry : this.consumerTable.entrySet()) {
            MQConsumerInner impl = entry.getValue();
            impl.doRebalance();
        }
    }

为每个Consumer的每个Topic做负载均衡

public void doRebalance(final boolean isOrder) {
    Map<String, SubscriptionData> subTable = this.getSubscriptionInner();
    for (final Map.Entry<String, SubscriptionData> entry : subTable.entrySet()) {
        final String topic = entry.getKey();
        this.rebalanceByTopic(topic, isOrder);
    }
}

RebalanceImpl类的rebalanceByTopic()方法，为每个Consumer的每个Topic做负载均衡
负载均衡是针对Topic来划分的，一个Topic比如4个Consumer，Topic1下有4个Queue，每个Consumer负责一个Queue

    //根据Topic获取有几个Queue
    Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
    //该Topic下有几个Consumer订阅了
    List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);

    //先对Topic下的消息消费队列、消费者Id排序，然后用消息队列分配策略算法（默认为：消息队列的平均分配算法），计算出待拉取的消息队列
    Collections.sort(mqAll);
    Collections.sort(cidAll);

默认按页数平均分配

将所有MessageQueue排好序类似于记录，将所有消费端Consumer排好序类似页数
并求出每一页需要包含的平均size和每个页面记录的范围range
最后遍历整个range而计算出当前Consumer端应该分配到的记录（这里即为：MessageQueue）

    List<MessageQueue> result = new ArrayList<MessageQueue>();
    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;

RocketMQ 负载均衡和限流

RocketMQ 负载均衡和限流

参考

PushConsumer限流：

负载均衡代码详解