消息回发：RocketMQ 的二级保险

后端

2024-01-02 18:36:38

RocketMQ 中的消息回发：可靠投递消息的保障

RocketMQ 作为一款分布式消息队列，以其高性能、高可靠性著称。为了确保消息的可靠投递，尤其是在消息消费失败的情况下，RocketMQ 提供了一套自动的消息回发机制。

消息回发的情形

RocketMQ 的消息回发主要在以下两种情况下触发：

消费者消费消息时抛出异常： 当消费者在消费消息时遇到错误（例如业务代码执行失败、消费者进程崩溃），回发机制将自动触发，将消息重新发送回 Broker。
消费者主动回发消息： 消费者也可以通过调用 API 主动将消息回发给 Broker。例如，当消费者发现某个消息无法消费时，它可以将消息回发，以便 Broker 重新发送给其他消费者。

消息回发的过程

消息回发的过程如下：

消费者消费消息时出现异常。
RocketMQ 检测到异常，触发消息回发机制。
回发的消息被重新发送到 Broker 的消费队列中。
其他消费者从队列中重新消费消息。

消息回发对消费者端的影响

当消息被回发后，消费者端需要重新消费该消息。这可能带来一定的性能开销，因为消费者端需要从消费队列的头部重新开始消费，这意味着它需要处理所有未消费的消息。

不过，消费者端可以通过特殊处理回发消息来降低重复消费的性能开销。例如，它可以维护一个已消费消息的集合，如果回发消息已经在集合中，则忽略该消息。

代码示例

public class ConsumerWithRetry {
    // ... 省略部分代码

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        for (MessageExt msg : msgs) {
            try {
                // 业务逻辑
                System.out.println("Received message: " + msg.getMsgId());
                // 重试次数
                int retryTimes = RETRY_TIMES_MAP.getOrDefault(msg.getMsgId(), 0);
                if (retryTimes >= 3) {
                    System.out.println("Message: " + msg.getMsgId() + " has been retried 3 times, will be discarded");
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                RETRY_TIMES_MAP.put(msg.getMsgId(), retryTimes + 1);
            } catch (Exception e) {
                System.err.println("Failed to consume message: " + msg.getMsgId());
                e.printStackTrace();
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    // ... 省略部分代码
}