Redisson延迟队列实现倒计时任务

问题背景

• 项目里刚好需要实现一个延迟订单取消任务。具体而言，如果一份订单在生成后的15分钟内未完成支付，系统需要自动取消该订单，并返还相关订单所使用的优惠券或免费额度等资源。
• 虽然引入MQ或者Kafka也是一种解决方法，但出于最大程度减少系统复杂性的角度考虑，强烈建议充分利用已有的Redis组件（例如Redisson）来解决这一问题，而不引入新组件。这样可以提高效率、减少维护负担，并确保充分发挥已有技术的潜力。

延迟队列

• Redisson中定义了分布式延迟队列RDelayedQueue，是一种基于zset结构实现的延时队列，，它允许以指定的延迟时长，将任务放到目标队列中。
• 其实就是在zset的基础上增加了一个基于内存的延迟队列，当我们要添加一个数据到延迟队列的时候，Redission会把数据和超时时间放到zset中，并且起一个延时任务，当任务到期时，再去zset中把数据取出来，返回给客户端使用。

解决方案

• 其实这里的实现和我之前写的消息队列差不多，感兴趣的可以去看我前面那篇文
  

  手写一个消息队列

  https://cyborg2077.github.io/2023/10/03/DIYTaskQueue/

1. 导入依赖

   <dependency>
       <groupId>org.redisson</groupId>
       <artifactId>redisson</artifactId>
       <version>3.16.2</version> 
   </dependency>

2. 编写配置类

   @Configuration
   public class RedisConfig {
   
       @Value("${spring.redis.host}")
       private String REDIS_HOST;  //地址配置在配置文件上
   
       @Value("${spring.redis.port}")
       private String REDIS_PORT;
   
       @Value("${spring.redis.password}")
       private String REDIS_PASSWORD;
   
       @Bean
       public RedissonClient createRedisAPi(){
           Config redissonConfig = new Config();
           redissonConfig.setCodec(new org.redisson.client.codec.StringCodec());
           //我这里单节点演示一下
           SingleServerConfig singleServerConfig = redissonConfig.useSingleServer();
           singleServerConfig.setAddress(String.format("redis://%s:%s", REDIS_HOST, REDIS_PORT));
           singleServerConfig.setPassword(REDIS_PASSWORD);
           //设置几号数据库
           singleServerConfig.setDatabase(0);
           singleServerConfig.setConnectTimeout(10000);
           singleServerConfig.setConnectionPoolSize(300);
           return Redisson.create(redissonConfig);
       }
   }

3. 延迟队列执行器

   @Slf4j
   public class DelayTaskQueueExecutor<T> {
       final private RBlockingQueue<T> queue;
       final private Thread msgLooper;
       private final DelayTaskQueueExecutor.Processor<T> processor;
   
       public interface Processor<T> {
           void process(T task) throws InterruptedException;
       }
   
       public DelayTaskQueueExecutor(String threadName, RBlockingQueue<T> queue, DelayTaskQueueExecutor.Processor<T> processor) {
           this.queue = queue;
           this.processor = processor;
           this.msgLooper = new Thread(this::looper);
           this.msgLooper.setName(threadName);
           this.msgLooper.start();
       }
   
       private void looper() {
           while(true) {
               try {
                   T task = queue.take();
                   processor.process(task);
               } catch (InterruptedException e) {
                   break;
               } catch (Exception e) {
                   log.error(String.format("TaskQueueExecutor %s run task exception",
                           this.msgLooper.getName()), e);
               }
           }
       }
   }

4. 服务类

   @Slf4j
   @Service
   public class DelayQueueService {
       @Resource
       private OrderMapper orderMapper;
   
       @Resource
       private UserInfoMapper userInfoMapper;
       
       @Resource
       private BonusInfoMapper bonusInfoMapper;
   
       @Autowired
       private RedissonClient redisson;
   
       private RDelayedQueue<String> delayedQueue;
   
       @PostConstruct
       public void initDelayQueue() {
           RBlockingQueue<String> blockingQueue = redisson.getBlockingQueue("orderDelayQueue");
           delayedQueue = redisson.getDelayedQueue(blockingQueue);
           new DelayTaskQueueExecutor<>("ORDER_DELAY", blockingQueue, this::processOrder);
       }
   
       /**
        * 将订单信息加入到延迟队列中，并设置TTL
        *
        * @param orderId 订单信息
        */
       public void addToDelayQueue(String orderId) {
           delayedQueue.offer(orderId, 2, TimeUnit.MINUTES);
       }
   
       @Transactional
       public void processOrder(String orderId) {
           Torder order = orderMapper.selectOne(Wrappers.lambdaQuery(Torder.class).eq(Torder::getOrderId, orderId));
           if (order != null) {
               Integer status = order.getStatus();
               // 如果订单仍然是未支付状态，并且字数已经预扣了，则返还预扣的字数，并且关闭订单
               if (status == Common.ORDER_ORIGINAL_STATUS) {
                   order.setStatus(Common.ORDER_CLOSED_STATUS);
                   String userId = order.getToken();
                   // 查询是否是登录用户
                   UserInfo userInfo = userInfoMapper.selectOneForUpdate(userId);
                   if (userInfo != null) {
                       UsedWordsBonusInfo wordsBonusInfo = usedWordsBonusInfoMapper.selectOne(Wrappers.lambdaQuery(UsedWordsBonusInfo.class).eq(UsedWordsBonusInfo::getOrderId, order.getOrderId()));
                       // 判断是否已经预扣了
                       if (wordsBonusInfo.getSelfWordsDeducted() == Common.ALREADY_DEDUCTION) {
                           userInfo.setRegisterBonusWords(userInfo.getRegisterBonusWords() + wordsBonusInfo.getUsedRegisterBonus());
                           userInfo.setInviteBonusWords(userInfo.getInviteBonusWords() + wordsBonusInfo.getUsedInviteBonus());
                           log.info("订单：{}，返还注册字数：{}，邀请字数：{}", orderId, wordsBonusInfo.getUsedRegisterBonus(), wordsBonusInfo.getUsedInviteBonus());
                           wordsBonusInfo.setSelfWordsDeducted(Common.ORIGIN_STATUS);
                           usedWordsBonusInfoMapper.updateById(wordsBonusInfo);
                           userInfoMapper.updateById(userInfo);
                       }
                   }
               }
               orderMapper.updateById(order);
           }
       }
   }

5. 使用方法，创建订单的时候将订单号加入到队列中，到期后会自动执行关闭订单的对应逻辑

   delayQueueService.addToDelayQueue(orderId);

注意事项

• 细心的同学可能注意到了我这里的事务

  @Transactional
    public void processOrder(String orderId) {
        Torder order = orderMapper.selectOne(Wrappers.lambdaQuery(Torder.class).eq(Torder::getOrderId, orderId));
        if (order != null) {
            Integer status = order.getStatus();
            // 如果订单仍然是未支付状态，并且字数已经预扣了，则返还预扣的字数，并且关闭订单
            if (status == Common.ORDER_ORIGINAL_STATUS) {
                order.setStatus(Common.ORDER_CLOSED_STATUS);
                String userId = order.getToken();
                // 查询是否是登录用户
                UserInfo userInfo = userInfoMapper.selectOneForUpdate(userId);
                if (userInfo != null) {
                    UsedWordsBonusInfo wordsBonusInfo = usedWordsBonusInfoMapper.selectOne(Wrappers.lambdaQuery(UsedWordsBonusInfo.class).eq(UsedWordsBonusInfo::getOrderId, order.getOrderId()));
                    // 判断是否已经预扣了
                    if (wordsBonusInfo.getSelfWordsDeducted() == Common.ALREADY_DEDUCTION) {
                        userInfo.setRegisterBonusWords(userInfo.getRegisterBonusWords() + wordsBonusInfo.getUsedRegisterBonus());
                        userInfo.setInviteBonusWords(userInfo.getInviteBonusWords() + wordsBonusInfo.getUsedInviteBonus());
                        log.info("订单：{}，返还注册字数：{}，邀请字数：{}", orderId, wordsBonusInfo.getUsedRegisterBonus(), wordsBonusInfo.getUsedInviteBonus());
                        wordsBonusInfo.setSelfWordsDeducted(Common.ORIGIN_STATUS);
                        usedWordsBonusInfoMapper.updateById(wordsBonusInfo);
                        userInfoMapper.updateById(userInfo);
                    }
                }
            }
            orderMapper.updateById(order);
        }
    }

• 重点是我这里的userInfoMapper.selectOneForUpdate(userId);，这句其实是我手写的一个SQL，SELECT XXX FROM user WHERE userId = xxx FOR UPDATE，主要是为了手动触发行锁。这里是使用的默认的事务隔离级别，可重复读。因为事务里的读操作默认是不会触发行锁的，所以这里可能会出现另一个事务将用户信息改了，并且提交了，由于可重复读的问题，当前事务中读取到的仍是修改前的数据，那么当前事务提交的时候，就会将另一个事务的提交结果覆盖掉，如果这里不触发行锁，会导致数据的不一致性。
• 加了行锁之后，可以确保只有一个会话可以访问该订单数据，从而避免并发问题。但是也不是所有在事务里的读操作都要加行锁，毕竟那样的执行效率就太慢了，只有涉及到读取信息，并且后续需要对该信息进行修改的时候，才加上行锁。