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 {
@Override
public 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选择queue
int 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