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