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，主要是为了手动触发行锁。这里是使用的默认的事务隔离级别，可重复读。因为事务里的读操作默认是不会触发行锁的，所以这里可能会出现另一个事务将用户信息改了，并且提交了，由于可重复读的问题，当前事务中读取到的仍是修改前的数据，那么当前事务提交的时候，就会将另一个事务的提交结果覆盖掉，如果这里不触发行锁，会导致数据的不一致性。
• 加了行锁之后，可以确保只有一个会话可以访问该订单数据，从而避免并发问题。但是也不是所有在事务里的读操作都要加行锁，毕竟那样的执行效率就太慢了，只有涉及到读取信息，并且后续需要对该信息进行修改的时候，才加上行锁。