Spring Cloud Stream 进阶配置——使用延迟队列

作者: sprainkle | 来源:发表于2020-03-24 21:25 被阅读0次

ps: 本文所有代码可在这里查看。

延迟队列

延迟队列 操作的对象是延迟消息，所谓 “延迟消息” 是指当消息被发送以后，并不想让消费者立刻消费消息，而是等待特定时间后，消费者才能拿到消息进行消费。

延迟队列比较经典的使用场景有：

在订单系统中，用户下单后，如果未在规定时间内（比如30分钟）支付，那么该订单会被关闭，即自动取消订单。
用户希望通过手机远程控制家里的智能设备在指定的时间进行工作。这时候可以将用户指令发送到延迟队列，当指令时间到了，再将指令推送到智能设备。

基于 RabbitMQ 的延迟队列

使用死信队列实现延迟队列

在 AMQP 协议中，或者 v3.5.8 之前的 RabbitMQ 本身并没有直接支持延迟队列功能，要想实现类似延迟队列的功能，可以通过死信队列的配合。即定义一组 ttl 为特定时长的队列，比如：5秒，10秒，30秒，1分钟等，然后再对这些队列，分别定义死信队列，当消息过期时，就会转存到相应的死信队列（即延迟队列）中，这样消费者根据业务自身的情况，分别选择不同延迟等级的延迟队列进行消费。

使用延迟消息交换机插件实现延迟队列

上面介绍的延迟队列实现方式，其实是比较繁琐的，好在，在版本 v3.5.8之后，RabbitMQ 推出了一个延迟消息交换机插件：rabbitmq_delayed_message_exchange，当启用该插件后，如果有一个队列声明为延迟交换机，那么当有消息发送到该交换机后，会根据延迟时长来决定投递的顺序，而如果延迟时长小于零，那么会立刻投递到相应的队列。

第一种实现方式，不在本文的讨论范围，就不细说，下面将对第二种实现方式进行介绍。

ps：RabbitMQ 的版本最好是 3.6.x 及以上，Erlang/OTP 的版本要在 18.0 以上。

使用延迟消息交换机插件

下载插件

因为该插件默认是没有在 RabbitMQ 的软件包的 plugins 目录下，需要自己下载然后放到 plugins 目录下，下载地址如下：

下载下来后，解压，然后拷贝到 plugins 目录下，如果是通过 rpm 是方式安装，目录应该是：/usr/lib/rabbitmq/lib/rabbitmq_server-3.7.15/plugins；如果是 Mac 用户，且使用 brew 安装，目录则在：/usr/local/Cellar/rabbitmq/3.7.7/plugins。

启用插件

# 启用插件 rabbitmq-delayed-message-exchange
rabbitmq-plugins enable rabbitmq-delayed-message-exchange

配合 Spring Cloud Stream 使用延迟交换机

首先来看一下延迟交换机如何配置：

spring:
  cloud:
    stream:
      bindings:
        delayedQueueOutput:
          destination: delayedQueueTopic
          content-type: application/json
          binder: rabbit

        delayedQueueInput:
          destination: delayedQueueTopic
          content-type: application/json
          group: ${spring.application.name}
          binder: rabbit

      rabbit:
        bindings:
          delayedQueueOutput:
            producer:
              delayedExchange: true # 是否将目标exchange声明为一个延迟消息交换机，默认false。即消息productor发布消息到延迟exchange后，延迟n长时间后才将消息推送到指定的queue中。 -RabbitMQ需要安装/启用插件: rabbitmq-delayed-message-exchange

          delayedQueueInput:
            consumer:
              delayedExchange: true # 是否将目标exchange声明为一个延迟消息交换机，默认false。即消息productor发布消息到延迟exchange后，延迟n长时间后才将消息推送到指定的queue中。 -RabbitMQ需要安装/启用插件: rabbitmq-delayed-message-exchange

重点关注2个配置：spring.cloud.stream.rabbit.bindings.ChannelName.producer.delayedExchange 和 spring.cloud.stream.rabbit.bindings.ChannelName.consumer.delayedExchange。

这2个配置分别属于生产者和消费者的配置，但都是用于告诉 Spring Cloud Stream 是否将交换机声明为一个延迟消息交换机。这2个是成对出现，如果少配置了一个，服务启动时会报一个警告，下文会说明。

延迟消息交换机的相关配置就这么简单，接下来通过测试用例来看一下效果。

ScasDelayedTest

@Data
@NoArgsConstructor
@AllArgsConstructor
public class DelayModel {

    /**
     * 延迟投递的时长. 单位: ms
     */
    private long delay;

}

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
@ActiveProfiles("delayed")
@EnableBinding({ScasDelayedTest.MessageSink.class, ScasDelayedTest.MessageSource.class})
public class ScasDelayedTest {

    @Autowired
    private DelayedQueueProducer delayedQueueProducer;

    @Test
    public void test() throws InterruptedException {
        for (int i = 0; i < 5; i++) {
            // 随机延迟 3-8 秒
            long delay = RandomUtil.randomLong(3, 8) * 1000;
            delayedQueueProducer.publish(new DelayModel(delay));
        }

        Thread.sleep(1000000);
    }

    @Component
    public static class DelayedQueueProducer {

        @Autowired
        private MessageSource messageSource;

        public void publish(DelayModel model) {
            long delay = model.getDelay();
            Message<DelayModel> message = MessageBuilder.withPayload(model).setHeader("x-delay", delay).build();
            messageSource.delayedQueueOutput().send(message);
            log.info("发布延迟队列消息: {}", model);
        }

    }

    @Component
    public static class DelayedQueueHandler {

        @StreamListener("delayedQueueInput")
        public void handle(DelayModel model) throws InterruptedException {
            log.info("消费延迟队列的消息. model: [{}].", model);
        }

    }

    public interface MessageSink {

        @Input("delayedQueueInput")
        SubscribableChannel delayedQueueInput();

    }

    public interface MessageSource {

        @Output("delayedQueueOutput")
        MessageChannel delayedQueueOutput();

    }

}

上面的代码很简单，重点是在构建消息时，比平常多了一个步骤，即 .setHeader("x-delay", delay)，其中变量 delay 为该消息需要延迟多久才被消费。

很好理解，通过 setHeader 方法，对 Message 添加一个名为 x-delay 的头部，对应的值则为延迟时长，单位为 ms。当该消息被投递到延迟交换机后，获取头部 x-delay 的值，如果小于0，那么立即将消息路由到相应的队列被消费，如果大于0，则延迟对应时间。

启动 ScasDelayedTest

启动测试用例后，控制台会出现类似如下图的输出：

ScasDelayedTest

查看延迟投递的消息数量

怎么查看延迟投递的消息数量？可以在 RabbitMQ Management 的对应交换机页面查看，
http://localhost:15672/#/exchanges/%2F/delayedQueueTopic

延迟投递的消息数量

ps: 为达到查看效果，可以适当增加延迟时长。

使用延迟队列实现“定时关闭超时未支付订单”

上面简单介绍了延迟交换机的使用方法，现在回到正题，如何使用延迟队列来实现 “定时关闭超时未支付订单” 呢？

针对上面的场景，一般的思路是：定义一个定时任务，比如每分钟查询一下订单表，找出接下来1、2钟内需要关闭的订单，然后再对每一笔订单执行 关闭订单 操作，当然在关闭之前需要再次确认订单是否 “已支付”。

为了简单，再通过一个测试用例来模拟一下具体场景。

ScasCloseUnpaidOrderTest

@Data
@NoArgsConstructor
@AllArgsConstructor
public class OrderModel {

    /**
     * 订单id
     */
    private Long id;

    /**
     * 订单失效时间
     */
    private Long expireTime;

    @Override
    public String toString() {
        return "OrderModel{" +
                "id=" + id +
                ", expireTime=" + TimeUtil.format(TimeUtil.toLocalDateTime(expireTime)) +
                '}';
    }
}

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
@ActiveProfiles("delayed")
@EnableBinding({ScasCloseUnpaidOrderTest.MessageSink.class, ScasCloseUnpaidOrderTest.MessageSource.class})
public class ScasCloseUnpaidOrderTest {

    @Autowired
    private CloseUnpaidOrderProducer closeUnpaidOrderProducer;

    @Test
    public void test() throws InterruptedException {

        // 模拟每分钟的0秒执行定时任务
        long toSleep = 60000 - System.currentTimeMillis() % 60000;
        Thread.sleep(toSleep);

        List<OrderModel> models = buildUnpaidOrderModel();
        for (OrderModel model : models) {
            closeUnpaidOrderProducer.publish(model);
        }

        Thread.sleep(1000000);

    }

    private List<OrderModel> buildUnpaidOrderModel() {

        long now = System.currentTimeMillis();

        List<OrderModel> models = new ArrayList<>(5);
        for (int i = 0; i < 5; i++) {

            long id = RandomUtil.randomLong(10000, 100000);
            // 模拟 订单将在小于60s内过期
            long expireTime = now + RandomUtil.randomLong(0, 60) * 1000;

            OrderModel model = new OrderModel();
            model.setId(id);
            model.setExpireTime(expireTime);
            models.add(model);
        }
        return models;
    }

    @Component
    public static class CloseUnpaidOrderProducer {

        @Autowired
        private MessageSource messageSource;

        public void publish(OrderModel model) {
            long now = System.currentTimeMillis();
            long delay = model.getExpireTime() - now;
            Message<OrderModel> message = MessageBuilder.withPayload(model).setHeader("x-delay", delay).build();
            messageSource.closeUnpaidOrderOutput().send(message);
            log.info("发布 [关闭超时未支付订单] 消息. delay: {}, model: {}", delay, model);
        }

    }

    @Component
    public static class CloseUnpaidOrderHandler {

        private Random random = new Random();

        @StreamListener("closeUnpaidOrderInput")
        public void handle(OrderModel model) throws InterruptedException {

            log.info("检查订单状态, 关闭支付超时订单. model: {}", model);

            if (isPaySuccess()) {
                log.info("订单 [{}] 支付超时. 关闭订单.", model.getId());
            } else {
                log.info("订单 [{}] 支付完成.", model.getId());
            }
        }

        private boolean isPaySuccess() {
            // 模拟从支付系统查询支付状态.
            return random.nextInt(10) % 3 == 0;
        }

    }

    public interface MessageSource {

        @Output("closeUnpaidOrderOutput")
        MessageChannel closeUnpaidOrderOutput();

    }

    public interface MessageSink {

        @Input("closeUnpaidOrderInput")
        SubscribableChannel closeUnpaidOrderInput();

    }

}

配置文件跟上一个测试用例基本一样：

spring:
  cloud:
    stream:
      bindings:
        closeUnpaidOrderOutput:
          destination: closeUnpaidOrderTopic
          content-type: application/json
          binder: rabbit

        closeUnpaidOrderInput:
          destination: closeUnpaidOrderTopic
          content-type: application/json
          group: ${spring.application.name}
          binder: rabbit

      rabbit:
        bindings:
          closeUnpaidOrderOutput:
            producer:
              delayedExchange: true # 是否将目标exchange声明为一个延迟消息交换机，默认false。即消息productor发布消息到延迟exchange后，延迟n长时间后才将消息推送到指定的queue中。 -RabbitMQ需要安装/启用插件: rabbitmq-delayed-message-exchange

          closeUnpaidOrderInput:
            consumer:
              delayedExchange: true # 是否将目标exchange声明为一个延迟消息交换机，默认false。即消息productor发布消息到延迟exchange后，延迟n长时间后才将消息推送到指定的queue中。 -RabbitMQ需要安装/启用插件: rabbitmq-delayed-message-exchange

启动 ScasCloseUnpaidOrderTest

启动后，可以看到控制台有类似输出：

ScasCloseUnpaidOrderTest

相信上面的代码对应各位看官来说，理解起来肯定是毫无压力的，这里就不在赘述。

Spring Cloud Stream 进阶配置——使用延迟队列

延迟队列

基于 RabbitMQ 的延迟队列

使用死信队列实现延迟队列

使用延迟消息交换机插件实现延迟队列

使用延迟消息交换机插件

下载插件

启用插件

配合 Spring Cloud Stream 使用延迟交换机

ScasDelayedTest

启动 ScasDelayedTest

查看延迟投递的消息数量

使用延迟队列实现“定时关闭超时未支付订单”

ScasCloseUnpaidOrderTest

启动 ScasCloseUnpaidOrderTest

相关链接

推荐阅读

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

SpringBoot