SpringBoot2.x整合RabbitMQ实现回调确认、备份队列、死信

柔情只为你懂 2023-06-13 08:56 10阅读 0赞

SpringBoot整合RabbitMQ官方地址：[https://docs.spring.io/spring-boot/docs/current/reference/html/spring-boot-features.html\#boot-features-amqp][https_docs.spring.io_spring-boot_docs_current_reference_html_spring-boot-features.html_boot-features-amqp]

# 一、添加依赖 #

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

# 二、在application.yml中配置 #

spring:
      application:
        name: springboot-rabbitmq
      rabbitmq:
        host: 192.168.184.138
        port: 5672
        username: admin
        password: 123456
        virtual-host: /
        publisher-confirm-type: correlated #必须配置这个才会确认回调
        publisher-returns: true
        listener:
          type: simple
          simple:
            acknowledge-mode: manual #手动确认
            prefetch: 1 #限制每次发送一条数据。
            concurrency: 3 #同一个队列启动几个消费者
            max-concurrency: 3 #启动消费者最大数量
            #重试策略相关配置
            retry:
              enabled: true #是否支持重试
              max-attempts: 5
              stateless: false
              #时间策略乘数因子
              multiplier: 1.0
              initial-interval: 1000ms
              max-interval: 10000ms
            default-requeue-rejected: true

完整配置查看：[https://docs.spring.io/spring-boot/docs/2.1.3.RELEASE/reference/html/common-application-properties.html][https_docs.spring.io_spring-boot_docs_2.1.3.RELEASE_reference_html_common-application-properties.html]

# 三、生产者配置类 #

package com.rabbitmq.config;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.core.*;
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.beans.factory.config.ConfigurableBeanFactory;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.context.annotation.Scope;
    
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * @Package: com.rabbitmq.config
     * @ClassName: :RabbitMQConfig
     * @Description: rabbitmq
     * @Author: Coding-Farmer
     * @Date: 2019/11/15 14:19
     */
    @Configuration
    public class RabbitMQConfig {
        private static Logger log = LoggerFactory.getLogger(RabbitMQConfig.class);
        @Autowired
        private CachingConnectionFactory connectionFactory;
    
        //目标转换器，需要哪种类型的转换器就创建哪种类型的转换器
        @Bean
        public DirectExchange exchangeHello(){
            Map<String, Object> eArguments = new HashMap<>();
            //备份交换器参数
            eArguments.put("alternate-exchange", "exchange.ae");
            return new DirectExchange("exchange.hello",true,false,eArguments);
        }
        //备份转换器
        @Bean
        public FanoutExchange exchangeAE(){
            return new FanoutExchange("exchange.ae",true,false,null);
        }
        //死信转换器
        @Bean
        public TopicExchange exchangeDLX(){
            return new TopicExchange("exchange.dlx",true,false,null);
        }
    
        //目标对列
        @Bean
        public Queue queueHello() {
            Map<String, Object> args = new HashMap<>();
            //声明死信交换器
            args.put("x-dead-letter-exchange", "exchange.dlx");
            //声明死信路由键
            args.put("x-dead-letter-routing-key", "dlx.test" );
            //声明队列消息过期时间 5000ms
            args.put("x-message-ttl", 5000);
            return new Queue("queue.hello",true,false,false,args);
        }
    
        //备份对列
        @Bean
        public Queue queueAE() {
            return new Queue("queue.ae",true,false,false,null);
        }
    
        //死信对列
        @Bean
        public Queue queueDLX() {
            return new Queue("queue.dlx",true,false,false,null);
        }
    
        //绑定目标对列
        @Bean
        public Binding bindingExchangeDirect(@Qualifier("queueHello")Queue queueHello, @Qualifier("exchangeHello") DirectExchange exchangeHello){
            return  BindingBuilder.bind(queueHello).to(exchangeHello).with("helloKey");
        }
        //绑定备份对列
        @Bean
        public Binding bindingExchangeAE(@Qualifier("queueAE")Queue queueAE, @Qualifier("exchangeAE") FanoutExchange exchangeAE){
            return  BindingBuilder.bind(queueAE).to(exchangeAE);
        }
    
        //绑定死信对列
        @Bean
        public Binding bindingExchangeDLX(@Qualifier("queueDLX")Queue queueAE, @Qualifier("exchangeDLX") TopicExchange exchangeDLX){
            return  BindingBuilder.bind(queueAE).to(exchangeDLX).with("dlx.*");
        }
    
        /**
         * 如果需要在生产者需要消息发送后的回调，
         * 需要对rabbitTemplate设置ConfirmCallback对象，
         * 由于不同的生产者需要对应不同的ConfirmCallback，
         * 如果rabbitTemplate设置为单例bean，
         * 则所有的rabbitTemplate实际的ConfirmCallback为最后一次申明的ConfirmCallback。
         * @return
         */
        @Bean
        @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
        public RabbitTemplate rabbitTemplate() {
            RabbitTemplate template = new RabbitTemplate(connectionFactory);
            return template;
        }
    
    }

# 四、发送消息类 #

package com.rabbitmq.producer;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    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.Controller;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.ResponseBody;
    
    import java.util.Date;
    import java.util.UUID;
    
    @Controller
    public class HelloSender implements  RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback{
        private static Logger log = LoggerFactory.getLogger(HelloSender.class);
        private RabbitTemplate rabbitTemplate;
    
       
        //构造方法注入     
        @Autowired
        public HelloSender(RabbitTemplate rabbitTemplate) {
            this.rabbitTemplate = rabbitTemplate;
            //这是是设置回调能收到发送到响应
            rabbitTemplate.setConfirmCallback(this);
            //如果设置备份队列则不起作用
            rabbitTemplate.setMandatory(true);
            rabbitTemplate.setReturnCallback(this);
        }
    
        @RequestMapping("/send")
        @ResponseBody
        public void send() {
            CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
            String sendMsg = "hello1 " + new Date();
            System.out.println("Sender : " + sendMsg);
            //convertAndSend(exchange:交换机名称,routingKey:路由关键字,object:发送的消息内容,correlationData:消息ID)
            rabbitTemplate.convertAndSend("exchange.hello","helloKey", sendMsg,correlationId);
        }
    
    
        //回调确认
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            if(ack){
                log.info("消息发送成功:correlationData({}),ack({}),cause({})",correlationData,ack,cause);
            }else{
                log.info("消息发送失败:correlationData({}),ack({}),cause({})",correlationData,ack,cause);
            }
        }
    
        //消息发送到转换器的时候没有对列,配置了备份对列该回调则不生效
        @Override
        public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
            log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}",exchange,routingKey,replyCode,replyText,message);
        }
    
    
    
    }

# 五、接收消息 #

package com.rabbitmq.consumer;
    
    import com.rabbitmq.client.Channel;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    import java.io.IOException;
    
    
    @Component
    public class HelloConsumer {
        private static Logger log = LoggerFactory.getLogger(HelloConsumer.class);
    
        @RabbitHandler
        @RabbitListener(queues = "queue.hello")
        public void process(Message message, Channel channel) throws IOException {
            log.info("receive: " + new String(message.getBody())+"《线程名：》"+Thread.currentThread().getName()+"《线程id:》"+Thread.currentThread().getId());
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
        }
    
    
    }

[https_docs.spring.io_spring-boot_docs_current_reference_html_spring-boot-features.html_boot-features-amqp]: https://docs.spring.io/spring-boot/docs/current/reference/html/spring-boot-features.html#boot-features-amqp
[https_docs.spring.io_spring-boot_docs_2.1.3.RELEASE_reference_html_common-application-properties.html]: https://docs.spring.io/spring-boot/docs/2.1.3.RELEASE/reference/html/common-application-properties.html