rabbitmq 延时队列和死信队列的分析

蔚落 2023-10-02 20:01 73阅读 0赞

一、生产者服务配置：

1、pom

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

2、服务器信息配置：

spring:
      rabbitmq:
        host: 94.191.83.120
        port: 5672
        username: admin
        password: admin
        publisher-confirms: true
        virtual-host: /

3、类实例化配置：

import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.*;
    import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
    import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.retry.backoff.ExponentialBackOffPolicy;
    import org.springframework.retry.policy.SimpleRetryPolicy;
    import org.springframework.retry.support.RetryTemplate;
    
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    
     
    @Slf4j
    @Configuration
    public class RabbitConfig {
    
        public final static String EXCHANGE_SECKILL = "order.seckill.delay.exchange";
        public final static String KEY_SECKILL = "order.seckill.delay.routingkey";
    
        public final static String EXCHANGE_SECKILL_DEAD = "order.seckill.dead.exchange";
        public final static String KEY_SECKILL_DEAD = "order.seckill.dead.routingkey";
    
        @Bean(name = "queueDelayMessage")
        public Queue queueDelayMessage() {
            Map<String,Object> arguments = new HashMap<>();
            arguments.put("x-message-ttl",1000 * 60 * 10);
    //        arguments.put("x-expires",1000 * 60);
    //        arguments.put("x-max-length",10000);
    //        arguments.put("x-max-length-bytes",50*1024);
            arguments.put("x-dead-letter-exchange",EXCHANGE_SECKILL_DEAD);
            arguments.put("x-dead-letter-routing-key",KEY_SECKILL_DEAD);
            return new Queue("order.seckill.delay.queue", true, false, false,arguments);
        }
    
        @Bean(name = "exchange")
        public DirectExchange exchange() {
            return new DirectExchange(EXCHANGE_SECKILL, true, false);
        }
    
        @Bean
        public Binding bindingSecKillExchangeMessage(@Qualifier("queueDelayMessage") Queue queueMessage,
                                                     @Qualifier("exchange") DirectExchange exchange) {
            return BindingBuilder
                    .bind(queueMessage)
                    .to(exchange)
                    .with(KEY_SECKILL);
        }
    
        @Bean(name = "deadqueueMessage")
        public Queue deadqueueMessage() {
            return new Queue("order.seckill.dead.queue",true);
        }
    
        @Bean(name = "deadexchange")
        public DirectExchange deadexchange() {
            return new DirectExchange(EXCHANGE_SECKILL_DEAD,true,false);
        }
    
        @Bean
        Binding bindingDeadExchangeMessage(@Qualifier("deadqueueMessage") Queue deadqueueMessage,
                                           @Qualifier("deadexchange") DirectExchange deadexchange) {
            return BindingBuilder.bind(deadqueueMessage).to(deadexchange).
                    with(KEY_SECKILL_DEAD);
        }
    
    
        @Bean
        public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory, List<SimpleMessageListenerContainer> list) {
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            factory.setConnectionFactory(connectionFactory);
            factory.setConcurrentConsumers(2);
    //抓取参数非常关键，一次抓取的消息多了，消费速度一慢，就会造成响应延迟，抓取少了又会导致并发量低，消息堵塞
            factory.setPrefetchCount(10);
    
            /*
             * AcknowledgeMode.NONE：自动确认
             * AcknowledgeMode.AUTO：根据情况确认
             * AcknowledgeMode.MANUAL：手动确认
             */
            factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
    /*        factory.setDefaultRequeueRejected(false);
            factory.setAdviceChain(
                    RetryInterceptorBuilder
                            .stateless()
                            .recoverer(new RejectAndDontRequeueRecoverer())
                            .retryOperations(retryTemplate())
                            .build()
            );*/
            return factory;
        }
    
        @Bean
        public RetryTemplate retryTemplate() {
            RetryTemplate retryTemplate = new RetryTemplate();
    /*        retryTemplate.registerListener(new RetryListener() {
                @Override
                public <T, E extends Throwable> boolean open(RetryContext retryContext, RetryCallback<T, E> retryCallback) {
                    return false;
                }
    
                @Override
                public <T, E extends Throwable> void close(RetryContext retryContext, RetryCallback<T, E> retryCallback, Throwable throwable) {
                }
    
                @Override
                public <T, E extends Throwable> void onError(RetryContext retryContext, RetryCallback<T, E> retryCallback, Throwable throwable) {
    
                }
            });*/
            retryTemplate.setBackOffPolicy(backOffPolicy());
            retryTemplate.setRetryPolicy(retryPolicy());
            return retryTemplate;
        }
    
        @Bean
        public ExponentialBackOffPolicy backOffPolicy() {
            ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
            backOffPolicy.setInitialInterval(1000);
            backOffPolicy.setMaxInterval(10000);
            return backOffPolicy;
        }
    
        @Bean
        public SimpleRetryPolicy retryPolicy() {
            SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
            retryPolicy.setMaxAttempts(3);
            return retryPolicy;
        }
    
    }

4、发送信息的工具类

import cn.nan.config.RabbitConfig;
    import cn.nan.mall.vo.KillOrderVo;
    import com.alibaba.fastjson.JSON;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.Message;
    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.UUID;
    
     
    @Slf4j
    @Component
    public class SecKillSender {
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        public void send(KillOrderVo vo) {
            String msg1 = JSON.toJSONString(vo);
            log.info("TopicSender send the 1st : " + msg1);
            this.rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_SECKILL, RabbitConfig.KEY_SECKILL, msg1);
        }
    
        public void send(KillOrderVo vo,CorrelationData correlationData) {
            String msg1 = JSON.toJSONString(vo);
            log.info("TopicSender send the 1st : " + msg1);
            this.rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_SECKILL, RabbitConfig.KEY_SECKILL, msg1,correlationData);
        }
    
        public String sendAndReceive(KillOrderVo vo) {
            String msg1 = JSON.toJSONString(vo);
            //设置消息唯一id
            CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
            //直接发送message对象
            MessageProperties messageProperties = new MessageProperties();
            //过期时间10秒,也是为了减少消息挤压的可能
    //        messageProperties.setExpiration("10000");
            messageProperties.setCorrelationId(correlationId.getId());
            Message message = new Message(msg1.getBytes(), messageProperties);
            log.info("TopicSender send the 1st : " + msg1);
            //设置消息唯一id
            Message message1 = rabbitTemplate.sendAndReceive(RabbitConfig.EXCHANGE_SECKILL, RabbitConfig.KEY_SECKILL, message, correlationId);
            return new String(message1.getBody());
        }
    }

5、业务生产信息的伪代码

@Autowired
        private SecKillSender secKillSender;
    
      @RabbitListener(queues = "order.seckill.producer"/*,errorHandler = "rabbitConsumerListenerErrorHandler"*/)
        @RabbitHandler // 此注解加上之后可以接受对象型消息
        public void process(Message message, Channel channel, @Headers Map<String, Object> headers) throws Exception {
            try {
                String msg = new String(message.getBody());
                log.info("UserReceiver>>>>>>>接收到消息:" + msg);
                try {
                    KillOrderVo vo = JSON.parseObject(msg, KillOrderVo.class);
                    Long orderId = orderService.killOrder(vo);
                    //把订单信息存储到缓存中
    //                setOrderToRedis(vo);
                    //发送消息到延迟队列
                    secKillSender.send(vo);
                    log.info("UserReceiver>>>>>>消息已消费");
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);//手工确认，可接下一条
                } catch (Exception e) {
                    System.out.println(e.getMessage());
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);//失败，则直接忽略此订单
                    log.info("UserReceiver>>>>>>拒绝消息，直接忽略");
                    throw e;
                }
            } catch (Exception e) {
                log.info(e.getMessage());
            }
        }

二、客户端消费者服务配置

1、pom文件配置同上

2、配置文件信息

spring: 
     rabbitmq:
        host: 94.191.83.120
        port: 5672
        username: admin
        password: admin
        publisher-confirms: true
        virtual-host: /
        publisher-returns: true

3、监听消费伪代码：

@RabbitListener(queues = "order.seckill.dead.queue"/*,errorHandler = "rabbitConsumerListenerErrorHandler"*/)
        @RabbitHandler // 此注解加上之后可以接受对象型消息
        public void processDead(Message message, Channel channel, @Headers Map<String, Object> headers) throws Exception {
            try {
                String msg = new String(message.getBody());
                log.info("order.seckill.dead.queue>>>>>>>consumer:" + msg);
                try {
                    KillOrderVo vo = JSON.parseObject(msg, KillOrderVo.class);
                    //1、校验订单是否已经支付，查询该订单的支付状态
                    Order order = orderService.search(vo.getOrderId());
                    //2、如果未支付就把订单取消，修改订单状态
                    if(order.getPayStatus() != null && "0".equals(order.getPayStatus().toString())) {
                        orderService.selfCancel(vo.getOrderId(),vo.getUserId());
                        //3、把库存+1操作
                        String killGoodCount = KillConstants.KILL_GOOD_COUNT + vo.getKillGoodsSpecPriceDetailVo().getId();
                        //返回的数值,执行了lua脚本
                        Long stock = stock(killGoodCount, 1, STOCK_LUA_INCR);
                        if(stock > 0) {
                            log.info("---------增加库存成功---stock:" + stock);
                        }
                    }
                    log.info("UserReceiver>>>>>>消息已消费");
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);//手工确认，可接下一条
                } catch (Exception e) {
                    log.info(e.getMessage());
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);//失败，则直接忽略此订单
                    log.info("UserReceiver>>>>>>拒绝消息，直接忽略");
                    throw e;
                }
            } catch (Exception e) {
                log.info(e.getMessage());
            }
        }

到此、rabbitmq 延时队列和死信队列的分析完毕，小伙伴多多联系，定会早日掌握。