public void run() { while (true){ try { //获取消息队列中的消息，对应的命令XREADGROUP GROUP g1 '消费者名称' COUNT 1 BLOCK 2000 STREAMS stream.order List<MapRecord<String, Object, Object>> r1 = stringRedisTemplate.opsForStream().read( Consumer.from("group1", "c1"), StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)), StreamOffset.create("stream.order", ReadOffset.lastConsumed()) ); //判别消息是否获取成功了 if(r1 == null || r1.isEmpty()){ //如果获取失败了，继续下次的循环 continue; } //获取消息成功了，开始下一步操作 //取出第一条消息 MapRecord<String, Object, Object> objectRecord = r1.get(0); Map<Object, Object> value = objectRecord.getValue(); tb_vouther_order tbVoutherOrder = BeanUtil.fillBeanWithMap(value, new tb_vouther_order(), true); //消息的确认（保证了消息的可靠性） Long g1 = stringRedisTemplate.opsForStream().acknowledge("stream.order", "g1", objectRecord.getId()); //通过开启的线程任务，异步的操控mysql数据库完成库存的扣减和下单的操作 tbvoutherorderservice.addseckillVouther2(tbVoutherOrder); }catch (Exception e){ //处理因为异常导致消费未确认的消息 while(true){ try { //获取消息队列中的消息，对应的命令XREADGROUP GROUP g1 '消费者名称' COUNT 1 BLOCK 2000 STREAMS stream.order List<MapRecord<String, Object, Object>> r2 = stringRedisTemplate.opsForStream().read( Consumer.from("group1", "c1"), StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)), StreamOffset.create("stream.order", ReadOffset.from("0")) ); //判别消息是否获取成功了 if(r2 == null || r2.isEmpty()){ //如果获取失败了，证明pendinglist没有该消息，继续执行外围的循环 break; } //获取消息成功了，开始下一步操作 //取出第一条消息 MapRecord<String, Object, Object> objectRecord2 = r2.get(0); Map<Object, Object> value2 = objectRecord2.getValue(); tb_vouther_order tbVoutherOrder2 = BeanUtil.fillBeanWithMap(value2, new tb_vouther_order(), true); //消息的确认（保证了消息的可靠性） Long g2 = stringRedisTemplate.opsForStream().acknowledge("stream.order", "g1", objectRecord2.getId()); //通过开启的线程任务，异步的操控mysql数据库完成库存的扣减和下单的操作 tbvoutherorderservice.addseckillVouther2(tbVoutherOrder2); }catch (Exception e2){ throw new RuntimeException("该消息不存在，请人工介入"); } } throw new RuntimeException(e); } } }这个线程异步的读取redis队列的订单完成订单的构建，我记得需要互斥分布式互斥锁保证线程的安全，这里面我没看见，你分析一下，因为这个代码早期版本内有//创建锁对象,redisson对象创建锁对象时，自动生成一个唯一值作为value，后续释放锁redisson自动通过该key"lock:order:" + tbVoutherOrder.getUid()获取value，比较锁是否被修改，才能进行释放 // RLock redisLock = redissonClient.getLock("lock:order:" + tbVoutherOrder.getUid().toString()); // // //通过redisson的获取redis的全局锁 // boolean b = redisLock.tryLock(2000, TimeUnit.MICROSECONDS);获取锁，你看看最新的基于redis队列的异步下单是不是有问题