springboot整合rabbitmq（二）

末蓝、 2023-10-04 20:01 29阅读 0赞

上篇文章讲了rabbitmq的主要几种交换机以及交换机的简单使用。

[springboot整合rabbitmq（一）\_qianQueen的博客-CSDN博客][springboot_rabbitmq_qianQueen_-CSDN]

接下来我们讲rabbitmq消息的回调以及自动/手动确认消息相关的内容。

配置文件需要加上：

spring:
        application:
          rabbitmq-provider
        rabbitmq:
          host: 127.0.0.1
          port: 5672
          username: guest
          password: guest
          #确认消息已发送到交换机(Exchange)
          publisher-confirm-type: correlated
          #确认消息已发送到队列(Queue)
          publisher-returns: true

然后是相关的配置文件RabbitConfig（消息发送的回调）：

/**
     * @Author : liuqian
     * @CreateTime : 2022/11/22
     * @Description :
     * 配置相关的消息确认回调函数
     * https://blog.csdn.net/qq_35387940/article/details/100514134 借鉴网址
     *
     **/
    @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和returnCallback：

发送消息时，会先找交换机，再找对应的队列。当生产者发送消息时找到交换机后找不到队列则会回调returnCallback函数，如果找不到交换机那么也自然找不到队列，只会回调confirmCallback函数，但是不管什么情况都一定会回调confirmCallback函数。

接下来我们讲消费者的消息确认机制（根据消息的重要性选择确认机制）：

1.自动ack：自动确认好处是可以提高吞吐量，缺点是可能会丢失消息。

2.手动ack：消息一般不会轻易丢失，但会造成消费者过载。该手动确认又分为三种：

1. Ack: 表示消息已被正确处理消费  
2. Reject: 消息未被正确消费，一次只能处理一条消息，除此和nack相同。  
3. Nack: 消息未被正确消费，可以处理多条消息。但拒绝消费时，可以指定是否丢弃消息或使用requeue标志重新发送消息。当启用requeue时，RabbitMQ将会把这条消息重新放回到队列中，但要注意，如果使用不当则会导致消息一直消费-入列-再消费-再入列这样的情况，会导致消息积压，请谨慎使用。

接下来我们看使用，消息监听的配置文件：

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;
    
    /**
     * @Author : JCccc
     * @CreateTime : 2019/9/4
     * @Description :
     **/
    @Configuration
    public class MessageListenerConfig {
     
        @Autowired
        private CachingConnectionFactory connectionFactory;
    
        @Autowired
        private MyAckReceiver myAckReceiver;//消息接收处理类
     
        @Bean
        public SimpleMessageListenerContainer simpleMessageListenerContainer() {
            SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
            container.setConcurrentConsumers(1);
            container.setMaxConcurrentConsumers(1);
            container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默认是自动确认，这里改为手动确认消息 也可以直接在applicatin.yml文件中修改
            //设置一个队列
            //container.setQueueNames("TestDirectQueue");
            //如果同时设置多个如下： 前提是队列都是必须已经创建存在的
            container.setQueueNames("TestDirectQueue","fanout.test1","fanout.test2", "topic.school.student", "topic.school.teacher");
     
     
            //另一种设置队列的方法,如果使用这种情况,那么要设置多个,就使用addQueues
            //container.setQueues(new Queue("TestDirectQueue",true));
            //container.addQueues(new Queue("TestDirectQueue2",true));
            //container.addQueues(new Queue("TestDirectQueue3",true));
            container.setMessageListener(myAckReceiver);
            return container;
        }
     
    }

可以处理所有队列消息的处理类：

import com.rabbitmq.client.Channel;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
    import org.springframework.stereotype.Component;
    
    import java.io.ByteArrayInputStream;
    import java.io.ObjectInputStream;
    import java.util.Map;
    
    
    /**
     * 可以处理所有监听的队列消息
     */
    @Component
    public class MyAckReceiver implements ChannelAwareMessageListener {
    
        @Override
        public void onMessage(Message message, Channel channel) throws Exception {
            long deliveryTag = message.getMessageProperties().getDeliveryTag();
            try {
                byte[] body = message.getBody();
                ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(body));
                Map<String,String> msgMap = (Map<String,String>) ois.readObject();
                String messageId = msgMap.get("messageId");
                String messageData = msgMap.get("messageData");
                String createTime = msgMap.get("createTime");
                ois.close();
    
                if ("TestDirectQueue".equals(message.getMessageProperties().getConsumerQueue())){
                    System.out.println("消费的消息来自的队列名为："+message.getMessageProperties().getConsumerQueue());
                    System.out.println("消息成功消费到  messageId:"+messageId+"  messageData:"+messageData+"  createTime:"+createTime);
                    System.out.println("执行TestDirectQueue中的消息的业务处理流程......");
    
                }
    
                if ("fanout.A".equals(message.getMessageProperties().getConsumerQueue())){
                    System.out.println("消费的消息来自的队列名为："+message.getMessageProperties().getConsumerQueue());
                    System.out.println("消息成功消费到  messageId:"+messageId+"  messageData:"+messageData+"  createTime:"+createTime);
                    System.out.println("执行fanout.A中的消息的业务处理流程......");
    
                }
    
                channel.basicAck(deliveryTag, true);
    //			channel.basicReject(deliveryTag, true);//为true会重新放回队列
            } catch (Exception e) {
                channel.basicReject(deliveryTag, false);
                e.printStackTrace();
            }
        }

接下来我们测试：

![29a7d702935a4370a09cecb6151f5b0f.png][]

springboot整合rabbitmq就这就基本完成啦，下次见！

[springboot_rabbitmq_qianQueen_-CSDN]: https://blog.csdn.net/qq_36090537/article/details/127982817?
[29a7d702935a4370a09cecb6151f5b0f.png]: https://img-blog.csdnimg.cn/29a7d702935a4370a09cecb6151f5b0f.png