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

RabbitMQ 确认模式（Acknowledgements Mode）详解

article 2024/10/25 5:49:55

RabbitMQ 是一款流行的开源消息代理软件，它实现了高级消息队列协议（AMQP）。在消息传递过程中，确保消息被正确处理是至关重要的。RabbitMQ 提供了多种机制来确保消息的可靠性，其中确认模式（Acknowledgements Mode）是一个关键特性。

什么是确认模式？

确认模式（Acknowledgements Mode）允许消费者在成功处理消息后显式地向 RabbitMQ 服务器发送确认信号（ack）。只有在收到确认信号后，RabbitMQ 服务器才会从队列中删除该消息。如果消费者未能发送确认信号（例如，由于消费者崩溃或网络故障），RabbitMQ 会认为消息尚未被处理，并在适当的时候重新发送消息。

RabbitMQ 提供了三种主要的确认模式：

手动确认（Manual Acknowledgement）：消费者需要显式地发送确认信号。
自动确认（Automatic Acknowledgement）：消息一旦被消费者接收，立即自动确认。
批量确认（Batch Acknowledgement）：消费者可以一次确认多条消息。

为什么使用确认模式？

确保消息不丢失：即使消费者崩溃，消息也会重新发送。
提高可靠性：通过控制确认时机，可以更好地管理消息处理流程。
灵活性：可以根据不同的业务需求选择不同的确认模式。

Java 代码示例

下面是一个使用 Java 和 Spring AMQP 实现 RabbitMQ 确认模式的示例。

Maven 依赖

首先，在你的 pom.xml 文件中添加 Spring AMQP 依赖：

<dependencies>  
    <dependency>  
        <groupId>org.springframework.boot</groupId>  
        <artifactId>spring-boot-starter-amqp</artifactId>  
    </dependency>  
    <!-- 其他依赖 -->  
</dependencies>

配置 RabbitMQ

在 application.properties 文件中配置 RabbitMQ 连接信息：

spring.rabbitmq.host=localhost  
spring.rabbitmq.port=5672  
spring.rabbitmq.username=guest  
spring.rabbitmq.password=guest

配置类

配置一个队列、交换机和绑定：

import org.springframework.amqp.core.*;  
import org.springframework.context.annotation.Bean;  
import org.springframework.context.annotation.Configuration;  
  
@Configuration  
public class RabbitMQConfig {  
  
    public static final String QUEUE_NAME = "exampleQueue";  
    public static final String EXCHANGE_NAME = "exampleExchange";  
    public static final String ROUTING_KEY = "exampleRoutingKey";  
  
    @Bean  
    public Queue queue() {  
        return new Queue(QUEUE_NAME, true); // 持久化队列  
    }  
  
    @Bean  
    public DirectExchange exchange() {  
        return new DirectExchange(EXCHANGE_NAME);  
    }  
  
    @Bean  
    public Binding binding(Queue queue, DirectExchange exchange) {  
        return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);  
    }  
}

消息监听器

使用手动确认模式来监听队列：

import org.springframework.amqp.rabbit.annotation.RabbitListener;  
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;  
import org.springframework.amqp.rabbit.core.RabbitTemplate;  
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener.ReturnCallback;  
import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;  
import org.springframework.amqp.rabbit.listener.config.SimpleRabbitListenerContainerFactory;  
import org.springframework.amqp.rabbit.listener.config.SimpleRabbitListenerEndpoint;  
import org.springframework.amqp.rabbit.connection.ConnectionFactory;  
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;  
import org.springframework.amqp.support.converter.MessageConverter;  
import org.springframework.amqp.support.converter.SimpleMessageConverter;  
import org.springframework.beans.factory.annotation.Autowired;  
import org.springframework.context.annotation.Bean;  
import org.springframework.context.annotation.Configuration;  
  
import com.rabbitmq.client.Channel;  
  
@Configuration  
public class ListenerConfig {  
  
    @Autowired  
    private ConnectionFactory connectionFactory;  
  
    @Bean  
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {  
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();  
        factory.setConnectionFactory(connectionFactory);  
        factory.setMessageConverter(jsonMessageConverter());  
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL); // 设置为手动确认模式  
        return factory;  
    }  
  
    @Bean  
    public MessageConverter jsonMessageConverter() {  
        return new SimpleMessageConverter();  
    }  
  
    @RabbitListener(queues = RabbitMQConfig.QUEUE_NAME, containerFactory = "rabbitListenerContainerFactory")  
    public void listen(String message, Channel channel, org.springframework.amqp.core.Message rabbitMessage) throws Exception {  
        try {  
            // 处理消息  
            System.out.println("Received <" + message + ">");  
              
            // 发送确认信号  
            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);  
        } catch (Exception e) {  
            // 发送拒绝信号，并设置为重新入队（requeue）  
            channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);  
            throw e;  
        }  
    }  
}

发送消息

编写一个简单的控制器来发送消息到 RabbitMQ：

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;  
    }  
}