rabbitmq消息确认机制

痛定思痛。 2022-09-05 04:04 238阅读 0赞

rabbitmq消费确认机制分三种：生产者消息确认机制、消费者消息确认机制、return消息机制

# 一、生产者消息确认机制 #

生产者消息确认机制：指生产者将消息投递给对应的Broker中的VHost里面的exchange后产生的应答，如果exchange不存在返回false，投递成功则返回true

## 如何使用使用生产者确认机制 ##

1、配置

spring:
      rabbitmq:
        host: 192.168.178.100
        port: 5672
        vhost: xxx
        username: xxx
        password: xxxx
        rawdataQueue: EV-FORWARD-RAWDATA-ONE-QUEUE
        publisher-confirms: true #ture表示使用生产者确认模式

2、代码

@Component
    public class RabbitMqProducer implements RabbitTemplate.ConfirmCallback { 
        private static final Logger logger= LoggerFactory.getLogger(RabbitMqProducer.class);
    
        private RabbitTemplate rabbitTemplate;
    
        @Autowired
        public void setRabbitTemplate(RabbitTemplate rabbitTemplate) { 
            this.rabbitTemplate = rabbitTemplate;
        }
    
        public void sendMessage(String exchangeKey,String routingkey,Object message){ 
            String id= UUID.randomUUID().toString();
            //执行rabbitmq消息发送后回调方法
            rabbitTemplate.setConfirmCallback(this);
            rabbitTemplate.convertAndSend(exchangeKey,routingkey,message,new CorrelationData(id));
            logger.info("交换机名称{}，routingkey名称{}，消息:{}，唯一ID{}",exchangeKey,routingkey,message,id);
        }
    
        @Override
        public void confirm(CorrelationData correlationData, boolean b, String s) { 
            logger.info("消息推送交换机状态{},消息ID{}",b,correlationData.getId());
        }
    }

3、加载配置

public ConnectionFactory connectionFactory() { 
            CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host,port);
            connectionFactory.setUsername(username);
            connectionFactory.setPassword(password);
            connectionFactory.setVirtualHost(virtual);
            //这里设置生产者确认机制为true
            connectionFactory.setPublisherConfirms(true);
            return connectionFactory;
        }

正确执行结果

交换机名称TBOX-TSP-REALTIME-FORMATDATA-EXCHANGE，routingkey名称Logon，消息:{ "batteryNoList":[],"common":{ "order":1,"timestamp":"2021-08-18 16:22:21","vin":"12345678901234567"},"iccid":"01234567899876543210","water":1}，唯一IDb0ac4394-1be3-44cd-9151-9fed51295ef4
    [2021-08-18 16:22:26.609] [INFO ] [Server-Worker[2]] [com.dayunmotor.tbox.gateway.netty.handler.WriteHandlerFirst.write] [49] [] [] :
                vin[12345678901234567]>>>WriteHandler发送数据给 TBOX:2323010131323334353637383930313233343536370100061508121016152a
    [2021-08-18 16:22:26.616] [INFO ] [AMQP Connection 192.168.178.100:5672] [com.dayunmotor.tbox.gateway.mq.RabbitMqProducer.confirm] [32] [] [] :
                消息推送交换机状态true,消息IDb0ac4394-1be3-44cd-9151-9fed51295ef4

将交换机改成错误的

[2021-08-18 16:38:04.860] [INFO ] [Server-Worker[2]] [com.dayunmotor.tbox.gateway.mq.RabbitMqProducer.sendMessage] [27] [] [] :
                交换机名称1111，routingkey名称Logon，消息:{ "common":{ "order":1,"timestamp":"2021-08-18 16:38:04","vin":"12345678901234567"},"data":[35,35,1,-2,49,50,51,52,53,54,55,56,57,48,49,50,51,52,53,54,55,1,0,30,21,8,18,16,38,4,0,1,48,49,50,51,52,53,54,55,56,57,57,56,55,54,53,52,51,50,49,48,0,0,-19]}，唯一ID16d12ec3-1b4e-4006-8412-b286b940115e
    [2021-08-18 16:38:04.864] [ERROR] [AMQP Connection 192.168.178.100:5672] [org.springframework.amqp.rabbit.connection.CachingConnectionFactory.log] [1517] [] [] :
                Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange '1111' in vhost 'dayunmotor', class-id=60, method-id=40)
    [2021-08-18 16:38:04.866] [INFO ] [connectionFactory1] [com.dayunmotor.tbox.gateway.mq.RabbitMqProducer.confirm] [32] [] [] :
                消息推送交换机状态false,消息ID16d12ec3-1b4e-4006-8412-b286b940115e

# 二、return消息确认机制 #

return消息机制：消息发送到Exchange后，没有找到绑定的队列，投递消息失败才执行return消息确认机制  
在CachingConnectionFactory 增加设置PublisherReturns

connectionFactory.setPublisherReturns(true);

添加RabbitTemplate.ReturnCallback

package com.dayunmotor.tsp.vehicle.mq;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.PostConstruct;
    import javax.annotation.Resource;
    
    /** * rabbitMq生产数据类 * * @Author liu **/
    @Slf4j
    @Component
    public class RabbitMqProducer implements RabbitTemplate.ReturnCallback { 
    
        private RabbitTemplate rabbitTemplate;
    
        public void sendMessage(String exchange, String routingkey, Object message) { 
            log.info("队列{} to send message:{}",routingkey, message);
            rabbitTemplate.setReturnCallback(this);
            rabbitTemplate.convertAndSend(exchange, routingkey, message);
        }
    
        @Autowired
        public void setRabbitTemplate(RabbitTemplate rabbitTemplate) { 
            this.rabbitTemplate = rabbitTemplate;
        }
    
        public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) { 
            log.info("交换机名称{},routingKey名称{},返回code{}和内容{},消息内容{}",exchange,routingKey,replyCode,replyText,new String(message.getBody()));
        }
    }

# 消费者消息确认机制 #

消费端是默认自动确认的，只要消息接收到，客户端就会自动确认消息，RabbitMQ就会移除这个消息，同时也可以设置为手动确认消息

开启手动确认模式后，每次收到消息都需要主动返回一个Basic.Ack；  
消息确认模式分为三种：  
1、Ack  
2、Reject  
3、Nack  
channel.basicAck：当正常消费消息时，调用该方法。  
我们看到除了basicAck，还有basicReject和basicNack。这两种，顾名思义，是用来拒绝消费的。  
channel.basicReject：从协议层面上，reject是发送一个Basic.Reject响应，告知RabbitMQ无法对这条消息进行处理，当拒绝时，可以指定是否丢弃消息或使用requeue标志重新发送消息。当启用requeue时，RabbitMQ将会把这条消息重新放回到队列中。  
不能使用basicReject一次拒绝多个消息。  
channel.basicNack：Basic.Nack实现与Basic.Reject相同的行为，但添加了批量拒绝的功能。

@Component
    @Slf4j
    @RabbitListener(queues = "DIRECT_QUEUE")
    public class DirectQueueListener { 
    
        @RabbitHandler
        public void process(String message,
                            Channel channel,
                            @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException, InterruptedException { 
            log.info("消费消息成功: {}", message);
            Thread.sleep(1000);
            switch (message) { 
                case "nack":
                    channel.basicNack(tag, true, false); // 第二个参数控制是否开启批量拒绝，第三个参数表示是否requeue
                    break;
                case "nack-requeue":
                    channel.basicNack(tag, true, true);
                    break;
                case "reject":
                    channel.basicReject(tag, false);
                    break;
                case "reject-requeue": // 启用了requeue，如果只有一个消费者，容易造成死循环
                    channel.basicReject(tag, true);
                    break;
                default:
                    channel.basicAck(tag, true);
                    break;
            }
        }
    
    }