day41 Redis三种消息队列与Stream实现异步秒杀

一、消息队列概述

1.1 什么是消息队列

消息队列（Message Queue）是一种存储和管理消息的中间件，也被称为消息代理（Message Broker）。最简单的消息队列模型包含三个角色：

生产者：发送消息到消息队列
消息队列：存储和管理消息
消费者：从消息队列获取消息并处理

1	生产者 → 消息队列 → 消费者

1.2 为什么引入消息队列

在秒杀场景中引入消息队列主要解决两个问题：

异步响应，提高速度：下单请求先经过 Redis 快速校验，将订单信息放入消息队列后立即返回，用户无需等待数据库操作完成
削峰填谷：高峰时期大量请求先进入消息队列，后端按自身处理能力匀速消耗，避免数据库被瞬时流量压垮

1.3 消息队列的基本要求

一个可靠的消息队列需要满足以下基本要求：

要求	说明
独立于 JVM	存储在 JVM 以外，不受 JVM 内存限制
数据持久化	对数据进行持久化，保证数据安全，服务重启不丢失
消息有序性	保证消息按发送顺序被消费
消息确认机制	消费者需要对消息进行确认，确保消息至少被消费一次

主流的专业消息中间件有 RabbitMQ、Kafka 等。在 Redis 中也可以实现消息队列，适合简单场景或已有 Redis 基础设施的情况。

二、Redis 三种消息队列实现

Redis 提供了三种实现消息队列的方式，各有优缺点，实际开发中需根据场景选择。

2.1 基于 List 实现

Redis 的 List 是双向链表，可以通过 LPUSH + BRPOP 或 RPUSH + BLPOP 模拟队列效果。

核心命令：

LPUSH key value：从左侧推入消息
BRPOP key timeout：从右侧阻塞弹出消息，队列为空时阻塞等待

优点：

存储在 Redis 中，不受 JVM 内存限制
基于 Redis 持久化机制，数据有安全保证
List 本身有序，可以满足消息有序性

缺点：

无法避免消息丢失：消息被弹出后，如果消费者处理异常，消息无法找回，没有确认机制
只支持单消费者：一条消息只能被一个消费者获取

2.2 基于 PubSub 实现

PubSub（发布订阅）是 Redis 2.0 引入的消息传递模型。生产者向频道发送消息，所有订阅该频道的消费者都能收到。

核心命令：

命令	作用
`PUBLISH channel message`	向指定频道发送消息
`SUBSCRIBE channel [channel ...]`	订阅一个或多个频道
`PSUBSCRIBE pattern [pattern ...]`	订阅匹配给定模式的所有频道

优点：

采用发布订阅模型，支持多生产、多消费

缺点：

不支持持久化：消息是瞬时的，无人订阅时直接丢失
无法避免消息丢失：消费者断开后重连，期间的消息会丢失
消息堆积有上限：消费者的消息缓冲区有上限，超出时数据丢失

2.3 基于 Stream 实现（XREAD 方式）

Stream 是 Redis 5.0 引入的新数据类型，可以实现功能较为完善的消息队列。

发送消息：

1	XADD key ID field value [field value ...]

key：队列名称
ID：消息唯一标识，* 表示自动生成（时间戳-序号）
field value：消息内容，以键值对形式存储

示例：

1	XADD stream.orders * id 1001 userId 1 voucherId 10

读取消息（XREAD）：

1	XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]

COUNT count：本次读取的最大消息数量
BLOCK milliseconds：阻塞读取的超时时间，0 表示无限等待
ID：起始消息 ID，$ 表示从最新的消息开始

XREAD 方式的特点：

特点	说明
消息可回溯	消息永久保存在 Stream 中，可以重新读取历史消息
可阻塞读取	支持 BLOCK 参数，队列为空时阻塞等待
支持多消费者	多个消费者可以同时读取同一条消息
有漏读风险	使用 `$` 读取最新消息时，如果处理一条消息过程中有新消息到达，下次读取仍从最新开始，中间的消息会被漏掉

漏读问题是 XREAD 方式的主要缺陷。例如：消费者正在处理消息 A，此时队列中来了消息 B 和 C，处理完 A 后用 $ 读取，只能读到最新的 C，中间的 B 就被漏掉了。

2.4 三种方式对比

对比项	List	PubSub	Stream (XREAD)
持久化	支持	不支持	支持
消息有序	支持	不保证	支持
消息回溯	不支持	不支持	支持
多消费者	不支持	支持	支持
消息确认	不支持	不支持	不支持（XREAD 方式）
消息丢失	弹出即丢	无订阅即丢	不会丢失，但可能漏读

实际开发中，List 和 PubSub 缺点过多，一般不考虑作为消息队列使用。简单业务场景可以使用 Redis Stream 的消费者组模式。

三、Stream 消费者组

消费者组（Consumer Group）是 Stream 提供的高级特性，将多个消费者划分到同一组中，共同消费同一个队列。

3.1 消费者组的核心概念

关键特点：

消息分流：同一个消费者组内的多个消费者共同消费队列中的消息，一条消息只会被组内一个消费者处理，从而提高消费速度
多组多消费：可以创建多个消费者组，不同组之间互不影响，同一条消息可以被不同组各消费一次
消息确认（ACK）：消费者读取消息后，消息进入 pending-list（待确认列表），只有消费者手动确认后，消息才从 pending-list 移除
消息不丢失：如果消费者处理异常，消息仍保留在 pending-list 中，可以重新读取处理，保证消息至少被消费一次
无漏读：使用 > 作为起始 ID 时，从下一个未消费的消息开始读取，不会漏掉中间的消息

3.2 常用命令

创建消费者组：

1	XGROUP CREATE key groupName ID [MKSTREAM]

参数	说明
`key`	队列名称
`groupName`	消费者组名称
`ID`	起始 ID 标识，`$` 代表从队列最后一个消息之后开始，`0` 代表从第一个消息开始
`MKSTREAM`	可选，队列不存在时自动创建队列

其他管理命令：

# 删除指定的消费者组
XGROUP DESTROY key groupName

# 给指定的消费者组添加消费者
XGROUP CREATECONSUMER key groupname consumername

# 删除消费者组中的指定消费者
XGROUP DELCONSUMER key groupname consumername

从消费者组读取消息：

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

参数	说明
`group`	消费者组名称
`consumer`	消费者名称，不存在时会自动创建
`COUNT count`	本次读取的最大消息数量
`BLOCK milliseconds`	没有消息时的最长阻塞等待时间
`NOACK`	无需手动 ACK，获取消息后自动确认（不推荐，有丢失风险）
`STREAMS key`	指定队列名称
`ID`	起始消息 ID

ID 的两种取值：

>：从下一个未消费的消息开始读取（正常消费时使用）
其他值（如 0）：从 pending-list 中获取已消费但未确认的消息（异常恢复时使用）

确认消息：

1	XACK key groupName ID [ID ...]

将指定消息从 pending-list 中移除，表示已成功处理。

3.3 消费者组解决的问题

消费者组相比 XREAD 独立消费方式，解决了以下问题：

漏读问题：使用 > 从下一个未消费的消息开始，每条消息都不会被跳过
消息确认机制：未确认的消息保存在 pending-list 中，消费者异常后可以重新处理，确保消息至少被消费一次
消费速度提升：同一个消费者组内可以有多个消费者并行消费，加快消息处理速度
多消费模式：设置多个消费者组，不同组独立消费，实现一条消息被多种业务处理

四、基于 Stream 的异步秒杀实现

本项目使用 Redis Stream 消费者组模式实现异步秒杀下单，替代之前的 JVM 阻塞队列方案。

4.1 需求与思路

需求：

创建一个 Stream 类型的消息队列，名为 stream.orders
修改秒杀 Lua 脚本，在判断有抢购资格后，直接向 stream.orders 中添加消息
项目启动时开启一个异步线程，持续从 Stream 队列中获取消息，完成数据库下单

整体流程：

前端发起秒杀请求
    ↓
Controller 接收请求，调用 Service
    ↓
Service 执行 Lua 脚本（Redis 中完成库存校验、一人一单判断、减库存、发消息）
    ↓
Lua 脚本返回结果：有资格则将订单信息写入 Stream 队列
    ↓
Service 直接返回订单 ID 给前端（异步响应）
    ↓
后台独立线程从 Stream 消费者组读取消息
    ↓
调用 createVoucherOrder 完成数据库下单（扣库存+写订单）
    ↓
确认消息（XACK），从 pending-list 移除

4.2 Lua 脚本修改

在原有秒杀 Lua 脚本基础上，增加向 Stream 队列发送消息的步骤：

-- 优惠券id
local voucherId = ARGV[1]
-- 用户id
local userId = ARGV[2]
-- 订单id
local orderId = ARGV[3]

-- 库存key
local stockKey = 'seckill:stock:' .. voucherId
-- 订单key
local orderKey = 'seckill:order:' .. voucherId

-- 1. 查库存，如果库存不足则返回1
if(tonumber(redis.call("get", stockKey)) < 1) then
    return 1
end

-- 2. 一人一单，如果用户已下单则返回2
if(redis.call("SISMEMBER", orderKey, userId) == 1) then
    return 2
end

-- 3. 减库存
redis.call("DECR", stockKey)

-- 4. 下单（记录用户已购买）
redis.call("SADD", orderKey, userId)

-- 5. 将订单信息添加到消息队列
redis.call("xadd", "stream.orders", "*", "id", orderId, "userId", userId, "voucherId", voucherId)

-- 返回0表示成功
return 0

脚本中使用 xadd 命令向 stream.orders 队列发送消息，消息内容包含订单 id、用户 id 和优惠券 id，使用 * 让 Redis 自动生成消息 ID。

4.3 Java 代码实现

在 VoucherOrderServiceImpl 中实现 Stream 消息的消费逻辑。

核心成员变量与初始化：

private final String queueName = "stream.orders";

// 单线程线程池，用于异步消费消息
private static final ExecutorService executorService = Executors.newSingleThreadExecutor();

// Bean 初始化后启动消费线程
@PostConstruct
private void init() {
    executorService.submit(new VoucherOrderHandler());
}

消息消费处理器：

private class VoucherOrderHandler implements Runnable {
    @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> value = record.getValue();
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                // 处理订单（数据库操作）
                iVoucherOrderService.createVoucherOrder(voucherOrder);
                // 确认消息
                stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
            } catch (Exception e) {
                log.error("从消息队列中获取订单异常", e);
                // 异常时处理 pending-list 中的消息
                handlePendingList();
            }
        }
    }
}

pending-list 异常消息处理：

private void handlePendingList() {
    while (true) {
        try {
            // 从 pending-list 中获取未确认的消息
            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; // 没有异常消息，结束循环
            }
            MapRecord<String, Object, Object> record = list.get(0);
            Map<Object, Object> value = record.getValue();
            VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
            // 处理订单
            iVoucherOrderService.createVoucherOrder(voucherOrder);
            // 确认消息
            stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
        } catch (Exception e) {
            log.error("从pending-list中获取订单异常", e);
            try {
                Thread.sleep(20); // 短暂休眠，避免死循环占满 CPU
            } catch (InterruptedException ex) {
                ex.printStackTrace();
            }
        }
    }
}

秒杀入口方法：

@Override
public Result seckillVoucher(Long voucherId) {
    // 订单 id
    long orderId = redisIdWorker.nextId("order");
    // 获取用户 id
    Long userId = UserHolder.getUser().getId();
    // 执行 Lua 脚本
    Long result = stringRedisTemplate.execute(
            SECKILL_SCRIPT,
            Collections.emptyList(),
            voucherId.toString(), userId.toString(), String.valueOf(orderId)
    );
    int r = result.intValue();
    if (r != 0) {
        return Result.fail(r == 1 ? "库存不足!" : "不允许重复下单!");
    }
    // 获取代理对象（用于事务生效）
    iVoucherOrderService = (IVoucherOrderService) AopContext.currentProxy();
    // 返回订单 id
    return Result.ok(orderId);
}

创建订单方法（数据库操作）：

@Transactional
public void createVoucherOrder(VoucherOrder voucherOrder) {
    Long voucherId = voucherOrder.getVoucherId();

    // 更新库存
    boolean success = seckillVoucherService.update()
            .setSql("stock=stock-1")
            .eq("voucher_id", voucherId)
            .gt("stock", 0)
            .update();
    if (!success) {
        log.error("库存不足");
        return;
    }

    // 保存订单
    save(voucherOrder);
}

4.4 执行流程说明

请求进入：用户发起秒杀请求，seckillVoucher 方法生成订单 ID，执行 Lua 脚本
Redis 校验：Lua 脚本在 Redis 中原子地完成库存检查、一人一单判断、扣减库存、记录购买用户、发送消息到 Stream 队列
立即返回：Lua 脚本执行成功后，直接将订单 ID 返回给前端，用户无需等待数据库操作
异步消费：VoucherOrderHandler 线程持续从 stream.orders 的消费者组 g1 中读取消息
数据库下单：将消息解析为 VoucherOrder 对象，调用 createVoucherOrder 完成数据库扣库存和写订单
消息确认：下单成功后调用 acknowledge 确认消息，将其从 pending-list 移除
异常处理：如果消费过程中发生异常，进入 handlePendingList 重新处理 pending-list 中的消息，确保不丢失

注意：使用前需要确保 Stream 队列和消费者组已创建。可以通过 XGROUP CREATE stream.orders g1 0 MKSTREAM 命令手动创建，或在代码中做初始化判断。

今日总结

模块	核心要点
消息队列基础	解耦、异步、削峰；四大基本要求：独立存储、持久化、有序性、确认机制
List 实现	`LPUSH + BRPOP` 模拟队列；缺点是无消息确认、只支持单消费者
PubSub 实现	发布订阅模式，支持多消费者；缺点是无持久化、消息易丢失、堆积有上限
Stream XREAD	消息可回溯、可阻塞读取、支持多消费者；但有漏读问题（$ 模式）
Stream 消费者组	解决漏读问题、提供 ACK 确认机制、支持组内多消费者并行、多组多消费
异步秒杀实现	Lua 脚本发消息到 Stream，后台线程消费，pending-list 保证消息不丢失
选型建议	List 和 PubSub 不推荐；简单场景用 Stream 消费者组；复杂场景用专业 MQ