RocketMQ的一些使用理解

1.RocketMQ的生产者生产负载策略（3种）
(1)SelectMessageQueueByHash （一致性hash）
(2)SelectMessageQueueByMachineRoom （机器随机）
(3)SelectMessageQueueByRandom （随机）

第1种一致性hash算法是对所有的队列进行hash计算
缺点：(1)如果出现队列增减可能会导致顺序消息断层。
(2)在多broker情况下，可能导致broker分布不均匀，即我加了一个broker维度的hash分层。

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;/*** 参考SelectMessageQueueByHash，很大可能放在同一个broker上了。* 我们hash选择broker来避免这个情况*/
public class SelectMessageQueueByBrokerHash implements MessageQueueSelector {@Overridepublic MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {Map<String, Map<Integer, MessageQueue>> brokerQueueMap = new HashMap<>(16);mqs.forEach(queue -> brokerQueueMap.computeIfAbsent(queue.getBrokerName(), key -> new HashMap<>(32)).put(queue.getQueueId(), queue));List<String> brokerNames = new ArrayList<>(brokerQueueMap.keySet());Collections.sort(brokerNames);// hash选择broker，再hash选择queueint brokerIndex = indexForBroker(arg, brokerNames);Map<Integer, MessageQueue> queueMap = brokerQueueMap.get(brokerNames.get(brokerIndex));int queueIndex = indexForQueue(arg, queueMap);return queueMap.get(queueIndex);}private int indexForBroker(Object arg, List<String> brokerNames) {int hashCode = arg.hashCode();hashCode = hashCode < 0 ? Math.abs(hashCode) : hashCode;return hashCode % brokerNames.size();}/*** 参考one to one hash算法* https://blog.51cto.com/u_14398214/5076158*/private int indexForQueue(Object arg, Map<Integer, MessageQueue> queueMap) {String key = String.valueOf(arg);int hash, i;for (hash = 0, i = 0; i < key.length(); ++i) {hash += key.charAt(i);hash += (hash << 10);hash ^= (hash >> 6);}hash += (hash << 3);hash ^= (hash >> 11);hash += (hash << 15);hash = hash < 0 ? Math.abs(hash) : hash;return hash % queueMap.size();}
}

2.consumer消费负载均衡策略：
(1）默认采用平均分配方法来实现负载均衡
　如果consumer个数和queue数不对等时：
consumer个数比queue个数多，多个consumer消费一个queue
consumer个数和queue个数一样，一个consumer消费一个queue
consumer个数比queue个数少，一个consumer消费多个queue
(2）AllocateMessageQueueConsistentHash：一致性哈希
何时reblance：
(1）当一个consumer宕机最多20秒执行reblance，新consumer重新消费
(2）当有新consumer接入时，立即执行reblance。

参考：
https://www.cnblogs.com/jijiecong/p/15182736.html