RocketMQ生产者发送消息失败是如何规避重试时发送给同一个broker？

一、引言

随着微服务架构的发展，消息队列已经成为分布式系统中的重要组件之一。RocketMQ这款消息中间件，性能超群、坚如磐石，早已成为分布式系统开发领域的“香饽饽”，被各种各样的项目团队热烈追捧并广泛应用着。这篇东西咱们要掰开了揉碎了讲讲怎么用Spring Boot给RocketMQ发生产者消息，而且还要重点聊聊万一消息发送失败，在进行重试时怎么巧妙避免再次把消息送到同一条Broker上。

二、背景介绍

在使用RocketMQ进行消息发送时，通常情况下我们会设置一个重试机制，以应对可能出现的各种网络、服务器等不可控因素导致的消息发送失败。但是，如果不加把劲儿控制一下，这种重试机制就很可能像一群疯狂的粉丝不断涌向同一个明星那样，让同一台Broker承受不住压力，这样一来，严重的性能问题也就随之爆发喽。所以呢，我们得在重试这套流程里头动点脑筋，加点策略进去。这样一来，当生产者小哥遇到状况失败了，就能尽可能地绕开那些已经闹情绪的Broker家伙，不让它们再添乱。

三、解决方案

为了解决这个问题，我们可以采用以下两种方案：

1. 设置全局的Broker列表

在创建Producer实例时，我们可以指定一个包含所有Broker地址的列表，然后在每次重试时随机选择一个Broker进行发送。这样可以有效地避免过多的请求集中在某一台Broker上，从而降低对Broker的压力。以下是具体的代码实现：

List<String> brokers = Arrays.asList("broker-a", "broker-b", "broker-c");
Set<String> failedBrokers = new HashSet<>();
public void sendMessage(String topic, String body) {
    for (int i = 0; i < RETRY_TIMES; i++) {
        Random random = new Random();
        String broker = brokers.get(random.nextInt(brokers.size()));
        if (!failedBrokers.contains(broker)) {
            try {
                producer.send(topic, new MessageQueue(topic, broker, 0), new DefaultMQProducer.SendResultHandler() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        System.out.println("Message send success");
                    }
                    @Override
                    public void onException(Throwable e) {
                        System.out.println("Message send exception: " + e.getMessage());
                        failedBrokers.add(broker);
                    }
                });
                return;
            } catch (Exception e) {
                System.out.println("Message send exception: " + e.getMessage());
                failedBrokers.add(broker);
            }
        }
    }
    System.out.println("Message send fail after retrying");
}

在上述代码中，我们首先定义了一个包含所有Broker地址的列表`brokers`，然后在每次重试时随机选择一个Broker进行发送。如果该Broker在之前已经出现过错误，则将其添加到已失败的Broker集合中。在下一次重试时，我们不再选择这个Broker。

2. 利用RocketMQ提供的重试机制

除了手动设置Broker列表之外，我们还可以利用RocketMQ自带的重试机制来达到相同的效果。简单来说，我们可以搞个“RetryMessageListener”这个小家伙来监听一下，它的任务就是专门盯着RocketMQ发出的消息。一旦消息发送失败，它就负责把这些失败的消息重新拉出来再试一次，确保消息能顺利送达。在用这个监听器的时候，我们就能知道当前的Broker是不是还在重试列表里混呢。如果发现它在的话，那咱们就麻利地把它从列表里揪出来；要是不是，那就继续让它“回炉重造”，执行重试操作呗。以下是具体的代码实现：

public class RetryMessageListener implements MQListenerMessageConsumeOrderlyCallback {
    private Set<String> retryBrokers = new HashSet<>();
    private List<String> brokers = Arrays.asList("broker-a", "broker-b", "broker-c");
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        for (String broker : brokers) {
            if (retryBrokers.contains(broker)) {
                retryBrokers.remove(broker);
            }
        }
        for (String broker : retryBrokers) {
            try {
                producer.send(msgs.get(0).getTopic(), new MessageQueue(msgs.get(0).getTopic(), broker, 0),