Spring Boot 整合 RabbitMQ：手动 ACK 与 QoS 配置详解

在分布式系统中，消息队列（Message Queue）是实现异步通信的重要组件。RabbitMQ 作为一个功能强大的消息代理，提供了多种消息传递模式和丰富的配置选项。在生产环境中，为了确保消息的可靠传递，我们通常需要配置手动确认（Manual Acknowledgment）和消息预取（QoS）。

本文将详细介绍如何在 Spring Boot 项目中通过 application.properties 配置文件来配置 RabbitMQ 的手动 ACK 和 QoS，并通过实际代码示例帮助你理解这些配置的作用和使用场景。

1. 手动 ACK 与 QoS 的概念

1.1 手动 ACK（Manual Acknowledgment）

在 RabbitMQ 中，消费者在接收到消息后，默认会自动确认（Auto Acknowledgment）消息。这种方式虽然简单，但在某些场景下可能会导致消息丢失或重复消费。例如，如果消费者在处理消息时发生异常，消息可能会被错误地确认，导致消息丢失。

手动确认机制允许消费者在处理完消息后，手动向 RabbitMQ 发送确认信号。这种方式可以确保消息只有在被成功处理后才会被确认，从而提高消息的可靠性。

1.2 QoS（Quality of Service）

QoS 配置用于控制消费者从 RabbitMQ 中预取消息的数量。默认情况下，RabbitMQ 会将所有消息一次性推送给消费者，这可能会导致消费者处理不过来，甚至导致内存溢出。

通过配置 QoS，我们可以限制消费者一次从 RabbitMQ 中预取的消息数量，从而避免消息堆积和资源浪费。

2. Spring Boot 整合 RabbitMQ 配置

2.1 环境准备

在开始之前，请确保你已经安装了以下环境：

Java 8 或更高版本
Maven 或 Gradle
RabbitMQ 服务器（可以通过 Docker 快速启动）

你可以通过以下命令使用 Docker 启动 RabbitMQ 服务器：

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management

启动后，可以通过 http://localhost:15672 访问 RabbitMQ 的管理界面，默认用户名和密码为 guest/guest。

2.2 创建 Spring Boot 项目

你可以通过 Spring Initializr 快速创建一个 Spring Boot 项目。选择以下依赖：

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-web</artifactId>
</dependency>

生成项目后，导入到你的 IDE 中。

2.3 配置 RabbitMQ

在 application.properties 文件中添加 RabbitMQ 的配置：

# RabbitMQ 配置
spring.rabbitmq.host=192.168.200.142
spring.rabbitmq.port=5672
spring.rabbitmq.virtual-host=/test
spring.rabbitmq.username=test
spring.rabbitmq.password=test

# 消费者配置
# 设置手动 ACK
spring.rabbitmq.listener.simple.acknowledge-mode=manual
# 设置 QoS，每次预取 1 条消息
spring.rabbitmq.listener.simple.prefetch=1
# 设置并发消费者数量
spring.rabbitmq.listener.simple.concurrency=3
# 设置最大并发消费者数量
spring.rabbitmq.listener.simple.max-concurrency=10

2.4 定义消息队列和交换机

在 Spring Boot 中，我们可以通过 @Bean 注解来定义 RabbitMQ 的队列、交换机和绑定关系。

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {

    public static final String QUEUE_NAME = "spring-boot-queue";
    public static final String EXCHANGE_NAME = "spring-boot-exchange";
    public static final String ROUTING_KEY = "spring-boot-routing-key";

    @Bean
    public Queue queue() {
        return new Queue(QUEUE_NAME, true); // 设置队列为持久化
    }

    @Bean
    public TopicExchange exchange() {
        return new TopicExchange(EXCHANGE_NAME);
    }

    @Bean
    public Binding binding(Queue queue, TopicExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
    }
}

2.5 发送消息

在 Spring Boot 中，我们可以通过 RabbitTemplate 来发送消息。

import com.allen.config.RabbitMQConfig;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class MessageController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/send")
    public String sendMessage(@RequestParam String message) {
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME, RabbitMQConfig.ROUTING_KEY, message);
        return "Message sent: " + message;
    }
}

2.6 接收消息并手动 ACK

通过 @RabbitListener 注解，我们可以监听指定的队列并处理接收到的消息。在手动 ACK 模式下，我们需要手动确认消息。

import com.allen.config.RabbitMQConfig;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component
public class MessageListener {

    @RabbitListener(queues = RabbitMQConfig.QUEUE_NAME)
    public void receiveMessage(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
        try {
            System.out.println("Received message: " + message);
            // 模拟业务处理
            Thread.sleep(1000);
            channel.basicAck(deliveryTag, false); // 手动确认消息
        } catch (Exception e) {
            channel.basicNack(deliveryTag, false, true); // 拒绝消息并重新入队
        }
    }
}