前景回顾
【mq】从零开始实现 mq-02-如何实现生产者调用消费者？
【mq】从零开始实现 mq-03-引入 broker 中间人
【mq】从零开始实现 mq-06-消费者心跳检测 heartbeat
【mq】从零开始实现 mq-07-负载均衡 load balance
为什么需要负载均衡
大家好，我是老马。
这一节让我们看一下如何实现 MQ 的负载均衡。
为什么需要负载均衡呢？
作用
负载均衡最核心的作用：
（1）可以避免单点故障
（2）可以让请求均分的分散到每一个节点
实现思路
负载均衡实现的方式比较多，最简单的就是随机选择一个。
拓展阅读：
从零手写实现负载均衡 http://houbb.github.io/2020/06/19/load-balance-03-hand-write
MQ 中用到负载均衡的地方
生产者发送
生产者发送消息时，可以发送给任一 broker。
broker 推送给消费者
broker 接收到消息以后，在推送给消费者时，也可以任一选择一个。
消费者的消费 ACK
消费者消费完，状态回执给 broker，可以选择任一一个。
消息黏连
有些消息比较特殊，比如需要保证消费的有序性，可以通过 shardingKey 的方式，在负载的时候固定到指定的片区。
代码实现
生产者发送
统一调整获取 channel 的方法。
@Override
public Channel getChannel(String key) {
// 等待启动完成
while (!statusManager.status()) {
log.debug("等待初始化完成...");
DateUtil.sleep(100);
}
RpcChannelFuture rpcChannelFuture = RandomUtils.loadBalance(this.loadBalance,
channelFutureList, key);
return rpcChannelFuture.getChannelFuture().channel();
}
工具类实现为核心实现：
/**
* 负载均衡
*
* @param list 列表
* @param key 分片键
* @return 结果
* @since 0.0.7
*/
public static <T extends IServer> T loadBalance(final ILoadBalance<T> loadBalance,
final List<T> list, String key) {
if(CollectionUtil.isEmpty(list)) {
return null;
}
if(StringUtil.isEmpty(key)) {
LoadBalanceContext<T> loadBalanceContext = LoadBalanceContext.<T>newInstance()
.servers(list);
return loadBalance.select(loadBalanceContext);
}
// 获取 code
int hashCode = Objects.hash(key);
int index = hashCode % list.size();
return list.get(index);
}
如果指定了 shardingKey，那么根据 shadringKey 进行 hash 判断。
如果没有，则进行默认的负载均衡策略。
Broker 消息推送给消费者
消费者订阅列表的获取：
@Override
public List<Channel> getSubscribeList(MqMessage mqMessage) {
final String topicName = mqMessage.getTopic();
Set<ConsumerSubscribeBo> set = subscribeMap.get(topicName);
if(CollectionUtil.isEmpty(set)) {
return Collections.emptyList();
}
//2. 获取匹配的 tag 列表
final List<String> tagNameList = mqMessage.getTags();
Map<String, List<ConsumerSubscribeBo>> groupMap = new HashMap<>();
for(ConsumerSubscribeBo bo : set) {
String tagRegex = bo.getTagRegex();
if(hasMatch(tagNameList, tagRegex)) {
//TODO: 这种设置模式，统一添加处理 haven
String groupName = bo.getGroupName();
List<ConsumerSubscribeBo> list = groupMap.get(groupName);
if(list == null) {
list = new ArrayList<>();
}
list.add(bo);
groupMap.put(groupName, list);
}
}
//3. 按照 groupName 分组之后，每一组只随机返回一个。最好应该调整为以 shardingkey 选择
final String shardingKey = mqMessage.getShardingKey();
List<Channel> channelList = new ArrayList<>();
for(Map.Entry<String, List<ConsumerSubscribeBo>> entry : groupMap.entrySet()) {
List<ConsumerSubscribeBo> list = entry.getValue();
ConsumerSubscribeBo bo = RandomUtils.loadBalance(loadBalance, list, shardingKey);
final String channelId = bo.getChannelId();
BrokerServiceEntryChannel entryChannel = registerMap.get(channelId);
if(entryChannel == null) {
log.warn("channelId: {} 对应的通道信息为空", channelId);
continue;
}
channelList.add(entryChannel.getChannel());
}
return channelList;
}
核心逻辑：RandomUtils.loadBalance(loadBalance, list, shardingKey); 获取，其他的保持不变。
消费者 ACK
消费者也是类似的，获取 channel 的方式调整如下：
public Channel getChannel(String key) {
// 等待启动完成
while (!statusManager.status()) {
log.debug("等待初始化完成...");
DateUtil.sleep(100);
}
RpcChannelFuture rpcChannelFuture = RandomUtils.loadBalance(loadBalance,
channelFutureList, key);
return rpcChannelFuture.getChannelFuture().channel();
}
小结
负载均衡在分布式服务中，是必备的特性之一。实现的原理并不算复杂。
希望本文对你有所帮助，如果喜欢，欢迎点赞收藏转发一波。
我是老马，期待与你的下次重逢。
开源地址
The message queue in java.(java 简易版本 mq 实现) https://github.com/houbb/mq