springBoot集成rabbitmq 之延时（死信）队列

Bertha 。 2024-03-27 16:49 36阅读 0赞

### springBoot集成rabbitmq 之延时（死信）队列 ###

#### comsumer 服务 ####

配置类

import org.springframework.amqp.core.*;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import java.util.HashMap;
    import java.util.Map;
    
    @Configuration
    public class RabbitDeadQueueConfig {
        
    
        /**
            * Queue 可以有4个参数
         *      1.队列名
         *      2.durable       持久化消息队列 ,rabbitmq重启的时候不需要创建新的队列 默认true
         *      3.auto-delete   表示消息队列没有在使用时将被自动删除 默认是false
         *      4.exclusive     表示该消息队列是否只在当前connection生效,默认是false
         */
    
        /**
         * 死信队列跟交换机类型没有关系 不一定为directExchange  不影响该类型交换机的特性.
         *
         * @return the exchange
         */
        @Bean
        public Exchange deadLetterExchange() {
        
            return ExchangeBuilder.directExchange("EXCHANGE_DL").durable(true).build();
        }
    
        /**
         * 声明一个死信队列.
         * x-dead-letter-exchange   对应  死信交换机
         * x-dead-letter-routing-key  对应 死信队列
         *
         * @return the queue
         */
        @Bean
        public Queue deadLetterQueue() {
        
            Map<String, Object> args = new HashMap<>(2);
    //       x-dead-letter-exchange    声明  死信交换机
            args.put("x-dead-letter-exchange", "EXCHANGE_DL");
    //       x-dead-letter-routing-key    声明 死信路由键
            args.put("x-dead-letter-routing-key","KEY_R");
            return QueueBuilder.durable("QUEUE_DL").withArguments(args).build();
        }
    
        /**
         * 定义死信队列转发队列.
         *
         * @return the queue
         */
        @Bean
        public Queue redirectQueue() {
        
            return QueueBuilder.durable("QUEUE_REDIRECT").build();
        }
    
        /**
         * 死信路由通过 DL_KEY 绑定键绑定到死信队列上.
         * @return the binding
         */
        @Bean
        public Binding deadLetterBinding() {
        
            return new Binding("QUEUE_DL", Binding.DestinationType.QUEUE,
                    "EXCHANGE_DL", "KEY_DL", null);
    
        }
    
        /**
         * 死信路由通过 KEY_R 绑定键绑定到最终处理队列上.
         * @return the binding
         */
        @Bean
        public Binding redirectBinding() {
        
            return new Binding("QUEUE_REDIRECT", Binding.DestinationType.QUEUE,
                    "EXCHANGE_DL", "KEY_R", null);
        }
    
    }

接收者

import com.rabbitmq.client.Channel;
    import org.springfram
    import org.springframework.amqp.core.*;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import java.util.HashMap;
    import java.util.Map;
    
    @Configuration
    public class RabbitDeadQueueConfig {
        
    
        /**
            * Queue 可以有4个参数
         *      1.队列名
         *      2.durable       持久化消息队列 ,rabbitmq重启的时候不需要创建新的队列 默认true
         *      3.auto-delete   表示消息队列没有在使用时将被自动删除 默认是false
         *      4.exclusive     表示该消息队列是否只在当前connection生效,默认是false
         */
    
        /**
         * 死信队列跟交换机类型没有关系 不一定为directExchange  不影响该类型交换机的特性.
         *
         * @return the exchange
         */
        @Bean
        public Exchange deadLetterExchange() {
        
            return ExchangeBuilder.directExchange("EXCHANGE_DL").durable(true).build();
        }
    
        /**
         * 声明一个死信队列.
         * x-dead-letter-exchange   对应  死信交换机
         * x-dead-letter-routing-key  对应 死信队列
         *
         * @return the queue
         */
        @Bean
        public Queue deadLetterQueue() {
        
            Map<String, Object> args = new HashMap<>(2);
    //       x-dead-letter-exchange    声明  死信交换机
            args.put("x-dead-letter-exchange", "EXCHANGE_DL");
    //       x-dead-letter-routing-key    声明 死信路由键
            args.put("x-dead-letter-routing-key","KEY_R");
            return QueueBuilder.durable("QUEUE_DL").withArguments(args).build();
        }
    
        /**
         * 定义死信队列转发队列.
         *
         * @return the queue
         */
        @Bean
        public Queue redirectQueue() {
        
            return QueueBuilder.durable("QUEUE_REDIRECT").build();
        }
    
        /**
         * 死信路由通过 DL_KEY 绑定键绑定到死信队列上.
         * @return the binding
         */
        @Bean
        public Binding deadLetterBinding() {
        
            return new Binding("QUEUE_DL", Binding.DestinationType.QUEUE,
                    "EXCHANGE_DL", "KEY_DL", null); 
        }
    
        /**
         * 死信路由通过 KEY_R 绑定键绑定到最终处理队列上.
         * @return the binding
         */
        @Bean
        public Binding redirectBinding() {
        
            return new Binding("QUEUE_REDIRECT", Binding.DestinationType.QUEUE,
                    "EXCHANGE_DL", "KEY_R", null);
        }
    
    }ework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    import java.util.Date; 
    
    @Component
    @RabbitListener(queues = "QUEUE_REDIRECT")
    public class DeadQueueReceiver {
         
        @RabbitHandler
        public void process(String hello, Message message, Channel channel) {
         
            try {
        
                //告诉服务器收到这条消息 已经被我消费了 可以在队列删掉 这样以后就不会再发了 否则消息服务器以为这条消息没处理掉 后续还会在发
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                System.out.println("消息消费成功！");
            } catch (Exception e) {
        
                System.out.println("消息消费失败:"+e.getMessage() + e);
                //丢弃这条消息
                //channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,false);
            }
            System.out.println("接收者 : " + hello +","+ new Date());
        }
    }

#### producer服务 ####

发送者

import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessagePostProcessor;
    import org.springframework.amqp.core.MessageProperties;
    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 java.util.Date;
    import java.util.UUID;
    
    @Component
    public class DeadQueueSender implements  RabbitTemplate.ConfirmCallback ,RabbitTemplate.ReturnCallback {
        
    
    
    	@Autowired
    	private RabbitTemplate rabbitTemplate;
    
    	public void send() {
        
    		//设置回调对象
    		this.rabbitTemplate.setConfirmCallback(this);
    		this.rabbitTemplate.setReturnCallback(this);
    		CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    		String content = "死信队列 = " + new Date() + ", 内容= " + UUID.randomUUID().toString();
    		MessagePostProcessor messagePostProcessor = message -> {
        
    			MessageProperties messageProperties = message.getMessageProperties();
    //            设置编码
    			messageProperties.setContentEncoding("utf-8");
    //            设置过期时间10*1000毫秒
    			messageProperties.setExpiration("10000");
    			return message;
    		};
    //         向DL_QUEUE 发送消息  10*1000毫秒后过期 形成死信
    		rabbitTemplate.convertAndSend("EXCHANGE_DL", "KEY_DL", content, messagePostProcessor, correlationData);
    		System.out.println("发送成功,"+new Date()+","+content);
    	}
    
    	/**
    	 * 消息回调确认方法
    	 * 如果消息没有到exchange,则confirm回调,ack=false
    	 * 如果消息到达exchange,则confirm回调,ack=true
    	 * @param
    	 */
    	@Override
    	public void confirm(CorrelationData correlationData, boolean isSendSuccess, String s) {
        
    		System.out.println("confirm--message:回调消息ID为: " + correlationData.getId());
    		if (isSendSuccess) {
        
    			System.out.println("确认--消息:消息发送成功");
    		} else {
        
    			System.out.println("确认--消息:消息发送失败" + s);
    		}
    	}
    
    	/**
    	 * exchange到queue成功,则不回调return
    	 * exchange到queue失败,则回调return(需设置mandatory=true,否则不回回调,消息就丢了)
    	 */
    	@Override
    	public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        
    		System.out.println("返回--消息:" + new String(message.getBody()) + ",replyCode:" + replyCode
    				+ ",replyText:" + replyText + ",exchange:" + exchange + ",routingKey:" + routingKey);
    	}
    }

测试

import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RestController; 
    @RestController
    public class DeadQueueTests {
        
        @Autowired
        private DeadQueueSender deadQueueSender; 
        @GetMapping("DeadQueueTests")
        public void hello() {
        
            deadQueueSender.send();
        }
    }

通过执行测试类，查看到了消息消费的情况，生产者共计生产了 1 个消息，被消费者消费了一次，但发送消息时间及实际消费时间差 10 秒钟。