分布式中间件：RabbitMQ确认消费机制

在分布式系统中，消息队列是实现异步通信和系统解耦的重要组件。RabbitMQ 作为一款功能强大的消息队列中间件，提供了丰富的特性来保证消息的可靠传输和消费。其中，确认消费机制是确保消息被正确处理的关键环节。本文将深入探讨 RabbitMQ 的确认消费机制，并给出不同场景下的配置示例。

确认消费机制概述

RabbitMQ 的确认消费机制主要涉及两个方面：生产者消息确认和消费者消息确认。生产者消息确认用于确保消息成功发送到 RabbitMQ 服务器，而消费者消息确认则用于确保消息被消费者正确处理。

消费者确认模式

RabbitMQ 提供了三种消费者确认模式：

AcknowledgeMode.AUTO：自动确认模式。当消费者接收到消息后，RabbitMQ 会自动将消息标记为已消费，无论消费者是否成功处理该消息。这种模式简单方便，但可能会导致消息丢失，因为如果消费者在处理消息时出现异常，消息已经被确认，无法再次处理。
AcknowledgeMode.MANUAL：手动确认模式。消费者需要显式地调用 channel.basicAck() 方法来确认消息已经被成功处理，或者调用 channel.basicNack() 或 channel.basicReject() 方法来拒绝消息。这种模式可以确保消息不会丢失，但需要开发者手动处理确认逻辑。
AcknowledgeMode.NONE：无确认模式。RabbitMQ 不会等待消费者的确认，一旦消息被发送给消费者，就会将其标记为已消费。这种模式适用于对消息可靠性要求不高的场景。

配置示例

单一消费者实例配置

import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitMQConfig {private final ConnectionFactory connectionFactory;public RabbitMQConfig(ConnectionFactory connectionFactory) {this.connectionFactory = connectionFactory;}@Bean("singleListenerContainer")public SimpleRabbitListenerContainerFactory singleListenerContainer() {// 创建一个监听容器工厂SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();// 设置连接工厂factory.setConnectionFactory(connectionFactory);// 设置消息转换器factory.setMessageConverter(new Jackson2JsonMessageConverter());// 设置并发消费factory.setConcurrentConsumers(1);// 设置最大并发消费factory.setMaxConcurrentConsumers(1);// 设置最大单条消费消息factory.setPrefetchCount(1);return factory;}
}

在这个配置中，我们创建了一个单一消费者实例的监听容器工厂。通过设置 ConcurrentConsumers 和 MaxConcurrentConsumers 为 1，确保只有一个消费者实例在处理消息。PrefetchCount 设置为 1，表示消费者一次只从队列中获取一条消息，保证消息的顺序处理。

多个消费者实例配置

import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.AcknowledgeMode;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitMQConfig {private final ConnectionFactory connectionFactory;public RabbitMQConfig(ConnectionFactory connectionFactory) {this.connectionFactory = connectionFactory;}@Bean("multiListenerContainer")public SimpleRabbitListenerContainerFactory multiListenerContainer() {// 创建一个监听容器工厂SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();// 设置连接工厂factory.setConnectionFactory(connectionFactory);// 设置消息转换器factory.setMessageConverter(new Jackson2JsonMessageConverter());// 设置手动提交factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);// 设置并发消费factory.setConcurrentConsumers(10);// 设置最大并发消费factory.setMaxConcurrentConsumers(20);// 设置最大单条消费消息factory.setPrefetchCount(10);return factory;}
}

在多个消费者实例的配置中，我们将 ConcurrentConsumers 设置为 10，MaxConcurrentConsumers 设置为 20，表示初始有 10 个消费者实例，最多可以扩展到 20 个。PrefetchCount 设置为 10，允许每个消费者一次从队列中获取 10 条消息，提高消费效率。同时，我们将确认模式设置为 AcknowledgeMode.MANUAL，需要开发者手动处理消息确认。

自定义 RabbitMQ 发送消息组件

import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitMQConfig {private final ConnectionFactory connectionFactory;public RabbitMQConfig(ConnectionFactory connectionFactory) {this.connectionFactory = connectionFactory;}@Beanpublic RabbitTemplate rabbitTemplate() {// 设置消息发送确认connectionFactory.setPublisherConfirms(true);// 设置消息发送返回connectionFactory.setPublisherReturns(true);// 创建 rabbitTemplateRabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);// 设置消息发送确认回调rabbitTemplate.setMandatory(true);// 设置消息发送确认回调rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {if (ack) {System.out.println("消息发送成功");} else {System.out.println("消息发送失败:" + cause + correlationData.toString());}});// 设置消息发送返回回调rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {System.out.println("消息丢失：exchange:" + exchange + ",route:" + routingKey + ",replyCode:" + replyCode + ",replyText:" + replyText);});// 设置消息转换器rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());return rabbitTemplate;}
}

在这个配置中，我们自定义了一个 RabbitTemplate 用于发送消息。通过设置 PublisherConfirms 和 PublisherReturns 为 true，开启生产者消息确认和返回机制。ConfirmCallback 用于处理消息发送确认结果，ReturnCallback 用于处理消息无法路由到队列的情况。