实体类

为了方便测试，直接在测试类中的写内部类：

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public class OrderInfo {
        /**
         * 订单id
         */
        private Integer id;
        /**
         * 描述：用来记录关闭时间，可以在测试时用来验证。关闭时间是否跟 expireTime相等
         */
        private String description;
        /**
         * 创建时间
         */
        private LocalDateTime createTime;
        /**
         * 过期时间：关闭时间
         */
        private LocalDateTime expireTime;
    }

生成订单

模拟生成订单并设置过期时间。

执行时会在redis创建2个key：

• redisson_delay_queue:{<closeKey> } ：订单数据

• redisson_delay_queue_timeout:{<closeKey> } ：zset类型，按时间戳排序

  /**
   * 创建订单，并设置过期时间
   *
   * @throws IOException
   */
  @Test
  void createOrder() {
      RBlockingDeque<OrderInfo> blockingDeque = redissonClient.getBlockingDeque(closeKey);
      RDelayedQueue<OrderInfo> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
      // 100条订单
      int n = 100;
      Random random = new Random();
      for (int i = 0; i < n; i++) {
          // 1~100之间的正整数
          int i1 = random.nextInt(100) + 1;
          LocalDateTime now = LocalDateTime.now();
          delayedQueue.offer(new OrderInfo(i + 1, "close: " + i1, now, now.plusSeconds(i1)), i1, TimeUnit.SECONDS);
      }
  }

关闭订单

关闭订单，这里会产生订阅。redis会出现redisson_delay_queue_channel。

    /**
     * 关闭订单
     *
     * @throws IOException
     */
    @Test
    void closeOrder() {
        ReentrantLock lock = new ReentrantLock();
        // 5个线程
        int poolSize = 5;
        List<CompletableFuture<Void>> futureList = new ArrayList<>();
        for (int i = 0; i < poolSize; i++) {
            futureList.add(CompletableFuture.runAsync(() -> {
                RBlockingDeque<OrderInfo> blockingDeque = redissonClient.getBlockingDeque(closeKey);
                // 加入监听
                redissonClient.getDelayedQueue(blockingDeque);
                while (true) {
                    OrderInfo take;
                    try {
                        take = blockingDeque.take();
                    } catch (Exception e) {
                        continue;
                    }
                    if (take == null) {
                        continue;
                    }
                    // 验证多次是否会重复关闭。正常里不会近，只是验证下。正式环境，可以删除
                    try {
                        lock.lock();
                        if(closed.contains(take.getId())){
                            log.info("测试是否会抢占：已存在其他线程处理关闭订单[{}]", take.getId());
                        }
                        closed.add(take.getId());
                    }finally {
                        lock.unlock();
                    }
                    // 处理订单关闭逻辑
                    log.info("订单[{}]关闭中。。。", take.getId());
                    log.info("订单[{}]已关闭！order={}", take.getId(), toJsonString(take));
                }
            }));
        }
        // 模拟正式环境中进程一直在运行，因为test时，没有join则会只执行一次出现消费完数据后进程就关闭了
        CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0])).join();
    }

完整测试类：

package cn.skyjilygao.demo;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

import static cn.skyjilygao.util.EntityUtil.toJsonString;

@Slf4j
@SpringBootTest
public class CloseOrderTests {
    @Autowired
    private RedissonClient redissonClient;
    public static String closeKey = "order_close_test";
    public volatile static Set<Integer> closed = new ConcurrentSkipListSet<>();

    /**
     * 创建订单，并设置过期时间
     *
     * @throws IOException
     */
    @Test
    void createOrder() {
        RBlockingDeque<OrderInfo> blockingDeque = redissonClient.getBlockingDeque(closeKey);
        RDelayedQueue<OrderInfo> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
        int a = 100;
        Random random = new Random(100);
        for (int i = 0; i < a; i++) {
            int i1 = random.nextInt(1 + i) + 1;
            delayedQueue.offer(new OrderInfo(i + 1, "close: " + i1, LocalDateTime.now(), LocalDateTime.now().plusSeconds(i1)), i1, TimeUnit.SECONDS);
        }
    }

    /**
     * 关闭订单
     *
     * @throws IOException
     */
    @Test
    void closeOrder() {
        ReentrantLock lock = new ReentrantLock();
        // 5个线程
        int poolSize = 5;
        List<CompletableFuture<Void>> futureList = new ArrayList<>();
        for (int i = 0; i < poolSize; i++) {
            futureList.add(CompletableFuture.runAsync(() -> {
                RBlockingDeque<OrderInfo> blockingDeque = redissonClient.getBlockingDeque(closeKey);
                // 加入监听
                redissonClient.getDelayedQueue(blockingDeque);
                while (true) {
                    OrderInfo take;
                    try {
                        take = blockingDeque.take();
                    } catch (Exception e) {
                        continue;
                    }
                    if (take == null) {
                        continue;
                    }
                    try {
                        lock.lock();
                        if(closed.contains(take.getId())){
                            log.info("测试是否会抢占：已存在其他线程处理关闭订单[{}]", take.getId());
                        }
                        closed.add(take.getId());
                    }finally {
                        lock.unlock();
                    }
                    log.info("订单[{}]关闭中。。。", take.getId());
                    log.info("订单[{}]已关闭！order={}", take.getId(), toJsonString(take));
                }
            }));
        }
        // 模拟正式环境中进程一直在运行，因为test时，没有join则会只执行一次出现消费完数据后进程就关闭了
        CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0])).join();
    }

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public class OrderInfo {
        private Integer id;
        private String description;
        private LocalDateTime createTime;
        private LocalDateTime expireTime;
    }
}