颠覆认知！高并发场景下订单也能异步处理？揭秘新策略

发送kafka

createOrderByMq 中的流程是发送kafka

private String createOrderByMq(OrderCreateDto orderCreateDto,List<SeatVo> purchaseSeatList){
    CreateOrderMqDomain createOrderMqDomain = new CreateOrderMqDomain();
    CountDownLatch latch = new CountDownLatch(1);
    //发送kafka
    createOrderSend.sendMessage(JSON.toJSONString(orderCreateDto),sendResult -> {
        //发送成功
        createOrderMqDomain.orderNumber = String.valueOf(orderCreateDto.getOrderNumber());
        assert sendResult != null;
        log.info("创建订单kafka发送消息成功 topic : {}",sendResult.getRecordMetadata().topic());
        latch.countDown();
    },ex -> {
        //发送失败
        log.error("创建订单kafka发送消息失败 error",ex);
        log.error("创建订单失败 需人工处理 orderCreateDto : {}",JSON.toJSONString(orderCreateDto));
        updateProgramCacheDataResolution(orderCreateDto.getProgramId(),purchaseSeatList,OrderStatus.CANCEL);
        createOrderMqDomain.daMaiFrameException = new DaMaiFrameException(ex);
        latch.countDown();
    });
    try {
        //使用CountDownLatch等待发送结果
        latch.await();
    } catch (InterruptedException e) {
        log.error("createOrderByMq InterruptedException",e);
        throw new DaMaiFrameException(e);
    }
    //如果发送失败，则直接抛出异常
    if (Objects.nonNull(createOrderMqDomain.daMaiFrameException)) {
        throw createOrderMqDomain.daMaiFrameException;
    }
    return createOrderMqDomain.orderNumber;
}

在调用发送kafka的api时，使用了带有回调方式，当发送成功或者发送失败后，可以进行回调，但有一点要注意，此回调是异步执行的！所以我们如果想要知道发送后到底是成功还是失败，需要进行等待

那么如果才能等待呢？这里我们使用了jdk中并发包下的工具 CountDownLatch，俗称门栓，当发送给kafka后，会直接到 latch.await()，由于没有门栓没有释放，所以这里会阻塞

当发送后进行了回调后，无论是成功还是失败，都会执行 latch.countDown()，此方法是进行门栓的释放，当释放后，上一步的阻塞就可以被唤醒，然后继续执行流程

执行到这里后，就可以借助 CreateOrderMqDomain对象，判断到底是发送成功了还是失败了，如果存在了异常，说明确实发送失败了，则直接抛出异常。如果没有异常说明发送成功了，则把订单编号返回

然后还是通过延迟队列发送订单关闭的消息，此流程在之前文章已经详细的讲解过，这里就不再赘述

消费消息队列的订单消息

当订单成功发送给kafka后，就会开始执行消息的消费流程

com.damai.service.kafka.CreateOrderConsumer

@Slf4j
@AllArgsConstructor
@Component
public class CreateOrderConsumer {
    
    @Autowired
    private OrderService orderService;
    
    public static Long MESSAGE_DELAY_TIME = 5000L;
    
    @KafkaListener(topics = {SPRING_INJECT_PREFIX_DISTINCTION_NAME+"-"+"${spring.kafka.topic:create_order}"})
    public void consumerOrderMessage(ConsumerRecord<String,String> consumerRecord){
        try {
            Optional.ofNullable(consumerRecord.value()).map(String::valueOf).ifPresent(value -> {
                
                OrderCreateDto orderCreateDto = JSON.parseObject(value, OrderCreateDto.class);
                
                long createOrderTimeTimestamp = orderCreateDto.getCreateOrderTime().getTime();
                
                long currentTimeTimestamp = System.currentTimeMillis();
                
                long delayTime = currentTimeTimestamp - createOrderTimeTimestamp;
                
                log.info("消费到kafka的创建订单消息 消息体: {} 延迟时间 : {} 毫秒",value,delayTime);
                //如果消费到消息时，延迟时间超过了5s，那么此订单丢弃，将库存回滚回去
                if (currentTimeTimestamp - createOrderTimeTimestamp > MESSAGE_DELAY_TIME) {
                    log.info("消费到kafka的创建订单消息延迟时间大于了 {} 毫秒 此订单消息被丢弃 订单号 : {}",
                            delayTime,orderCreateDto.getOrderNumber());
                    Map<Long, List<OrderTicketUserCreateDto>> orderTicketUserSeatList =
                            orderCreateDto.getOrderTicketUserCreateDtoList().stream().collect(Collectors.groupingBy(OrderTicketUserCreateDto::getTicketCategoryId));
                    Map<Long,List<Long>> seatMap = new HashMap<>(orderTicketUserSeatList.size());
                    orderTicketUserSeatList.forEach((k,v) -> {
                        seatMap.put(k,v.stream().map(OrderTicketUserCreateDto::getSeatId).collect(Collectors.toList()));
                    });
                    //数据恢复
                    orderService.updateProgramRelatedDataMq(orderCreateDto.getProgramId(),seatMap, OrderStatus.CANCEL);
                }else {
                    String orderNumber = orderService.createMq(orderCreateDto);
                    log.info("消费到kafka的创建订单消息 创建订单成功 订单号 : {}",orderNumber);
                }
            });
        }catch (Exception e) {
            log.error("处理消费到kafka的创建订单消息失败 error",e);
        }
    }
}

当消息到消息时，要判断延时时间是否大于了5s，如果大于了，此订单就要被丢弃，丢弃此订单也要将数据进行回滚，因为之前在节目服务中通过redis+lua进行了扣减余票数量，座位的状态锁定操作，所以要把这些数据要恢复回去，