前景回顾

为什么需要负载均衡

大家好，我是老马。

这一节让我们看一下如何实现 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

拓展阅读

rpc-从零开始实现 rpc https://github.com/houbb/rpc

巴特西

【mq】从零开始实现 mq-07-负载均衡 load balance

前景回顾

为什么需要负载均衡

作用

实现思路

MQ 中用到负载均衡的地方

生产者发送

broker 推送给消费者

消费者的消费 ACK

消息黏连

代码实现

生产者发送

Broker 消息推送给消费者

消费者 ACK

小结

开源地址

拓展阅读

最新文章

热门文章