一、首先为什么要从阻塞队列改成Redis stream，并且怎么做？

1、为什么要从阻塞队列改成 Redis Stream

1. 阻塞队列缺点

   1. 内存存储，服务重启消息丢失

   2. 无持久化、无 ACK（消息确认，保证不丢）、无重试

   3. 分布式环境下无法共享队列

   4. 高并发下容易 OOM（内存满了，服务崩溃）、不稳定

2. Redis Stream 优点

   1. 持久化，消息不丢失

   2. 消费组 + ACK 确认，异常可重试

   3. 支持分布式多实例消费

   4. 配合 Lua 实现原子秒杀，更安全高效

---

2、怎么做（改造步骤）

1. 秒杀接口改造

   1. 校验 + 扣库存 + 发消息 全部放进 Lua 脚本（原子性）

   2. 脚本执行成功直接返回，不操作数据库

2. 引入 Redis Stream

   1. 秒杀资格校验通过后，发送消息到 Stream 队列

   2. 消息包含：userId、voucherId、orderId

3. 启动后台消费线程

   1. 监听 Stream 队列

   2. 收到消息异步创建订单

   3. 处理完手动 ACK

   4. 异常消息进入 PendingList，自动重试

---

3、为什么这么做（核心目的）

1. 超高并发：Redis 抗住秒杀流量，DB 异步写入

2. 不丢消息：Stream 持久化 + ACK 机制

3. 原子安全：Lua 脚本解决超卖、重复下单

4. 分布式支持：多服务实例共享消息队列

5. 高可用：服务重启不丢失任务，支持重试

总结：用 Redis Stream 替代内存阻塞队列，实现：秒杀快、不丢单、更稳定、支持分布式、高可用

---

二、代码变更逐模块解析

一、新增：Redis Stream 异步消费线程

1.1 废弃的 BlockingQueue 方案

// 已删除private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024*1024);
private static ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();

1. 核心代码解释

ExecutorService

• Java 提供的线程池接口

• 作用：管理、复用线程，避免频繁创建 / 销毁线程

Executors.newSingleThreadExecutor()

• 创建单线程线程池

• 内部只有 1 个工作线程

• 任务会进入队列排队执行

---

2. 整行代码作用

创建一个全局唯一、单线程的线程池，专门用来异步处理秒杀订单。

---

3. 工作流程

1. 秒杀资格校验成功

2. 将 “创建订单” 任务丢入线程池

3. 线程池内唯一线程依次执行订单任务

4. 主线程直接返回，不等待数据库执行完成

1.2 新增 @PostConstruct 初始化消费者

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

1. 注解与方法解释

• @PostConstruct

  • Spring 生命周期注解

  • 项目启动、Bean 初始化完成后自动执行

• init()

  • 初始化方法

  • 项目一启动就自动运行

• SECKILL_ORDER_EXECUTOR.submit(...)

  • 向单线程线程池提交一个任务

• new VoucherOrderHandler()

  • 秒杀订单异步消费任务

  • 专门监听队列、创建订单

---

2. 整体作用

项目启动时，自动开启一个后台线程，持续监听消息队列，异步处理秒杀订单。

---

3. 执行流程

1. 项目启动 → Bean 初始化

2. 自动执行 init()

3. 把 VoucherOrderHandler 提交给线程池

4. 线程池开始永久后台运行，循环处理秒杀订单

5. 不影响主线程，不阻塞接口

1.3 新增 VoucherOrderHandler 消费者任务

private class VoucherOrderHandler implements Runnable{
    String queueName = "stream.orders";
    @Override
    public void run() {
        while (true) {
            try {
                // 1. 读取Redis Stream消息
                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())
                );
                // 2. 无消息则继续循环
                if(list == null || list.isEmpty()) continue;
                // 3. 解析消息
                MapRecord<String, Object, Object> record = list.get(0);
                Map<Object, Object> value = record.getValue();
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                // 4. 处理订单
                handleVoucherOrder(voucherOrder);
                // 5. ACK确认（标记消息已处理）
                stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
            } catch (Exception e) {
                log.error("处理订单异常", e);
                handlePendingList(); // 处理PendingList中的异常消息
            }
        }
    }
}

1. 类定义

• implements Runnable：表示当前类是一个可执行任务类；重写 run() 方法，里面写任务逻辑；交给线程池执行，run () 方法会自动运行。

• queueName = "stream.orders"：Redis Stream 队列名称

2. run () 核心逻辑

• while(true)：死循环，一直监听消息，永不停止

• 作用：后台持续消费秒杀订单消息

3. 读取消息

Consumer.from("g1", "c1")

作用：声明当前消费者身份

• g1：消费组名称多个服务实例可以属于同一个组，共同消费消息

• c1：消费者名称同一个组内区分不同的消费者

• 意义：让 Redis 知道是谁在消费消息，用于记录消费位置、ACK 确认

---

StreamReadOptions.empty().count(1)

作用：设置读取选项

• count(1)：每次只读取 1 条消息避免一次拿太多，保证订单处理有序、稳定

---

.block(Duration.ofSeconds(2))

作用：阻塞等待消息

• 没有消息时，阻塞等待 2 秒

• 2 秒内来消息 → 立刻返回

• 2 秒没消息 → 结束本次读取，进入下一次循环

• 好处：避免 CPU 空转、浪费资源

---

StreamOffset.create(queueName, ReadOffset.lastConsumed())

作用：从哪里开始读消息

• queueName：队列名称 stream.orders

• lastConsumed()：从 “上一次消费成功的位置” 继续读

4. 消息处理流程

1. 无消息 → 继续循环

2. 有消息 → 解析成 VoucherOrder 订单对象

3. 调用 handleVoucherOrder 异步创建订单、写数据库

4. ACK 确认：告诉 Redis 消息已处理完成

5. 异常处理

• 出错打印日志

• 调用 handlePendingList()

• 处理未 ACK 的失败消息，保证不丢单

总结：这是一个后台无限循环任务，持续监听 Redis Stream 队列，读取秒杀消息 → 异步创建订单 → ACK 确认，保证订单安全、可靠、不丢失。

1.4 新增 handlePendingList() 异常处理

private void handlePendingList() {
    while (true) {
        try {
            // 1. 读取PendingList中的消息（未ACK的异常消息）
            List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                Consumer.from("g1", "c1"),
                StreamReadOptions.empty().count(1),
                StreamOffset.create(queueName, ReadOffset.from("0"))
            );
            if(list == null || list.isEmpty()) break;
            // 2. 处理消息 + ACK
            MapRecord<String, Object, Object> record = list.get(0);
            Map<Object, Object> value = record.getValue();
            VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
            handleVoucherOrder(voucherOrder);
            stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
        } catch (Exception e) {
            log.error("处理pending-list订单异常", e);
            try { Thread.sleep(20); } catch (InterruptedException ex) { throw new RuntimeException(ex); }
        }
    }
}

1、方法作用

处理 PendingList 消息

• PendingList：没有 ACK 的失败消息

• 作用：重试处理异常订单，保证不丢单

---

2.while(true)

• 死循环

• 把 PendingList 里的消息一次性全部处理完再退出

---

3.read(...) 读取消息

stringRedisTemplate.opsForStream().read(
    Consumer.from("g1", "c1"),          // 消费者身份：g1组c1
    StreamReadOptions.empty().count(1), // 每次读1条
    StreamOffset.create(queueName, ReadOffset.from("0")) // 从第一条开始读
)

4.关键：ReadOffset.from("0")

• 读取 PendingList 中所有未 ACK 的历史消息

• 从第一条失败消息开始重试

---

5.解析消息

• 把 Redis 消息转成 VoucherOrder 订单对象

• 调用 handleVoucherOrder 重新创建订单

---

6.acknowledge(...)

• 处理成功 → ACK 确认

• 标记消息完成，移出 PendingList

总结：处理之前消费失败、未 ACK 的订单消息，循环重试，确保秒杀订单绝不丢失。

---

二、核心业务改造：seckillVoucher 方法

2.1 改造前（同步秒杀）

@Override
public Result seckillVoucher(Long voucherId) {
    // 1. 查优惠券、校验库存
    // 2. 扣库存（update set stock = stock-1 where id=? and stock>0）
    // 3. 生成订单
    // 4. 返回订单ID
}

问题：

• 高并发下，扣库存和生成订单的逻辑同步执行，数据库压力极大。

• 校验、扣减、下单全在 Java 代码中执行，存在原子性问题和性能瓶颈。

2.2 改造后（Redis+Lua 预校验 + 异步消息）

@Override
public Result seckillVoucher(Long voucherId) {
    // 1. 获取用户和订单ID
    Long userId = UserHolder.getUser().getId();
    long orderId = redisIdWorker.nextId("order");
    // 2. 执行Lua脚本（原子性预校验：库存+重复下单+发送消息）
    Long result = stringRedisTemplate.execute(
        SECKILL_SCRIRT,
        Collections.emptyList(),
        voucherId.toString(), userId.toString(), String.valueOf(orderId)
    );
    int r = result.intValue();
    // 3. 校验结果
    if(r != 0) {
        return Result.fail(r == 1 ? "库存不足":"不能重复下单");
    }
    // 4. 返回订单ID（直接返回，无需等待DB下单）
    return Result.ok(orderId);
}

关键变更点：

• 核心逻辑全部下沉到 Lua 脚本中，Redis 单线程原子执行，解决超卖、重复下单问题。

• 直接返回订单 ID，前端无需等待数据库下单，秒杀响应从几百 ms 降至几 ms。

• 订单生成逻辑由 Stream 消费者异步执行，与主流程完全解耦。

---

1. Lua 脚本变更：新增 Stream 消息发送

关键变更点：

• 新增 orderId 参数，用于生成订单消息。

• 新增 xadd 命令，将秒杀订单信息发送到 Redis Stream 队列，供消费者异步处理。

• 脚本执行成功即表示秒杀资格已锁定，后续 DB 下单由消费者异步完成。

---

三、改造的核心技术亮点

1. Redis+Lua 原子性校验：库存扣减、重复下单校验、消息发送全在 Lua 中完成，单线程执行无并发问题。

2. Redis Stream 异步消息队列：

   1. 持久化存储消息，进程重启不丢失。

   2. 消费组 + ACK 机制，支持异常重试，消息不丢。

   3. 削峰填谷，将瞬时高并发转为平稳的数据库写入。

3. 前后端体验优化：秒杀接口直接返回订单 ID，前端无等待，响应极快。

4. 可靠性保障：PendingList 重试机制，处理失败的订单不会丢失，最终一致性得到保障。