用路由的交换机 通过routingKey 将消息分发到对应的队列
每一条消息 分别 放入一个队列使用完成后删除队列

定义死信队列 死信交换机 普通交换机

  1. package com.matrix.queue.rabbitmq.config;
  3. import com.matrix.queue.rabbitmq.constant.QueuePassConstant;
  4. import org.springframework.amqp.core.Binding;
  5. import org.springframework.amqp.core.BindingBuilder;
  6. import org.springframework.amqp.core.DirectExchange;
  7. import org.springframework.amqp.core.Queue;
  8. import org.springframework.context.annotation.Bean;
  9. import org.springframework.context.annotation.Configuration;
  11. import java.util.HashMap;
  12. import java.util.Map;
  14. /**
  15.  * @ClassName: QueueConfiguration
  16.  * @Package:   com.matrix.common.config
  17.  * @description: 消息队列-演示队列 配置类
  18.  * @author: swordmeng
  19.  * @date: 2021.11.10
  20.  * @version v1.0
  21.  */
  22. @Configuration
  23. public class QueuePassConfiguration {
  26.     //====================1.声明正常队列====================
  28.     /**
  29.      * 排队 信道配置
  30.      * @return
  31.      */
  32.     @Bean
  33.     public DirectExchange goodsExchange() {
  34.         return new DirectExchange(QueuePassConstant.ANTIEPIDEMIC_QUEUE_EXCHANGE, true, false);
  35.     }
  37.     /**
  38.      * 排队 队列
  39.      * @return:
  40.      */
  41.     @Bean
  42.     public Queue goodsQueue() {
  43.         Map<String, Object> queueArgs = new HashMap<>();
  44.         queueArgs.put("x-dead-letter-exchange", QueuePassConstant.ANTIEPIDEMIC_QUEUE_DLX_EXCHANGE);
  45.         Queue queue = new Queue(QueuePassConstant.ANTIEPIDEMIC_QUEUE_PASS_NAME,true,false,false,queueArgs);
  46.         return queue;
  47.     }
  49.     /**
  50.      * 排队 队列绑定
  51.      * @return
  52.      */
  53.     @Bean
  54.     public Binding goodsBinding() {
  55.         return BindingBuilder.bind(goodsQueue()).to(goodsExchange()).with(QueuePassConstant.ANTIEPIDEMIC_QUEUE_EXCHANGE_ROUTING_KEY);
  56.     }
  57. 普通队列和 绑定其实在 这里是用不到的
  59.     //====================2.声明死信队列====================
  60.     /**
  61.      * 死信 信道配置
  62.      * @return
  63.      */
  64.     @Bean
  65.     public DirectExchange queueDlxExchange() {
  66.         return new DirectExchange(QueuePassConstant.ANTIEPIDEMIC_QUEUE_DLX_EXCHANGE, true, false);
  67.     }
  69.     /**
  70.      * 排队 死信队列
  71.      * @return
  72.      */
  73.     @Bean
  74.     public Queue queueDlxQueue() {
  75.         Queue queue = new Queue(QueuePassConstant.ANTIEPIDEMIC_QUEUE_DLX_PASS_NAME,true,false,false);
  76.         return queue;
  77.     }
  79.     /**
  80.      * 绑定 死信队列
  81.      * @author  swordmeng
  82.      * @Datetime  2021/11/12 9:40
  83.      * @return:
  84.      */
  85.     @Bean
  86.     public Binding queueDlxBinding() {
  87.         return BindingBuilder.bind(queueDlxQueue()).to(queueDlxExchange()).with(QueuePassConstant.ANTIEPIDEMIC_QUEUE_EXCHANGE_ROUTING_KEY);
  88.     }
  90. }

nacos配置

  1. spring:   
  2.   rabbitmq:
  3.     host: 119.3.13.80
  4.     port: 5672
  5.     username: admin
  6.     password: matrix2019
  7.     virtual-host: /alarm
  8.     publisher-confirms: true #是否开启confirms机制
  9.     publisher-returns: true #是否开启returns机制
  10.     connection-timeout: 10000 #过期时间
  11.     listener: # 消费者监听
  12.       simple:
  13.         concurrency: 5                   
  14.         max-concurrency: 10              
  15.         acknowledge-mode: manual # ACK 手动签收
  16.         prefetch: 2

rabbitmq配置类引入

  1. package com.matrix.queue.rabbitmq.entity;
  3. import lombok.Data;
  4. import org.springframework.boot.context.properties.ConfigurationProperties;
  5. import org.springframework.stereotype.Component;
  6. @Data
  7. @Component
  8. @ConfigurationProperties(prefix = "spring.rabbitmq")
  9. public class RabbitMQProperties {
  10.     /**
  11.      * rabbit 地址
  12.      */
  13.     private String host;
  14.     /**
  15.      * rabbit 端口
  16.      */
  17.     private String port;
  18.     /**
  19.      * rabbit 用户名
  20.      */
  21.     private String username;
  22.     /**
  23.      * rabbit password
  24.      */
  25.     private String password;
  26.     /**
  27.      * rabbit virtualHost
  28.      */
  29.     private String virtualHost;
  30.     /**
  31.      * rabbit  publisher-confirms
  32.      */
  33.     private String  publisherConfirms;
  34.     /**
  35.      * rabbit publisher-returns
  36.      */
  37.     private String  publisherReturns;
  38.     /**
  39.      * rabbit  connection-timeout
  40.      */
  41.     private String   connectionTimeout;
  43. }

生产者

  1. package com.matrix.queue.rabbitmq.service.queuepass.service.impl;
  3. import com.matrix.queue.rabbitmq.constant.QueuePassConstant;
  4. import com.matrix.queue.rabbitmq.entity.QueuePassMessage;
  5. import com.matrix.queue.rabbitmq.entity.RabbitMQProperties;
  6. import com.matrix.queue.rabbitmq.service.comm.CommSendService;
  7. import com.matrix.queue.rabbitmq.service.queuepass.service.IQueuePassService;
  8. import com.rabbitmq.client.Channel;
  9. import com.rabbitmq.client.Connection;
  10. import com.rabbitmq.client.ConnectionFactory;
  11. import lombok.RequiredArgsConstructor;
  12. import lombok.extern.slf4j.Slf4j;
  13. import org.springblade.core.tool.utils.Func;
  14. import org.springframework.beans.factory.annotation.Autowired;
  15. import org.springframework.stereotype.Service;
  17. import java.io.IOException;
  18. import java.util.HashMap;
  19. import java.util.Map;
  20. import java.util.concurrent.TimeoutException;
  22. import static com.matrix.queue.rabbitmq.constant.QueuePassConstant.ANTIEPIDEMIC_QUEUE_EXCHANGE;
  23. import static com.matrix.queue.rabbitmq.constant.QueuePassConstant.ANTIEPIDEMIC_QUEUE_EXCHANGE_ROUTING_KEY;
  24. import static com.matrix.queue.rabbitmq.constant.QueuePassConstant.ANTIEPIDEMIC_QUEUE_SERVICE_NAME;
  26. /**
  27.  * ClassName:    GoodsOnlineServiceImpl
  28.  * Package:    com.matrix.rabbitmq.service.impl
  29.  * Description: 消息队列实现类
  30.  * @Datetime: 2021/11/9   15:59
  31.  * @Author: swordmeng8@163.com
  32.  */
  33. @Slf4j
  34. @Service(ANTIEPIDEMIC_QUEUE_SERVICE_NAME)
  35. @RequiredArgsConstructor(onConstructor = @__(@Autowired))
  36. public class QueuePassServiceImpl implements IQueuePassService {
  38.     /** 通用发送方法 */
  39.     private final CommSendService commSendService;
  40.     //注入rabbitmq的配置
  41.     private final RabbitMQProperties rabbitMQConfig;
  42.     /**
  43.      * 发送消息
  44.      * @param reservationQueueId
  45.      * @param times
  46.      * @return
  47.      */
  48.     @Override
  49.     public void send(Long reservationQueueId, long times, String callTime)  {
  51.         //临时队列 尝试解决 mq阻塞问题 后续可能用到
  52.         try {
  53.             QueuePassMessage shopMessage = new QueuePassMessage(reservationQueueId,times,callTime);
  54.             Map<String, Object> queueArgs = new HashMap<>();
  55.             queueArgs.put("x-dead-letter-exchange", QueuePassConstant.ANTIEPIDEMIC_QUEUE_DLX_EXCHANGE);
  56.             queueArgs.put("x-dead-letter-routing-key", ANTIEPIDEMIC_QUEUE_EXCHANGE_ROUTING_KEY);//x-dead-letter-routing-key
  57.             //获取connectionFactory
  58.             ConnectionFactory connectionFactory = new ConnectionFactory();
  59.             //配置链接属性
  60.             connectionFactory.setVirtualHost(rabbitMQConfig.getVirtualHost());
  61.             connectionFactory.setUsername(rabbitMQConfig.getUsername());
  62.             connectionFactory.setPassword(rabbitMQConfig.getPassword());
  63.             connectionFactory.setHost(rabbitMQConfig.getHost());
  64.             Connection connection = connectionFactory.newConnection();
  65.             Channel channel = connection.createChannel();
  66.             //获取随机 队列名
  67.             String queueName = Func.randomUUID();
  68.             shopMessage.setQueueName(queueName);
  69.             //生成队列
  70.             channel.queueDeclare(queueName, true, false, false, queueArgs);
  71.             //获取随机routingKey
  72.             String routingKey = Func.randomUUID();
  73.             //绑定队列
  74.             channel.queueBind(queueName,QueuePassConstant.ANTIEPIDEMIC_QUEUE_EXCHANGE,routingKey);
  75.             commSendService.commSend(shopMessage,times,ANTIEPIDEMIC_QUEUE_EXCHANGE,routingKey);
  76.             channel.close();
  77.             connection.close();
  78.             log.info("排队过号时间修改增加消息队列：商品id：{},失效时间Long值：{},叫号时间LocalDateTime值：{}，推送结果:{}",reservationQueueId,times,callTime,true);
  79.         } catch (IOException e) {
  80.             e.printStackTrace();
  81.         } catch (TimeoutException e) {
  82.             e.printStackTrace();
  83.         }
  84.         //最初的死信队列 但是 不同时时间有消息阻塞问题
  85.         QueuePassMessage shopMessage = new QueuePassMessage(reservationQueueId,times,callTime);
  86.     }
  87. }

给每条消息分别定时

  1. package com.matrix.queue.rabbitmq.service.comm;
  3. import com.alibaba.fastjson.JSON;
  4. import lombok.AllArgsConstructor;
  5. import org.springframework.amqp.AmqpException;
  6. import org.springframework.amqp.core.Message;
  7. import org.springframework.amqp.core.MessagePostProcessor;
  8. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  9. import org.springframework.stereotype.Service;
  11. /**
  12.  * @className: CommSendService
  13.  * @package: com.matrix.rabbitmq.service
  14.  * @description: 消息队列通用发送类
  15.  * @datetime: 2021/11/30   18:50
  16.  * @author: swordmeng8@163.com
  17.  */
  18. @Service
  19. @AllArgsConstructor
  20. public class CommSendService {
  22.     /** 使用RabbitTemplate,这提供了接收/发送等等方法 */
  23.     private final RabbitTemplate rabbitTemplate;
  26.     public void commSend(Object obj, long times,String exchange,String exchangeRoutingKey){
  27.         MessagePostProcessor processor = new MessagePostProcessor(){
  28.             @Override
  29.             public Message postProcessMessage(Message message) throws AmqpException {
  30.                 message.getMessageProperties().setExpiration(String.valueOf(times));
  31.                 return message;
  32.             }
  33.         };
  34.         rabbitTemplate.convertAndSend(
  35.             exchange,
  36.             exchangeRoutingKey,
  37.             JSON.toJSONString(obj), processor);
  38.     }
  41. }

消费者

  1. package com.matrix.queue.rabbitmq.listener;
  3. import com.alibaba.fastjson.JSONObject;
  4. import com.matrix.queue.rabbitmq.entity.QueuePassMessage;
  5. import com.matrix.queue.rabbitmq.service.comm.BasicAckService;
  6. import com.matrix.queue.reservationQueue.service.IReservationQueueService;
  7. import com.rabbitmq.client.Channel;
  8. import lombok.AllArgsConstructor;
  9. import lombok.extern.slf4j.Slf4j;
  10. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  11. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  12. import org.springframework.messaging.Message;
  13. import org.springframework.stereotype.Component;
  15. import static com.matrix.queue.rabbitmq.constant.QueuePassConstant.ANTIEPIDEMIC_QUEUE_DLX_PASS_NAME;
  17. /**
  18.  * ClassName:    MqMessageController
  19.  * Package:    com.matrix.rabbitmq.listener
  20.  * Description: 处理消息监听逻辑
  21.  * @Datetime: 2021/11/10   11:12
  22.  * @Author: swordmeng8@163.com
  23.  */
  24. @Slf4j
  25. @Component
  26. @AllArgsConstructor
  27. public class QueuePassListener {
  29.     /**排队信息*/
  30.     private final IReservationQueueService reservationQueueService;
  31.     /** 签收服务 */
  32.     private final BasicAckService basicAckService;
  34.     /**
  35.      * 描述：处理排队过号状态
  36.      * 使用 @Payload 和 @Headers 注解可以获取消息中的 body 与 headers 信息
  37.      * @param message
  38.      * @param channel
  39.      */
  40.     @RabbitHandler
  41.     @RabbitListener(queues = ANTIEPIDEMIC_QUEUE_DLX_PASS_NAME)
  42.     public void handleQueuePassStatus(Message message, Channel channel) {
  43.         QueuePassMessage queuePassMessage = JSONObject.parseObject(message.getPayload().toString(), QueuePassMessage.class);
  44.         Long reservationQueueId = queuePassMessage.getReservationQueueId();
  45.         log.info("排队叫号消费消息，排队叫号消息信息:{}",queuePassMessage);
  47.         // 校验排队加号时间
  48.         boolean checkResult = reservationQueueService.checkQueuePass(reservationQueueId);
  49.         if (checkResult) {
  50.             // 更新排队过号状态和过号时间
  51.             //临时修改 状态码 9为3
  52.             boolean result = reservationQueueService.updateQueuePassStatus(9,reservationQueueId).isSuccess();
  53.             // 添加商品操作日志
  54.             if (!result) {
  55.                 log.info(reservationQueueId+"排队自动过号失败");
  56.             } else {
  57.                 log.info(reservationQueueId+"排队自动过号成功");
  58.             }
  59.         }
  60.         // 签收
  61.         basicAckService.basicAck(message,channel);
  63.     }
  66. }

签收

  1. package com.matrix.queue.rabbitmq.service.comm;
  3. import com.alibaba.fastjson.JSONObject;
  4. import com.matrix.queue.rabbitmq.entity.QueuePassMessage;
  5. import com.rabbitmq.client.Channel;
  6. import lombok.extern.slf4j.Slf4j;
  7. import org.springframework.amqp.support.AmqpHeaders;
  8. import org.springframework.messaging.Message;
  9. import org.springframework.messaging.MessageHeaders;
  10. import org.springframework.stereotype.Service;
  12. import java.io.IOException;
  14. /**
  15.  * @className: BasicAckService
  16.  * @package: com.matrix.rabbitmq.service.comm
  17.  * @description: BasicAck
  18.  * @datetime: 2021/11/30   18:57
  19.  * @author: swordmeng8@163.com
  20.  */
  21. @Slf4j
  22. @Service
  23. public class BasicAckService {
  25.     /**
  26.      * 手动签收
  27.      * @param message
  28.      * @param channel
  29.      */
  30.     public void basicAck(Message message, Channel channel){
  31.         try {
  32.             //手工签收
  33.             Long deliveryTag = (Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
  34.             log.info("接受deliveryTag:{}",deliveryTag);
  35.             channel.basicAck(deliveryTag, false);
  36.             删除 对应的队列 这是 临时队列代码尝试解决 mq 阻塞问题
  37.             QueuePassMessage queuePassMessage = JSONObject.parseObject(message.getPayload().toString(), QueuePassMessage.class);
  38.             String queueName = queuePassMessage.getQueueName();
  39.             channel.queueDelete(queueName);
  40.         } catch (
  41.             IOException e) {
  42.             e.printStackTrace();
  43.             log.info("手动签收失败：Cause:{},Message:{}",e.getCause(),e.getMessage());
  44.         }
  45.     }
  46. }