延迟队列预约提醒

一、背景

用户预约直播成功后，会在直播开始前20分钟或30分钟收到一个直播即将开始的短信。

也就是一个可变的定时任务

二、所用技术

Redisson

2.1 Redisson 介绍：

Redisson是一个在Redis的基础上实现的框架，它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务。

Redission中定义了分布式延迟队列RDelayedQueue，这是一种基于我们前面介绍过的zset结构实现的延时队列，它允许以指定的延迟时长将元素放到目标队列中。

其实就是在zset的基础上增加了一个基于内存的延迟队列。当我们要添加一个数据到延迟队列的时候，redission会把数据+超时时间放到zset中，并且起一个延时任务，当任务到期的时候，再去zset中把数据取出来，返回给客户端使用。

2.2 RDelayedQueue 源码分析

2.2.1 新建延迟队列

参考 QueueUtils#addDelayedQueueObject 方法可知，在添加数据前，先调用了 getDelayedQueue 方法获取一个延迟队列

RedissonDelayedQueue#RedissonDelayedQueue

2.2.2 添加数据

三、实际业务逻辑

3.1 在项目启动时初始化redis订阅

SystemApplicationRunner


@Slf4j
@RequiredArgsConstructor
@Component
public class SystemApplicationRunner implements ApplicationRunner {
		private final ISmsService smsService;

		@Override
    public void run(ApplicationArguments args) throws Exception {
				// 订阅redis队列
        RedisUtils.subscribeBlockingQueue("live", (String msg) -> {
            // 观察接收时间
            log.info("通道: {}, 收到数据: {}", "live", msg);
            // TODO: 2022/12/8 时间到了，该给用户发送短信了
            smsService.sendSms(msg);

        });
        log.info("开始订阅redis队列");

		}
}

3.2 当用户预约直播成功后，将该用户手机号等信息，加到队列中，并计算出，需要发送短信的时间


				boolean b = insert > 0;
        // 添加预约成功后，判断其是否需要发送短信
        // 当前时间距离直播开始时间小于提前提醒时间，发送短信
        // 当前时间距离直播开始时间
        long time = liveRome.getLiveTime().getTime() - System.currentTimeMillis();
        // 提前提醒时间
        long remindTime = liveRome.getRemindTime() * 60 * 1000;
        if (b && time > remindTime) {
            // 发送短信
            String mobile = dto.getMobile();
            String content = "您已成功预约" + liveRome.getTitle() + "直播，直播将于" + liveRome.getRemindTime() + "分钟后开始，请提前做好准备。";
            SmsVo smsVo = new SmsVo();
            smsVo.setMobile(mobile);
            smsVo.setCode(content);
            String smsJson = JSON.toJSONString(smsVo);

            // 加入队列，时间为  time - remindTime 再转化成秒
            RedisUtils.addDelayedQueueObject("live", smsJson, (time - remindTime) / 1000, TimeUnit.SECONDS);
            // 观察发送时间
            log.info("通道: {} , 发送数据: {}, 将于{}秒后发送短信", "live", smsJson, (time - remindTime) / 1000);
        }