当前位置：首页 > news >正文

在使用 RabbitMQ 时，手动确认消息和死信队列

news 来源：原创 2025/5/9 8:50:21

在使用 RabbitMQ 时，手动确认消息和死信队列是两个常见的需求。下面是一个使用 Spring Boot 的示例，展示如何手动确认消息以及如何使用死信队列。

1. 手动确认消息

在 RabbitMQ 中，默认情况下，消息是自动确认的。为了手动确认消息，我们需要将 acknowledge-mode 设置为 manual，并在消费者中手动调用 basicAck 或 basicNack。

1.1 配置

首先，在 application.yml 中配置 RabbitMQ：

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    listener:
      simple:
        acknowledge-mode: manual

1.2 消费者

接下来，创建一个消费者类，手动确认消息：

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component
public class RabbitMQConsumer {

    @RabbitListener(queues = "myQueue")
    public void receiveMessage(Message message, Channel channel) throws IOException {
        try {
            // 处理消息
            String body = new String(message.getBody());
            System.out.println("Received message: " + body);

            // 手动确认消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            // 处理异常，拒绝消息并重新入队
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        }
    }
}

2. 死信队列

死信队列（Dead Letter Queue, DLQ）用于处理无法被正常消费的消息。当消息被拒绝、过期或队列达到最大长度时，消息会被路由到死信队列。

2.1 配置死信队列

首先，配置一个普通队列和一个死信队列：

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {

    // 普通队列
    @Bean
    public Queue myQueue() {
        return QueueBuilder.durable("myQueue")
                .withArgument("x-dead-letter-exchange", "dlxExchange") // 死信交换机
                .withArgument("x-dead-letter-routing-key", "dlxQueue") // 死信路由键
                .build();
    }

    // 死信队列
    @Bean
    public Queue dlxQueue() {
        return QueueBuilder.durable("dlxQueue").build();
    }

    // 普通交换机
    @Bean
    public DirectExchange myExchange() {
        return new DirectExchange("myExchange");
    }

    // 死信交换机
    @Bean
    public DirectExchange dlxExchange() {
        return new DirectExchange("dlxExchange");
    }

    // 绑定普通队列到普通交换机
    @Bean
    public Binding binding() {
        return BindingBuilder.bind(myQueue()).to(myExchange()).with("myRoutingKey");
    }

    // 绑定死信队列到死信交换机
    @Bean
    public Binding dlxBinding() {
        return BindingBuilder.bind(dlxQueue()).to(dlxExchange()).with("dlxQueue");
    }
}

2.2 消费者处理死信队列

创建一个消费者来处理死信队列中的消息：

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component
public class DLQConsumer {

    @RabbitListener(queues = "dlxQueue")
    public void handleDeadLetterMessage(Message message, Channel channel) throws IOException {
        String body = new String(message.getBody());
        System.out.println("Received dead letter message: " + body);

        // 手动确认死信消息
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
}

3. 测试

你可以通过发送消息到 myQueue 并手动拒绝消息来测试死信队列的功能。被拒绝的消息会被路由到 dlxQueue，并由 DLQConsumer 处理。

4. 总结

手动确认消息：通过设置 acknowledge-mode 为 manual，并在消费者中手动调用 basicAck 或 basicNack 来确认或拒绝消息。
死信队列：通过配置 x-dead-letter-exchange 和 x-dead-letter-routing-key，将无法处理的消息路由到死信队列。

通过这种方式，你可以更好地控制消息的处理流程，并确保不会丢失重要的消息。