redis 关闭服务（利用Redisson实现订单关闭）

实体类

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

@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; } }