item2-10 (秒杀优化2 - 消息队列)

XGROUP CREATE key groupName ID [MKSTREAM]

XGROUP DESTORY key groupName

XGROUP CREATECONSUMER key groupname consumername

XGROUP DELCONSUMER key groupname consumername

XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...]

XGROUP CREATE stream.orders g1 0 MKSTREAM

@Override
public Result seckillVoucher(Long voucherId) {
  Long userId = UserHolder.getUser().getId();
  long orderId = redisIdWorker.nextId("order");
  // 执行lua脚本
  // 变成 3 个参数
  // 这样就要先创建订单id
  // lua 脚本执行完 已经将订单信息放入 stream.orders 消息队列中
  Long result = stringRedisTemplate.execute(
          SECKILL_SCRIPT,
          Collections.emptyList(),
          voucherId.toString(), userId.toString(), String.valueOf(orderId));

  int r = result.intValue();
  // 判断结果是否为0
  if (r != 0) {
      // 不为0 ，代表没有购买资格
      return Result.fail(r == 1 ? "库存不足" : "不能重复下单");
  }
  
  // 获取代理对象
  proxy = (IVoucherOrderService) AopContext.currentProxy();

  // 返回订单id
  return Result.ok(orderId);
}

@PostConstruct
private void init() {
  SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
}


//异步处理线程池
private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();

private class VoucherOrderHandler implements Runnable {
  String queueName = "stream.orders";
  @Override
  public void run() {
      while(true){
          // 获取消息队列订单信息
          try {
              // 获取消息队列订单信息
              List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                      Consumer.from("g1", "c1"),
                      StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                      StreamOffset.create(queueName, ReadOffset.lastConsumed())
              );

              // 判断消息获取是否成功

              // 不存在 没有消息 继续下一次尝试获取
              if(list == null || list.isEmpty())
                  continue;

              // 存在 可以下单
              MapRecord<String, Object, Object> record = list.get(0);
              Map<Object, Object> values = record.getValue();
              VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);
              // 创建订单
              handleVoucherOrder(voucherOrder);

              // ACK确认
              stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
          } catch (Exception e) {
              log.error("订单异常：", e);
              e.printStackTrace();
              //处理异常消息
              handlePendingList();
          }
      }
  }

  private void handlePendingList() {
      while (true) {
          try {
              // 1.获取pending-list中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 0
              List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                      Consumer.from("g1", "c1"),
                      StreamReadOptions.empty().count(1),
                      StreamOffset.create("stream.orders", ReadOffset.from("0"))
              );
              // 2.判断订单信息是否为空
              if (list == null || list.isEmpty()) {
                  // 如果为null，说明没有异常消息，结束循环
                  break;
              }
              // 解析数据
              MapRecord<String, Object, Object> record = list.get(0);
              Map<Object, Object> value = record.getValue();
              VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
              // 3.创建订单
              createVoucherOrder(voucherOrder);
              // 4.确认消息 XACK
              stringRedisTemplate.opsForStream().acknowledge("s1", "g1", record.getId());
          } catch (Exception e) {
              log.error("处理pendding订单异常", e);
              try{
                  Thread.sleep(20);
              }catch(Exception ec){
                  ec.printStackTrace();
              }
          }
      }
  }
}