RocketMQ(八)再均衡

public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll,List<String> cidAll) {

		//省略参数校验代码...

		// 获取当前Consumer在消费者集合(cidAll)中下标的位置
		int index = cidAll.indexOf(currentCID);

		/**
	     * Queue数除以Consumer数取余
	     * 如果mod等于0，说明Queue可以被平均分配
	     * 如果mod大于0，说明不可以被平均分配，一定程度平均分配后仍剩余mod个队列
	     */
		int mod = mqAll.size() % cidAll.size();

        /**
	     * averageSize代表当前Consumer可以得到的队列数
	     * Queue数小于等于Consumer，averageSize=1，也就是平均分配1个队列(虽然靠后的Consumer可能啥都没分到)
	     * Queue数大于Consumer的情况下，根据Queue能否被Consumer平均分配继续处理:
	     * 如果能平均分配，直接计算Queue数除以Consumer数得到平均值
	     * 如果不能平均分配，继续判断当前Consumer在cidAll的index是否小于mod:
	     *		如果小于则说明最大化平均后剩余的mod个队列还有自己一份，averageSize = Queue数除以Consumer数取整后+1
	     *		如果大于等于说明最大化平均后剩余的mod个队列和自己没缘分了，averageSize = Queue数除以Consumer数取整
	     */
		int averageSize =
			mqAll.size() <= cidAll.size() ? 1 : (mod > 0 && index < mod ? mqAll.size() / cidAll.size()
				+ 1 : mqAll.size() / cidAll.size());

		/**
	     * 后续会遍历mqAll集合提取属于当前Consumer的Queue，这里需要计算遍历的起始坐标
	     * 如果不能平均消费，且剩余的mod个队列有自己一份，startIndex = index * averageSize
	     * 如果不能平均消费，且剩余的mod个队列和自己无缘，startIndex = index * averageSize + mod
	     */
		int startIndex = (mod > 0 && index < mod) ? index * averageSize : index * averageSize + mod;

		// 根据Math.min()计算消费者最终需要消费的数量
		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;
}

public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll, List<String> cidAll) {

        //省略参数校验代码...

		// 获取当前Consumer在消费者集合(cidAll)中下标的位置
		int index = cidAll.indexOf(currentCID);

		// 以index作为起点遍历Queue集合
		for (int i = index; i < mqAll.size(); i++) {
			// 对下标取模(mod), 如果与index相等, 则存储到result集合中
			if (i % cidAll.size() == index) {
				result.add(mqAll.get(i));
			}
		}
		return result;
}