rabbitmq使用topic exchange实现消息队列的水平扩展

一、背景

在通道发送消息的时候，由于我们没有维护人与通道节点的映射关系，所以是采用rabbitmq的广播模式，将消息发送到所有节点去消费。当查找到了通道，则发送；反之，则忽略。
其次，我们设置了消费的过期时间，一旦超过了指定时间比如3秒，还未被消费，则进入死信队列。目的是防止异常场景下（比如消费消息过慢），出现消息队列的堆积。

二、目标

1、增加Mq队列的数量，以缩短消息消费的等待时间
2、广播模式修改为主题模式，不同的消息路由到不同的队列，互不影响

三、设计思路

image.png

发送消息

基于房间ID，根据hash算法求模，得到其routing key。

接收消息

每个节点各4个队列，总共8个队列。但是两两队列有一个特点，就是有共同的标签。比如，节点1有一个队列是“first.channel.queue.xxx”，节点2也有一个first开头的队列，叫做“first.channel.queue.yyy”。

四、源码示例

1、发送消息

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private MQConfig mqConfig;

    public int getHashByRoomId(String roomId) {
        return Math.abs(MD5Utils.getMd5(roomId).hashCode()) % TOPIC_PREFIX_ARRAY.length;
    }

    @Override
    public void send(OutMessage message) {
        String routingKey = TOPIC_PREFIX_ARRAY[0];
        if (StringUtils.isNotEmpty(message.getRoomId())) {
            int iHash = getHashByRoomId(message.getRoomId());
            log.info("对房间ID:{}进行hash求模, {}", message.getRoomId(), iHash);
            routingKey = TOPIC_PREFIX_ARRAY[iHash];
        }

        rabbitTemplate.convertAndSend(mqConfig.getChannelExchange(), routingKey, message);
    }

2、接收消息

    @RabbitHandler
    @RabbitListener(queues = "#{mqConfig.firstChannelQueue}")
    public void process1(OutMessage outMessage) {
        log.info("队列11111收到消息：{}", JSON.toJSONString(outMessage));
        dealMessage(outMessage);
    }

    @RabbitHandler
    @RabbitListener(queues = "#{mqConfig.secondChannelQueue}")
    public void process2(OutMessage outMessage) {
        log.info("队列22222收到消息：{}", JSON.toJSONString(outMessage));
        dealMessage(outMessage);
    }

    @RabbitHandler
    @RabbitListener(queues = "#{mqConfig.thirdChannelQueue}")
    public void process3(OutMessage outMessage) {
        log.info("队列33333收到消息：{}", JSON.toJSONString(outMessage));
        dealMessage(outMessage);
    }

    @RabbitHandler
    @RabbitListener(queues = "#{mqConfig.fourChannelQueue}")
    public void process4(OutMessage outMessage) {
        log.info("队列44444收到消息：{}", JSON.toJSONString(outMessage));
        dealMessage(outMessage);
    }

3、队列和交换机的绑定

    public static final String[] TOPIC_PREFIX_ARRAY = {"first", "second", "third", "four"};

    /**
     * 通道队列名(第一个)
     */
    private String firstChannelQueue = TOPIC_PREFIX_ARRAY[0] + ".channel.queue." + HostUtils.getMac();

    /**
     * 通道队列名(第二个)
     */
    private String secondChannelQueue = TOPIC_PREFIX_ARRAY[1] + ".channel.queue." + HostUtils.getMac();

    /**
     * 通道队列名(第三个)
     */
    private String thirdChannelQueue = TOPIC_PREFIX_ARRAY[2] + ".channel.queue." + HostUtils.getMac();

    /**
     * 通道队列名(第四个)
     */
    private String fourChannelQueue = TOPIC_PREFIX_ARRAY[3] + ".channel.queue." + HostUtils.getMac();

    /**
     * 通道交换机名
     */
    private String channelExchange = "channel.exchange";

   // 队列的消息超时时间，如果超过5秒，则进入死信队列。
    private Map<String, Object> assembleParam(){
        Map<String, Object> params = new HashMap<>();
        params.put("x-message-ttl", 5 * 1000);
        params.put("x-dead-letter-exchange", deadExchange);
        params.put("x-dead-letter-routing-key", deadRoutingKey);
        return params;
    }

    @Bean
    public Queue firstChannelQueue() {
        return new Queue(firstChannelQueue, false, false, false, assembleParam());
    }

    @Bean
    public Queue secondChannelQueue() {
        return new Queue(secondChannelQueue, false, false, false, assembleParam());
    }

    @Bean
    public Queue thirdChannelQueue() {
        return new Queue(thirdChannelQueue, false, false, false, assembleParam());
    }

    @Bean
    public Queue fourChannelQueue() {
        return new Queue(fourChannelQueue, false, false, false, assembleParam());
    }

    /**
     * 通道交换机 topic主题模式
     *
     * @return: org.springframework.amqp.core.Exchange
     **/
    @Bean
    public Exchange channelExchange() {
        return ExchangeBuilder.topicExchange(channelExchange).durable(true).build();
    }

   
    @Bean
    public Binding firstChannelBinding() {
        return BindingBuilder.bind(firstChannelQueue()).to(channelExchange()).with(TOPIC_PREFIX_ARRAY[0] + ".#").noargs();
    }

    @Bean
    public Binding secondChannelBinding() {
        return BindingBuilder.bind(secondChannelQueue()).to(channelExchange()).with(TOPIC_PREFIX_ARRAY[1] + ".#").noargs();
    }

    @Bean
    public Binding thirdChannelBinding() {
        return BindingBuilder.bind(thirdChannelQueue()).to(channelExchange()).with(TOPIC_PREFIX_ARRAY[2] + ".#").noargs();
    }

    @Bean
    public Binding fourChannelBinding() {
        return BindingBuilder.bind(fourChannelQueue()).to(channelExchange()).with(TOPIC_PREFIX_ARRAY[3] + ".#").noargs();
    }

总结

mq消费者有4个队列，也即4个消费线程池，隔离性更好。这里是以房间为hash取模，如果是多租户的业务模型，也可以业务ID为hash取模。

对比topic主题模式，广播模式是一种特殊的模式，广播模式的routingkey都是相同的值，也就不用传。

参考链接

https://zhuanlan.zhihu.com/p/582701519