RabbitMQ（四）消息Ack确认机制

2023-10-26

RabbitMQ（四）消息Ack确认机制

确认种类

RabbitMQ的消息确认有两种。

消息发送确认：这种是用来确认生产者将消息发送给交换器，交换器传递给队列的过程中，消息是否成功投递。发送确认分为两步，一是确认是否到达交换器，二是确认是否到达队列。
消费接收确认。这种是确认消费者是否成功消费了队列中的消息。

环境配置

为了测试，我们先配置rabbit环境

引入Maven依赖


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

配置文件

spring.application.name=rabbitmq
server.port=8084

spring.rabbitmq.host=192.168.3.253
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123456

Rabbit配置RabbitConfig.java

package com.lay.rabbitmqtwo.config;

/**
 * @Description:
 * @Author: lay
 * @Date: Created in 13:34 2018/12/20
 * @Modified By:IntelliJ IDEA
 */
@Configuration
public class RabbitConfig {
    public static final String CONFIRM_QUEUE_A = "confirm_queue_A";
    public static final String CONFIRM_QUEUE_B = "confirm_queue_B";
    public static final String CONFIRM_EXCHANGE = "confirm_topic_exchange";
    private static final String CONFIRM_QUEUE_A_RoutingKey="topic.message";
    private static final String CONFIRM_QUEUE_B_RoutingKey="topic.#";
    
    //Json格式转换
    private static final MessageConverter jsonMessageConverter=new Jackson2JsonMessageConverter();

    @Autowired
    private RabbitTemplate rabbitTemplate;

    //测试队列A
    @Bean
    public Queue confirmQueryA() {
        return new Queue(CONFIRM_QUEUE_A);
    }

    //测试队列B
    @Bean
    public Queue confirmQueryB() {
        return new Queue(CONFIRM_QUEUE_B);
    }

    //测试交换机,类型为topic
    @Bean
    TopicExchange confirmTopicExchange() {
        return new TopicExchange(CONFIRM_EXCHANGE);
    }

    //绑定测试交换机和测试队列A
    @Bean
    Binding bindingConfirmExchangeA(Queue confirmQueryA, TopicExchange confirmTopicExchange) {
        return BindingBuilder.bind(confirmQueryA).to(confirmTopicExchange).with(CONFIRM_QUEUE_A_RoutingKey);
    }

    //绑定测试交换机和测试队列B
    @Bean
    Binding bindingConfirmExchangeB(Queue confirmQueryB, TopicExchange confirmTopicExchange) {
        return BindingBuilder.bind(confirmQueryB).to(confirmTopicExchange).with(CONFIRM_QUEUE_B_RoutingKey);
    }


}

消息发送确认

（1）ConfirmCallback

通过实现ConfirmCallBack接口，消息发送到交换器Exchange后触发回调。

ConfirmCallBackHandler.java

package com.lay.rabbitmqtwo.config;

import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

/**
 * @Description:通过实现ConfirmCallBack接口，消息发送到交换器Exchange后触发回调。
 * @Author: lay
 * @Date: Created in 10:20 2018/12/20
 * @Modified By:IntelliJ IDEA
 */

public class ConfirmCallBackHandler implements RabbitTemplate.ConfirmCallback {
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        System.out.println("消息唯一标识："+correlationData);
        System.out.println("确认结果："+ack);
        System.out.println("失败原因："+cause);
    }
}

在RabbitConfig中配置RabbitTempalte

    //初始化加载方法，对RabbitTemplate进行配置
    @PostConstruct
    void rabbitTemplate(){
        //消息发送确认，发送到交换器Exchange后触发回调
        rabbitTemplate.setConfirmCallback(new ConfirmCallBackHandler());
    }

该功能需要开启确认，spring-boot中配置如下：

#消息发送交换机确认
spring.rabbitmq.publisher-confirms = true

（2）ReturnCallback

通过实现ReturnCallback接口，如果消息从交换器发送到对应队列失败时触发（比如根据发送消息时指定的routingKey找不到队列时会触发）

ReturnCallBackHandler.java

package com.lay.rabbitmqtwo.config;

/**
 * @Description:通过实现ReturnCallback接口
 * 如果消息从交换器发送到对应队列失败时触发（比如根据发送消息时指定的routingKey找不到队列时会触发）
 * @Author: lay
 * @Date: Created in 10:31 2018/12/20
 * @Modified By:IntelliJ IDEA
 */

public class ReturnCallBackHandler implements RabbitTemplate.ReturnCallback {
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        System.out.println("消息主体 message："+message);
        System.out.println("应答码 replyCode: ："+replyCode);
        System.out.println("原因描述 replyText："+replyText);
        System.out.println("交换机 exchange："+exchange);
        System.out.println("消息使用的路由键 routingKey："+routingKey);
    }
}

在RabbitConfig中配置RabbitTempalte

    //初始化加载方法，对RabbitTemplate进行配置
    @PostConstruct
    void rabbitTemplate(){
        //消息发送确认，发送到交换器Exchange后触发回调
        rabbitTemplate.setConfirmCallback(new ConfirmCallBackHandler());
        //消息发送确认，如果消息从交换器发送到对应队列失败时触发（比如根据发送消息时指定的routingKey找不到队列时会触发）
        rabbitTemplate.setReturnCallback(new ReturnCallBackHandler());
        //自定义格式转换
        //rabbitTemplate.setMessageConverter(jsonMessageConverter);
    }

使用该功能需要开启确认，spring-boot中配置如下：

#消息发送队列回调
spring.rabbitmq.publisher-returns = true

消息接收确认

（1）确认模式

AcknowledgeMode.NONE：不确认
AcknowledgeMode.AUTO：自动确认
AcknowledgeMode.MANUAL：手动确认

spring-boot中配置方法：

spring.rabbitmq.listener.simple.acknowledge-mode = manual

如果使用自定义监听容器

 //RabbitMQ监听容器
    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        //设置并发
        factory.setConcurrentConsumers(1);
        SimpleMessageListenerContainer s=new SimpleMessageListenerContainer();
        //最大并发
        factory.setMaxConcurrentConsumers(1);
        //消息接收——手动确认
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        //设置超时
        factory.setReceiveTimeout(2000L);
        //设置重试间隔
        factory.setFailedDeclarationRetryInterval(3000L);
        //监听自定义格式转换
        //factory.setMessageConverter(jsonMessageConverter);
        return factory;
    }

（2）手动确认

RabbitMQ的消息确认机制

未确认的消息数

上图为channel中未被消费者确认的消息数。

通过RabbitMQ的host地址加上默认端口号15672访问管理界面。

成功确认

void basicAck(long deliveryTag, boolean multiple) throws IOException;

deliveryTag：该消息的index
multiple：是否批量. true：将一次性ack所有小于deliveryTag的消息。

消费者成功处理后，调用channel.basicAck(message.getMessageProperties().getDeliveryTag(), false)方法对消息进行确认。

示例：

@Component
@RabbitListener(queues = "confirm_queue_B")
public class Customer {
    @RabbitHandler
    public void process(Message message, Channel channel){
        System.out.println("ReceiverA："+new String(message.getBody()));
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);
}

失败确认

失败确认一：

void basicNack(long deliveryTag, boolean multiple, boolean requeue) throws IOException;

deliveryTag：该消息的index。
multiple：是否批量. true：将一次性拒绝所有小于deliveryTag的消息。
requeue：被拒绝的是否重新入队列。

示例

@Component
@RabbitListener(queues = "confirm_queue_B")
public class Customer {
    @RabbitHandler
    public void processJsonMessage(@Payload String body, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Message message,Channel channel){      
        System.out.println("ReceiverA："+new String(message.getBody()));
        channel.basicNack(deliveryTag,true,true);
}

失败确认二：

void basicReject(long deliveryTag, boolean requeue) throws IOException;

deliveryTag：该消息的index。
requeue：被拒绝的是否重新入队列。

channel.basicNack 与 channel.basicReject 的区别在于basicNack可以批量拒绝多条消息，而basicReject一次只能拒绝一条消息。

@Component
@RabbitListener(queues = "confirm_queue_B")
public class Customer {
     public void processJsonMessage(@Payload String body, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Message message,Channel channel){  
        System.out.println("ReceiverA："+new String(message.getBody()));
        channel.basicReject(deliveryTag,true);
}

思考

（1）手动确认模式，消息手动拒绝中如果requeue为true会重新放入队列，但是如果消费者在处理过程中一直抛出异常，会导致入队-》拒绝-》入队的循环，该怎么处理呢？

第一种方法是根据异常类型来选择是否重新放入队列。

第二种方法是先成功确认，然后通过**channel.basicPublish()**重新发布这个消息。重新发布的消息网上说会放到队列后面，进而不会影响已经进入队列的消息处理。

void basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, BasicProperties props, byte[] body) throws IOException;

（2）消息确认的作用是什么？

为了防止消息丢失。消息丢失分为发送丢失和消费者处理丢失，相应的也有两种确认机制。

这里我写了一个处理的模板

package com.lay.rabbitmqtwo.customer;

/**
 * @Description:
 * @Author: lay
 * @Date: Created in 13:19 2018/12/25
 * @Modified By:IntelliJ IDEA
 */
@Component
@RabbitListener(queues = "confirm_queue_B")
public class AckTempalte {
    enum Action{
        ACCEPT, // 处理成功
        RETRY, // 可以重试的错误
        REJECT, // 无需重试的错误
    }
    @RabbitHandler
    public void processJsonUser(Message message, Channel channel){
        Action action=Action.ACCEPT;
        long tag=message.getMessageProperties().getDeliveryTag();
        try{

            message.getMessageProperties().getConsumerTag();
            System.out.println( message.getMessageProperties().getConsumerTag());
            String message1 = new String(message.getBody(), "UTF-8");
            System.out.println("获取消息'" + message1 + "'");

        }catch (Exception e){
            // 根据异常种类决定是ACCEPT、RETRY还是 REJECT
            action = Action.RETRY;
            e.printStackTrace();

        }finally {
            try {
                // 通过finally块来保证Ack/Nack会且只会执行一次
                if (action == Action.ACCEPT) {
                    channel.basicAck(tag, true);
                    // 重试
                } else if (action == Action.RETRY) {
                    channel.basicNack(tag, false, true);
                    Thread.sleep(2000L);
                    // 拒绝消息也相当于主动删除mq队列的消息
                } else {
                    channel.basicNack(tag, false, false);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

github源码

附录：Channel接口参数

channel.exchangeDeclare()：

type：direct、fanout、topic三种
durable：true、false true：服务器重启会保留下来Exchange。警告：仅设置此选项，不代表消息持久化。即不保证重启后消息还在。原文：true if we are declaring a durable exchange (the exchange will survive a server restart)
autoDelete：true、false.true:当已经没有消费者时，服务器是否可以删除该Exchange。原文1：true if the server should delete the exchange when it is no longer in use。

/**
     * Declare an exchange.
     * @see com.rabbitmq.client.AMQP.Exchange.Declare
     * @see com.rabbitmq.client.AMQP.Exchange.DeclareOk
     * @param exchange the name of the exchange
     * @param type the exchange type
     * @param durable true if we are declaring a durable exchange (the exchange will survive a server restart)
     * @param autoDelete true if the server should delete the exchange when it is no longer in use
     * @param arguments other properties (construction arguments) for the exchange
     * @return a declaration-confirm method to indicate the exchange was successfully declared
     * @throws java.io.IOException if an error is encountered
     */
    Exchange.DeclareOk exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete,Map<String, Object> arguments) throws IOException;

chanel.basicQos()

prefetchSize：0
prefetchCount：会告诉RabbitMQ不要同时给一个消费者推送多于N个消息，即一旦有N个消息还没有ack，则该consumer将block掉，直到有消息ack
global：true\false 是否将上面设置应用于channel，简单点说，就是上面限制是channel级别的还是consumer级别
备注：据说prefetchSize 和global这两项，rabbitmq没有实现，暂且不研究

/**
 * Request specific "quality of service" settings.
 *
 * These settings impose limits on the amount of data the server
 * will deliver to consumers before requiring acknowledgements.
 * Thus they provide a means of consumer-initiated flow control.
 * @see com.rabbitmq.client.AMQP.Basic.Qos
 * @param prefetchSize maximum amount of content (measured in
 * octets) that the server will deliver, 0 if unlimited
 * @param prefetchCount maximum number of messages that the server
 * will deliver, 0 if unlimited
 * @param global true if the settings should be applied to the
 * entire channel rather than each consumer
 * @throws java.io.IOException if an error is encountered
 */
void basicQos(int prefetchSize, int prefetchCount, boolean global) throws IOException;

channel.basicPublish()

routingKey：路由键，#匹配0个或多个单词，*匹配一个单词，在topic exchange做消息转发用
mandatory：true：如果exchange根据自身类型和消息routeKey无法找到一个符合条件的queue，那么会调用basic.return方法将消息返还给生产者。false：出现上述情形broker会直接将消息扔掉
immediate：true：如果exchange在将消息route到queue(s)时发现对应的queue上没有消费者，那么这条消息不会放入队列中。当与消息routeKey关联的所有queue(一个或多个)都没有消费者时，该消息会通过basic.return方法返还给生产者。
BasicProperties ：需要注意的是BasicProperties.deliveryMode，0:不持久化 1：持久化这里指的是消息的持久化，配合channel(durable=true),queue(durable)可以实现，即使服务器宕机，消息仍然保留
简单来说：mandatory标志告诉服务器至少将该消息route到一个队列中，否则将消息返还给生产者；
immediate：标志告诉服务器如果该消息关联的queue上有消费者，则马上将消息投递给它，如果所有queue都没有消费者，直接把消息返还给生产者，不用将消息入队列等待消费者了。

/**
* Publish a message.
*
* Publishing to a non-existent exchange will result in a channel-level
* protocol exception, which closes the channel.
*
* Invocations of Channel#basicPublish will eventually block if a
* resource-driven alarm is in effect.
*
* @see com.rabbitmq.client.AMQP.Basic.Publish
* @see Resource-driven alarms.
* @param exchange the exchange to publish the message to
* @param routingKey the routing key
* @param mandatory true if the ‘mandatory’ flag is to be set
* @param immediate true if the ‘immediate’ flag is to be
* set. Note that the RabbitMQ server does not support this flag.
* @param props other properties for the message - routing headers etc
* @param body the message body
* @throws java.io.IOException if an error is encountered
*/
void basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, BasicProperties props, byte[] body) throws IOException;


### channel.basicAck();

- deliveryTag：该消息的index
- multiple：是否批量.true:将一次性ack所有小于deliveryTag的消息。



```java
/**
* Acknowledge one or several received
* messages. Supply the deliveryTag from the {@link com.rabbitmq.client.AMQP.Basic.GetOk}
* or {@link com.rabbitmq.client.AMQP.Basic.Deliver} method
* containing the received message being acknowledged.
* @see com.rabbitmq.client.AMQP.Basic.Ack
* @param deliveryTag the tag from the received {@link com.rabbitmq.client.AMQP.Basic.GetOk} or {@link com.rabbitmq.client.AMQP.Basic.Deliver}
* @param multiple true to acknowledge all messages up to and
* including the supplied delivery tag; false to acknowledge just
* the supplied delivery tag.
* @throws java.io.IOException if an error is encountered
*/
void basicAck(long deliveryTag, boolean multiple) throws IOException;

channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true)

deliveryTag：该消息的index
multiple：是否批量.true:将一次性拒绝所有小于deliveryTag的消息。
requeue：被拒绝的是否重新入队列

/**
 * Reject one or several received messages.
 *
 * Supply the <code>deliveryTag</code> from the {@link com.rabbitmq.client.AMQP.Basic.GetOk}
 * or {@link com.rabbitmq.client.AMQP.Basic.GetOk} method containing the message to be rejected.
 * @see com.rabbitmq.client.AMQP.Basic.Nack
 * @param deliveryTag the tag from the received {@link com.rabbitmq.client.AMQP.Basic.GetOk} or {@link com.rabbitmq.client.AMQP.Basic.Deliver}
 * @param multiple true to reject all messages up to and including
 * the supplied delivery tag; false to reject just the supplied
 * delivery tag.
 * @param requeue true if the rejected message(s) should be requeued rather
 * than discarded/dead-lettered
 * @throws java.io.IOException if an error is encountered
 */
void basicNack(long deliveryTag, boolean multiple, boolean requeue)
        throws IOException;

channel.basicReject(delivery.getEnvelope().getDeliveryTag(), false);

deliveryTag：该消息的index
requeue：被拒绝的是否重新入队列

channel.basicNack 与 channel.basicReject 的区别在于basicNack可以拒绝多条消息，而basicReject一次只能拒绝一条消息

channel.basicConsume(QUEUE_NAME, true, consumer);

autoAck：是否自动ack，如果不自动ack，需要使用channel.ack、channel.nack、channel.basicReject 进行消息应答

/**
 * Start a non-nolocal, non-exclusive consumer, with
 * a server-generated consumerTag.
 * @param queue the name of the queue
 * @param autoAck true if the server should consider messages
 * acknowledged once delivered; false if the server should expect
 * explicit acknowledgements
 * @param callback an interface to the consumer object
 * @return the consumerTag generated by the server
 * @throws java.io.IOException if an error is encountered
 * @see com.rabbitmq.client.AMQP.Basic.Consume
 * @see com.rabbitmq.client.AMQP.Basic.ConsumeOk
 * @see #basicConsume(String, boolean, String, boolean, boolean, Map, Consumer)
 */
String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;

chanel.exchangeBind()

channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
用于通过绑定bindingKey将queue到Exchange，之后便可以进行消息接收

/**
 * Bind an exchange to an exchange, with no extra arguments.
 * @see com.rabbitmq.client.AMQP.Exchange.Bind
 * @see com.rabbitmq.client.AMQP.Exchange.BindOk
 * @param destination the name of the exchange to which messages flow across the binding
 * @param source the name of the exchange from which messages flow across the binding
 * @param routingKey the routine key to use for the binding
 * @return a binding-confirm method if the binding was successfully created
 * @throws java.io.IOException if an error is encountered
 */
Exchange.BindOk exchangeBind(String destination, String source, String routingKey) throws IOException;

channel.queueDeclare(QUEUE_NAME, false, false, false, null);

durable：true、false true：在服务器重启时，能够存活
exclusive ：是否为当前连接的专用队列，在连接断开后，会自动删除该队列，生产环境中应该很少用到吧。
autodelete：当没有任何消费者使用时，自动删除该队列。this means that the queue will be deleted when there are no more processes consuming messages from it.

 /**
     * Declare a queue
     * @see com.rabbitmq.client.AMQP.Queue.Declare
     * @see com.rabbitmq.client.AMQP.Queue.DeclareOk
     * @param queue the name of the queue
     * @param durable true if we are declaring a durable queue (the queue will survive a server restart)
     * @param exclusive true if we are declaring an exclusive queue (restricted to this connection)
     * @param autoDelete true if we are declaring an autodelete queue (server will delete it when no longer in use)
     * @param arguments other properties (construction arguments) for the queue
     * @return a declaration-confirm method to indicate the queue was successfully declared
     * @throws java.io.IOException if an error is encountered
     */
    Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,Map<String, Object> arguments) throws IOException;

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)