rocketmq--死信队列

在RocketMQ中，死信队列（Dead Letter Queue，DLQ）用于存放无法成功消费的消息。当消息重试消费次数超过设定的阈值后，消息将被转移到死信队列。使用Spring Boot集成RocketMQ时，可以通过以下步骤来处理死信队列中的消息。

首先，在pom.xml中添加RocketMQ Spring Boot Starter的依赖：

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>你的版本号</version>
</dependency>

接下来，在application.properties或application.yml中配置RocketMQ的相关属性：

# application.properties 示例
rocketmq.name-server=你的NameServer地址
rocketmq.producer.group=你的生产者分组
rocketmq.consumer.group=你的消费者分组

然后，创建生产者服务发送消息：

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class ProducerService {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void send(String topic, String message) {
        rocketMQTemplate.convertAndSend(topic, message);
    }
}

创建消费者服务来消费消息，并在无法消费的情况下抛出异常，触发重试机制：

import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;

@Service
@RocketMQMessageListener(
    topic = "test-topic",
    consumerGroup = "你的消费者分组",
    consumeMode = ConsumeMode.ORDERLY,
    selectorType = SelectorType.TAG,
    selectorExpression = "*"
)
public class ConsumerService implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        try {
            // 处理消息的逻辑
            System.out.println("Received message: " + message);
            // 假设在这里发生了异常，触发重试
            throw new RuntimeException("Failed to process message.");
        } catch (Exception e) {
            // 异常处理逻辑，例如记录日志等
            System.err.println("Error processing message: " + e.getMessage());
            // 抛出异常，让RocketMQ进行重试
            throw e;
        }
    }
}

在上面的消费者服务中，我们通过抛出异常来模拟消费失败的场景。RocketMQ默认会进行重试，重试次数可以通过consumer.maxReconsumeTimes属性进行配置。当超过最大重试次数后，消息会被转移到死信队列。

处理死信队列的消息，需要监听 %DLQ%你的消费者分组 主题：

@Service
@RocketMQMessageListener(
    topic = "%DLQ%你的消费者分组",
    consumerGroup = "你的消费者分组-DLQ",
    consumeMode = ConsumeMode.ORDERLY,
    selectorType = SelectorType.TAG,
    selectorExpression = "*"
)
public class DeadLetterQueueConsumerService implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        // 处理死信队列的消息
        System.out.println("Received dead letter message: " + message);
        // 在这里编写处理死信消息的逻辑，例如记录日志、报警、人工干预等
    }
}

以上就是一个简单的死信队列处理例子。需要注意的是，死信队列的主题名称是固定格式的，以 %DLQ% 开头，后面跟上原消费者分组的名称。另外，在实际的业务场景中，你可能需要根据消息内容或错误类型进行更详细的错误处理逻辑。