Redis 独占锁、分段锁、发布/订阅，自动注入Spring容器高级编码分享！

redis:
  sdk:
    config:
      host: localhost
      port: 6379
      password: 123456
      pool-size: 10
      min-idle-size: 5
      idle-timeout: 30000
      connect-timeout: 5000
      retry-attempts: 3
      retry-interval: 1000
      ping-interval: 60000
      keep-alive: true

// 设置到缓存，在创建订单完成后写入缓存
redissonService.setValue(orderId, orderEntity);

@Override
public OrderEntity queryOrder(String orderId) {
    OrderEntity orderEntity = redissonService.getValue(orderId);
    if (null == orderEntity) {
        UserOrderPO userOrderPO = userOrderDao.selectByOrderId(orderId);
        orderEntity = new OrderEntity();
        orderEntity.setUserName(userOrderPO.getUserName());
        orderEntity.setUserId(userOrderPO.getUserId());
        // 设置到缓存
        redissonService.setValue(orderId, orderEntity);
    }
    return orderEntity;
}

/** 独占锁 */
@Override
public String createOrderByLock(OrderAggregate orderAggregate) {
    RLock lock = redissonService.getLock("create_order_lock_".concat(orderAggregate.getSkuEntity().getSku()));
    try {
        lock.lock();
        long decrCount = redissonService.decr(orderAggregate.getSkuEntity().getSku());
        if (decrCount < 0) return "已无库存[初始化的库存和使用库存，保持一致。orderService.initSkuCount(\"13811216\", 10000);]";
        return createOrder(orderAggregate);
    } finally {
        lock.unlock();
    }
}

/** 分段锁，也可以称为无锁化 */
@Override
public String createOrderByNoLock(OrderAggregate orderAggregate) {
    UserEntity userEntity = orderAggregate.getUserEntity();
    SKUEntity skuEntity = orderAggregate.getSkuEntity();
    // 模拟锁商品库存
    long decrCount = redissonService.decr(skuEntity.getSku());
    if (decrCount < 0) return "已无库存[初始化的库存和使用库存，保持一致。orderService.initSkuCount(\"13811216\", 10000);]";
    String lockKey = userEntity.getUserId().concat("_").concat(String.valueOf(decrCount));
    RLock lock = redissonService.getLock(lockKey);
    try {
        lock.lock();
        return createOrder(orderAggregate);
    } finally {
        lock.unlock();
    }
}

// 创建Redisson客户端
RedissonClient redisson = Redisson.create();

// 获取RTopic对象
RTopic<String> topic = redisson.getTopic("myTopic");

// 发布消息
topic.publish("Hello, Redisson!");

// 添加监听器
topic.addListener(String.class, (channel, msg) -> {
    System.out.println("Received message: " + msg);
});

// 关闭Redisson客户端
redisson.shutdown();

@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.TYPE})
@Documented
public @interface RedisTopic {

    String topic() default "";

}

@Slf4j
@Service
@RedisTopic(topic = "testRedisTopic02")
public class RedisTopicListener02 implements MessageListener<String> {

    @Override
    public void onMessage(CharSequence channel, String msg) {
        log.info("02-监听消息(Redis 发布/订阅): {}", msg);
    }

}

// 添加监听
String[] beanNamesForType = applicationContext.getBeanNamesForType(MessageListener.class);
for (String beanName : beanNamesForType) {
    MessageListener bean = applicationContext.getBean(beanName, MessageListener.class);
    Class<?> beanClass = bean.getClass();
    if (beanClass.isAnnotationPresent(RedisTopic.class)) {
        RedisTopic redisTopic = beanClass.getAnnotation(RedisTopic.class);
        
        RTopic topic = redissonClient.getTopic(redisTopic.topic());
        topic.addListener(String.class, bean);
        
        // 动态创建 bean 对象，注入到 spring 容器，bean 的名称为 redisTopic，对象为 RTopic
        ConfigurableListableBeanFactory beanFactory = applicationContext.getBeanFactory();
        beanFactory.registerSingleton(redisTopic.topic(), topic);
    }
}

@Slf4j
@Repository
public class OrderRepository implements IOrderRepository {

    @Resource
    private IRedisService redissonService;
    @Resource
    private IUserOrderDao userOrderDao;
    
    @Resource
    private RTopic testRedisTopic;

    @Resource(name = "testRedisTopic02")
    private RTopic testRedisTopic02;

    @Resource(name = "testRedisTopic03")
    private RTopic testRedisTopic03;


    @Override
    public String createOrder(OrderAggregate orderAggregate) {
    
        // 省略...
      
        testRedisTopic02.publish(JSON.toJSONString(orderEntity));
        testRedisTopic03.publish(JSON.toJSONString(orderEntity));

        return orderId;
    }    
}    

@Test
public void test_createOrder() throws InterruptedException {
    String sku = RandomStringUtils.randomNumeric(9);
    int count = 10000;
    orderService.initSkuCount(sku, count);
  
    for (int i = 0; i < count; i++) {
        threadPoolExecutor.execute(() -> {
            UserEntity userEntity = UserEntity.builder()
                    .userId("小傅哥")
                    .userName("xfg".concat(RandomStringUtils.randomNumeric(3)))
                    .userMobile("+86 13521408***")
                    .build();
            SKUEntity skuEntity = SKUEntity.builder()
                    .sku(sku)
                    .skuName("《手写MyBatis：渐进式源码实践》")
                    .quantity(1)
                    .unitPrice(BigDecimal.valueOf(128))
                    .discountAmount(BigDecimal.valueOf(50))
                    .tax(BigDecimal.ZERO)
                    .totalAmount(BigDecimal.valueOf(78))
                    .build();
            DeviceVO deviceVO = DeviceVO.builder()
                    .ipv4("127.0.0.1")
                    .ipv6("2001:0db8:85a3:0000:0000:8a2e:0370:7334".getBytes())
                    .machine("IPhone 14 Pro")
                    .location("shanghai")
                    .build();
            long threadBeginTime = System.currentTimeMillis(); // 记录线程开始时间
            // 耗时:4毫秒
               String orderId = orderService.createOrder(new OrderAggregate(userEntity, skuEntity, deviceVO));
            // 耗时:106毫秒
              String orderId = orderService.createOrderByLock(new OrderAggregate(userEntity, skuEntity, deviceVO));
            // 耗时:4毫秒
            String orderId = orderService.createOrderByNoLock(new OrderAggregate(userEntity, skuEntity, deviceVO));
            long took = System.currentTimeMillis() - threadBeginTime;
            totalExecutionTime.addAndGet(took); // 累加线程耗时
            log.info("写入完成 {} 耗时 {} (ms)", orderId, took / 1000);
        });
    }
    new Thread(() -> {
        while (true) {
            if (threadPoolExecutor.getActiveCount() == 0) {
                log.info("执行完毕，总耗时：{} (ms)", (totalExecutionTime.get() / 1000));
                  log.info("执行完毕，总耗时:{}", "\r\033[31m" + (totalExecutionTime.get() / 1000) + "\033[0m");
                break;
            }
            try {
                Thread.sleep(350);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }).start();
    // 等待
    new CountDownLatch(1).await();
}

/**
 * 延迟队列场景应用；https://mp.weixin.qq.com/s/jJ0vxdeKXHiYZLrwDEBOcQ
 */
@Test
public void test_getDelayedQueue() throws InterruptedException {
    RBlockingQueue<Object> blockingQueue = redissonService.getBlockingQueue("xfg-dev-tech-task");
    RDelayedQueue<Object> delayedQueue = redissonService.getDelayedQueue(blockingQueue);
    new Thread(() -> {
        try {
            while (true){
                Object take = blockingQueue.take();
                log.info("测试结果 {}", take);
                Thread.sleep(10);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }).start();
    int i = 0;
    while (true){
        delayedQueue.offerAsync("测试" + ++i, 100L, TimeUnit.MILLISECONDS);
        Thread.sleep(1000L);
    }
}