SpringBoot2.3整合RabbitMQ实现延迟消费消息

时间 2020-06-10

标签 springboot2.3 springboot 整合 rabbitmq 实现延迟消费消息栏目 Spring 繁體版

原文原文链接

1.源码获取地址

文章末尾有源代码地址
https://www.sunnyblog.top/detail.html?id=1265257400324063232
本章节主要实现消息的延迟消费，在学习延迟消费以前必须先了解RabbitMQ两个基本概念，消息的TTL和死信Exchange，经过这二者的组合来实现消息的延迟消费。
不想看原理讲解的，直接经过标题6看代码实现html

2.消息的TTL(Time To Live)

消息的TTL就是消息的存活时间。RabbitMQ能够对队列和消息分别设置TTL。对队列设置就是队列没有消费者连着的保留时间，也能够对每个单独的消息作单独的设置。超过了这个时间，咱们认为这个消息就死了，称之为死信。app

3.死信交换器 Dead Letter Exchanges

一个消息在知足以下条件下，会进死信路由，记住这里是路由而不是队列，一个路由能够对应不少队列。
一个消息被Consumer拒收了，而且reject方法的参数里requeue是false。也就是说不会被再次放在队列里，被其余消费者使用。
上面的消息的TTL到了，消息过时了
队列的长度限制满了。排在前面的消息会被丢弃或者扔到死信路由上。死信交换器(Dead Letter Exchange)其实就是一种普通的exchange，和建立其余exchange没有两样。只是在某一个设置Dead Letter Exchange的队列中有消息过时了，会自动触发消息的转发，发送到Dead Letter Exchange中去。

4.实现延迟消费原理

大概原理:首先发送消息到死信队列，死信队列设置ttl过时时间，到期以后会自动将消息发送到通常队列实现消息的消费
实现步骤以下
建立死信交换器
建立死信队列
将死信队列与死信交换机绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）消息的发送方在向 Exchange发送消息时，也必须指定消息的RoutingKey。Exchange再也不把消息交给每个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的Routing key彻底一致，才会接收到消息.
建立正常交换器
建立正常队列
将正常队列绑定到正常交换器

5.基于案例实现消息的延迟消费

这里咱们以最熟悉的12306购票为例进行案例场景的分析，12306购票步骤以下：学习

首先登陆12306根据日期和起点站等条件进行抢票下订单
抢到票订单处于未支付状态,并提示支付时间30分钟内

这里就能够使用到延时队列，在下订单完成的时候将订单号发送到MQ的死信队列，并设置30分钟过时，30分钟之后死信队列的数据会转发到正常队列，从正常队列中获取到下订单的订单号，而后咱们根据订单号查询订单的支付状态，若是已经支付咱们不作任何操做，若是未支付取消订单，关闭支付状态，将票回滚到票池供其余用户购买

6.代码实现

在RabbitMQConfig中建立队列、交换机以及绑定关系测试

@Configuration
public class RabbitMQConfig {ui
```
/**
     * 测试发送消息到MQ
     * @return
```

*/
            @Bean
            public Queue testHello() {
                    return new Queue(SysConstant.QUEUE_TEST_HELLO);
            }


            /**
             * 死信交换机
             * @return
             */
            @Bean
            public DirectExchange sysOrderDelayExchange() {
                    return new DirectExchange(SysConstant.SYS_ORDER_DELAY_EXCHANGE);
            }

            /**
             * 死信队列
             * @return
             */
            @Bean
            public Queue sysOrderDelayQueue() {
                    Map<String, Object> map = new HashMap<String, Object>(16);
                    map.put("x-dead-letter-exchange",SysConstant.SYS_ORDER_RECEIVE_EXCHANGE); //指定死信送往的交换机
                    map.put("x-dead-letter-routing-key", SysConstant.SYS_ORDER_RECEIVE_KEY); //指定死信的routingkey
                    return new Queue(SysConstant.SYS_ORDER_DELAY_QUEUE, true, false, false, map);
            }

            /**
             * 给死信队列绑定死信交换机
             * @return
             */
            @Bean
            public Binding sysOrderDelayBinding() {
                    return BindingBuilder.bind(sysOrderDelayQueue()).to(sysOrderDelayExchange()).with(SysConstant.SYS_ORDER_DELAY_KEY);
            }

            /**
             * 死信接收交换机,用于接收死信队列的消息
             * @return
             */
            @Bean
            public DirectExchange sysOrderReceiveExchange() {
                    return new DirectExchange(SysConstant.SYS_ORDER_RECEIVE_EXCHANGE);
            }

            /**
             * 死信接收队列
             * @return
             */
            @Bean
            public Queue sysOrderReceiveQueue() {
                    return new Queue(SysConstant.SYS_ORDER_RECEIVE_QUEUE);
            }

            /**
             * 死信接收交换机绑定接收死信队列消费队列
             * @return
             */
            @Bean
            public Binding sysOrdeReceiveBinding() {
                    return BindingBuilder.bind(sysOrderReceiveQueue()).to(sysOrderReceiveExchange()).with(SysConstant.SYS_ORDER_RECEIVE_KEY);
            }
    }

发送延时消息到死信交换器方法spa

@Service
public class MsgService {日志

@Autowired
    private RabbitTemplate rabbitTemplate;
    /**
     * 发送延时消息到mq
     * @param exchange 死信交换机
     * @param routeKey 路由key
     * @param data 发送数据
     * @param delayTime 过时时间，单位毫秒

*/
                    public void sendDelayMsgToMQ(String exchange, String routeKey, String data,int delayTime) {
                            rabbitTemplate.convertAndSend(exchange, routeKey, data, message -> {
                                    message.getMessageProperties().setExpiration(delayTime + "");
                                    return message;
                            });
                    }
            }

监听队列消息ReceiveMsgListener类code

/**orm

* 获取到的延时消息
    * 这里接收到消息进行对应的业务处理(例如:取消订单，关闭支付，回滚库存等 ...)
    * @param msg

*/
            @RabbitListener(queues = SysConstant.SYS_ORDER_RECEIVE_QUEUE)
            @RabbitHandler
            public void getdelayMsg(String msg) {
                    log.info("MQ接收消息时间:{},消息内容:{}", DateUtil.formatDateTime(DateUtil.date()),msg);
                    log.info("------->这里实现订单关闭、支付关闭、回滚库存业务逻辑...");
            }

建立Controller向队列发送消息，设置过时时间10秒htm

@RestController
@RequestMapping("mq")
@Slf4j
public class MQController {

@Autowired
    private MsgService msgService;

    @GetMapping("sendMsg")
    public String sendMsg() {
            log.info("发送延时消息时间:" + DateUtil.formatDateTime(DateUtil.date()));

            OrderInfo orderInfo = new OrderInfo();
            orderInfo.setOrderId(IdUtil.fastSimpleUUID());
            orderInfo.setOrderState("待支付");
            orderInfo.setPayMoney(999.88);
            msgService.sendDelayMsgToMQ(SysConstant.SYS_ORDER_DELAY_EXCHANGE,SysConstant.SYS_ORDER_DELAY_KEY, JSONUtil.toJsonStr(orderInfo),10*1000);//1分钟
            return JSONUtil.toJsonStr("发送延时消息成功");
    }

}

启动服务，能够看到MQ中建立对应的队列和交换器

控制台日志能够看到发送消息与消费消息间隔时间是10s

详细开发技术文档尽在点击这里查看技术文档；更多技术文章： https://www.sunnyblog.top；任何疑问加QQ群咨询:534073451