RabbitMq（四）Springboot 整合RabbitMq应用——消息回调、手动确认

灰太狼 2023-07-24 14:58 1阅读 0赞

## 消息回调 ##

消息的回调，其实就是消息确认（生产者推送消息成功，消费这接收消息成功）。  
在上篇文章代码中application.yml加入消息确认  
rabbitmq\_prov\_demo

server:
      port: 8021
    string:
      application:
        name: rabbitmq_prov_demo
      rabbitmq:
       host: localhost
       port: 5672
       username: admin
       password: 123456
       #虚拟host 可以不设置,使用server默认host
       #virtual-host: IManiyHost
       #消息确认配置项
    
       #确认消息已发送到交换机(Exchange)
       publisher-confirms: true
       #确认消息已发送到队列(Queue)
       publisher-returns: true

配置相关的消息确认回调函数

import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.rabbit.support.CorrelationData;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /**
     * @author liuzonghua
     * @Package com.example.rabbitmq_prov_demo.confirm
     * @Description:
     * @date 2020/4/12 21:15
     */
    @Configuration
    public class RabbitConfig {
    
        @Bean
        public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){
            RabbitTemplate rabbitTemplate = new RabbitTemplate();
            rabbitTemplate.setConnectionFactory(connectionFactory);
            //设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数
            rabbitTemplate.setMandatory(true);
    
            rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
                @Override
                public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                    System.out.println("ConfirmCallback:     "+"相关数据："+correlationData);
                    System.out.println("ConfirmCallback:     "+"确认情况："+ack);
                    System.out.println("ConfirmCallback:     "+"原因："+cause);
                }
            });
    
            rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
                @Override
                public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                    System.out.println("ReturnCallback:     "+"消息："+message);
                    System.out.println("ReturnCallback:     "+"回应码："+replyCode);
                    System.out.println("ReturnCallback:     "+"回应信息："+replyText);
                    System.out.println("ReturnCallback:     "+"交换机："+exchange);
                    System.out.println("ReturnCallback:     "+"路由键："+routingKey);
                }
            });
    
            return rabbitTemplate;
        }
    
    }

生产者推送消息的消息确认调用回调函数已经完毕。  
可以看到上面写了两个回调函数，一个叫 ConfirmCallback ，一个叫 RetrunCallback；  
那么以上这两种回调函数都是在什么情况会触发呢？

先从总体的情况分析，推送消息存在四种情况：

①消息推送到server，但是在server里找不到交换机,①这种情况触发的是 ConfirmCallback 回调函数。  
②消息推送到server，找到交换机了，但是没找到队列,②这种情况触发的是 ConfirmCallback和RetrunCallback两个回调函数。  
③消息推送到sever，交换机和队列啥都没找到 ,③这种情况触发的是 ConfirmCallback 回调函数。  
④消息推送成功, ④这种情况触发的是 ConfirmCallback 回调函数

## 手动确认 ##

container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默认是自动确认，这里改为手动确认消息

import com.elegant.rabbitmqconsumer.receiver.DirectReceiver;
    import com.elegant.rabbitmqconsumer.receiver.FanoutReceiverA;
    import org.springframework.amqp.core.AcknowledgeMode;
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class MessageListenerConfig {
     
        @Autowired
        private CachingConnectionFactory connectionFactory;
        @Autowired
        private DirectReceiver directReceiver;//Direct消息接收处理类
    //    @Autowired
    //    FanoutReceiverA fanoutReceiverA;//Fanout消息接收处理类A
        @Autowired
        DirectRabbitConfig directRabbitConfig;
    //    @Autowired
    //    FanoutRabbitConfig fanoutRabbitConfig;
        @Bean
        public SimpleMessageListenerContainer simpleMessageListenerContainer() {
            SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
            container.setConcurrentConsumers(1);
            container.setMaxConcurrentConsumers(1);
            container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默认是自动确认，这里改为手动确认消息
            container.setQueues(directRabbitConfig.TestDirectQueue());  //这里如果在这边设置了监听的队列 和下面设置了消费的监听类，那么对应的监听类可以去掉监听类的注解 ，谢谢有朋友指出这一点
            container.setMessageListener(directReceiver);
    //        container.addQueues(fanoutRabbitConfig.queueA());
    //        container.setMessageListener(fanoutReceiverA);
            return container;
        }
    }

#### rabbitmq\_cons\_demo ####

手动模式消费者要加上  
channel.basicAck(deliveryTag, true);  
或  
channel.basicReject(deliveryTag, false);