利用Redisson实现订单关闭

实体类

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

1.     @Data
2.     @AllArgsConstructor
3.     @NoArgsConstructor
4.     public class OrderInfo {
5.         /**
6.          * 订单id
7.          */
8.         private Integer id;
9.         /**
10.          * 描述：用来记录关闭时间，可以在测试时用来验证。关闭时间是否跟 expireTime相等
11.          */
12.         private String description;
13.         /**
14.          * 创建时间
15.          */
16.         private LocalDateTime createTime;
17.         /**
18.          * 过期时间：关闭时间
19.          */
20.         private LocalDateTime expireTime;
21.     }

复制代码

生成订单

模拟生成订单并设置过期时间。
执行时会在redis创建2个key：

• redisson_delay_queue:{ } ：订单数据
  
• redisson_delay_queue_timeout:{ } ：zset类型，按时间戳排序
  

1.     /**
2.      * 创建订单，并设置过期时间
3.      *
4.      * @throws IOException
5.      */
6.     @Test
7.     void createOrder() {
8.         RBlockingDeque<OrderInfo> blockingDeque = redissonClient.getBlockingDeque(closeKey);
9.         RDelayedQueue<OrderInfo> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
10.         // 100条订单
11.         int n = 100;
12.         Random random = new Random();
13.         for (int i = 0; i < n; i++) {
14.             // 1~100之间的正整数
15.             int i1 = random.nextInt(100) + 1;
16.             LocalDateTime now = LocalDateTime.now();
17.             delayedQueue.offer(new OrderInfo(i + 1, "close: " + i1, now, now.plusSeconds(i1)), i1, TimeUnit.SECONDS);
18.         }
19.     }

复制代码

关闭订单

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

1.     /**
2.      * 关闭订单
3.      *
4.      * @throws IOException
5.      */
6.     @Test
7.     void closeOrder() {
8.         ReentrantLock lock = new ReentrantLock();
9.         // 5个线程
10.         int poolSize = 5;
11.         List<CompletableFuture<Void>> futureList = new ArrayList<>();
12.         for (int i = 0; i < poolSize; i++) {
13.             futureList.add(CompletableFuture.runAsync(() -> {
14.                 RBlockingDeque<OrderInfo> blockingDeque = redissonClient.getBlockingDeque(closeKey);
15.                 // 加入监听
16.                 redissonClient.getDelayedQueue(blockingDeque);
17.                 while (true) {
18.                     OrderInfo take;
19.                     try {
20.                         take = blockingDeque.take();
21.                     } catch (Exception e) {
22.                         continue;
23.                     }
24.                     if (take == null) {
25.                         continue;
26.                     }
27.                     // 验证多次是否会重复关闭。正常里不会近，只是验证下。正式环境，可以删除
28.                     try {
29.                         lock.lock();
30.                         if(closed.contains(take.getId())){
31.                             log.info("测试是否会抢占：已存在其他线程处理关闭订单[{}]", take.getId());
32.                         }
33.                         closed.add(take.getId());
34.                     }finally {
35.                         lock.unlock();
36.                     }
37.                     // 处理订单关闭逻辑
38.                     log.info("订单[{}]关闭中。。。", take.getId());
39.                     log.info("订单[{}]已关闭！order={}", take.getId(), toJsonString(take));
40.                 }
41.             }));
42.         }
43.         // 模拟正式环境中进程一直在运行，因为test时，没有join则会只执行一次出现消费完数据后进程就关闭了
44.         CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0])).join();
45.     }

复制代码

完整测试类：

1. package cn.skyjilygao.demo;
3. import lombok.AllArgsConstructor;
4. import lombok.Data;
5. import lombok.NoArgsConstructor;
6. import lombok.extern.slf4j.Slf4j;
7. import org.junit.jupiter.api.Test;
8. import org.redisson.api.RBlockingDeque;
9. import org.redisson.api.RDelayedQueue;
10. import org.redisson.api.RedissonClient;
11. import org.springframework.beans.factory.annotation.Autowired;
12. import org.springframework.boot.test.context.SpringBootTest;
14. import java.io.IOException;
15. import java.time.LocalDateTime;
16. import java.util.ArrayList;
17. import java.util.List;
18. import java.util.Random;
19. import java.util.Set;
20. import java.util.concurrent.CompletableFuture;
21. import java.util.concurrent.ConcurrentSkipListSet;
22. import java.util.concurrent.TimeUnit;
23. import java.util.concurrent.locks.ReentrantLock;
25. import static cn.skyjilygao.util.EntityUtil.toJsonString;
27. @Slf4j
28. @SpringBootTest
29. public class CloseOrderTests {
30.     @Autowired
31.     private RedissonClient redissonClient;
32.     public static String closeKey = "order_close_test";
33.     public volatile static Set<Integer> closed = new ConcurrentSkipListSet<>();
35.     /**
36.      * 创建订单，并设置过期时间
37.      *
38.      * @throws IOException
39.      */
40.     @Test
41.     void createOrder() {
42.         RBlockingDeque<OrderInfo> blockingDeque = redissonClient.getBlockingDeque(closeKey);
43.         RDelayedQueue<OrderInfo> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
44.         int a = 100;
45.         Random random = new Random(100);
46.         for (int i = 0; i < a; i++) {
47.             int i1 = random.nextInt(1 + i) + 1;
48.             delayedQueue.offer(new OrderInfo(i + 1, "close: " + i1, LocalDateTime.now(), LocalDateTime.now().plusSeconds(i1)), i1, TimeUnit.SECONDS);
49.         }
50.     }
52.     /**
53.      * 关闭订单
54.      *
55.      * @throws IOException
56.      */
57.     @Test
58.     void closeOrder() {
59.         ReentrantLock lock = new ReentrantLock();
60.         // 5个线程
61.         int poolSize = 5;
62.         List<CompletableFuture<Void>> futureList = new ArrayList<>();
63.         for (int i = 0; i < poolSize; i++) {
64.             futureList.add(CompletableFuture.runAsync(() -> {
65.                 RBlockingDeque<OrderInfo> blockingDeque = redissonClient.getBlockingDeque(closeKey);
66.                 // 加入监听
67.                 redissonClient.getDelayedQueue(blockingDeque);
68.                 while (true) {
69.                     OrderInfo take;
70.                     try {
71.                         take = blockingDeque.take();
72.                     } catch (Exception e) {
73.                         continue;
74.                     }
75.                     if (take == null) {
76.                         continue;
77.                     }
78.                     try {
79.                         lock.lock();
80.                         if(closed.contains(take.getId())){
81.                             log.info("测试是否会抢占：已存在其他线程处理关闭订单[{}]", take.getId());
82.                         }
83.                         closed.add(take.getId());
84.                     }finally {
85.                         lock.unlock();
86.                     }
87.                     log.info("订单[{}]关闭中。。。", take.getId());
88.                     log.info("订单[{}]已关闭！order={}", take.getId(), toJsonString(take));
89.                 }
90.             }));
91.         }
92.         // 模拟正式环境中进程一直在运行，因为test时，没有join则会只执行一次出现消费完数据后进程就关闭了
93.         CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0])).join();
94.     }
97.     @Data
98.     @AllArgsConstructor
99.     @NoArgsConstructor
100.     public class OrderInfo {
101.         private Integer id;
102.         private String description;
103.         private LocalDateTime createTime;
104.         private LocalDateTime expireTime;
105.     }
106. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！