秒杀全链路流程
整体架构
用户点击"限时抢购"
│
▼
┌──────────────────────────────────────────────┐
│ 同步部分 (主线程, <5ms 返回) │
│ │
│ ① 前端校验 → ② 后端兜底查DB加载库存 │
│ → ③ 生成全局唯一订单ID │
│ → ④ Lua脚本原子执行(库存扣减+去重+写Stream) │
│ → ⑤ 返回订单号给用户 │
└──────────────────────────────────────────────┘
│ XADD stream.orders
▼
┌──────────────────────────────────────────────┐
│ 异步部分 (单线程消费, 后台慢慢写DB) │
│ │
│ ⑥ XREADGROUP 读取Stream消息 │
│ → ⑦ Redisson分布式锁 │
│ → ⑧ @Transactional 写MySQL │
│ ├─ DB幂等校验 │
│ ├─ 乐观锁扣库存 │
│ └─ INSERT 订单 │
│ → ⑨ XACK确认消费 │
│ │
│ 失败处理: 不ACK → Pending重试 → 死循环兜底 │
└──────────────────────────────────────────────┘
阶段一: 同步响应（主线程，3~5ms）
第 0 步: 前端调用
用户在商铺详情页点"限时抢购"按钮，shop-detail.html 发请求：
axios.post("/voucher-order/seckill/" + voucherId)
请求经过两层拦截器：RefreshTokenInterceptor 从 authorization header 读取 token 加载用户到 ThreadLocal → LoginInterceptor 校验通过。
第 1 步: 兜底加载 Redis 库存
// VoucherOrderServiceImpl.seckillVoucher()
Long userId = UserHolder.getUser().getId(); // 从ThreadLocal拿当前用户
String stockKey = "seckill:stock:" + voucherId;
Boolean hasStock = stringRedisTemplate.hasKey(stockKey);
if (Boolean.FALSE.equals(hasStock)) {
// Redis里没有库存key → 从DB加载到Redis
SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
stringRedisTemplate.opsForValue().set(stockKey, voucher.getStock().toString());
}
seckill:stock:10 是一个 Redis String，存库存数量。如果这个 key 不存在（Redis 重启后），从 DB 的 tb_seckill_voucher.stock 字段恢复。
第 2 步: 生成全局唯一订单 ID
long orderId = redisIdWorker.nextId("order");
RedisIdWorker 生成一个 Snowflake 风格的 64 位 ID：
高 32 位 = 当前秒 – 基准时间戳(2022-01-01 00:00:00)
低 32 位 = Redis INCR icr:order:2026:07:04 的返回值
比如 609848294759202817 这个数字，拆开就是 时间戳部分 + 自增序列号，保证全局唯一 + 趋势递增。
第 3 步: Lua 脚本原子执行（核心）
— KEYS来自ARGV参数
local voucherId = ARGV[1] — "10"
local userId = ARGV[2] — "1010"
local orderId = ARGV[3] — "609848294759202817"
local stockKey = 'seckill:stock:' .. voucherId — seckill:stock:10
local orderKey = 'seckill:order:' .. voucherId — seckill:order:10
— ① 判库存
local stock = redis.call('get', stockKey)
if (not stock or tonumber(stock) <= 0) then
return 1 — 库存不足
end
— ② 判重复
if (redis.call('sismember', orderKey, userId) == 1) then
return 2 — 不能重复下单
end
— ③ 扣库存 + 记录用户 + 写消息队列（三步原子）
redis.call('incrby', stockKey, -1) — stock = stock – 1
redis.call('sadd', orderKey, userId) — Set里加userId防重
redis.call('xadd', 'stream.orders', '*', — Stream里写订单
'userId', userId,
'voucherId', voucherId,
'id', orderId)
return 0 — 成功
这段 15 行 Lua 是整个秒杀方案的灵魂。关键点：
操作 Redis 命令 O(1) 作用
查库存 GET ✅ 判断是否还有券
查重复 SISMEMBER ✅ Set 集合判断用户是否买过
扣库存 INCRBY ✅ 原子减 1
去重 SADD ✅ 记录已购用户
发消息 XADD ✅ 写 Stream 消息队列
五个操作在同一个 Lua 脚本里执行，Redis 单线程保证不会被其他命令插入——这就是"零锁开销的原子性"。
第 4 步: 返回结果
Long result = stringRedisTemplate.execute(getSeckillScript(),
Collections.emptyList(),
voucherId.toString(), userId.toString(), String.valueOf(orderId));
int r = result.intValue();
if (r == 1) return Result.fail("库存不足");
if (r == 2) return Result.fail("不能重复下单");
return Result.ok(orderId); // ← 用户拿到订单号，前端弹"抢购成功"
从用户点击到拿到订单号，主线程只做了 3 件事：① 兜底加载库存（极少数情况触发）② 生成 ID（1 次 Redis INCR）③ 执行 Lua（1 次 EVAL）。总共 2-3 次网络往返，<5ms 返回。完全没碰 MySQL，这就是"快"的秘诀。
阶段二: 异步消费（后台线程）
Lua 脚本的 XADD stream.orders * userId voucherId orderId 把订单信息写入了 Redis Stream。下面是怎么消费的：
第 5 步: 消费者线程启动
应用启动时 @PostConstruct 自动拉起一个单线程 Consumer：
@PostConstruct
public void init() {
SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
}
这个线程是一个 while (running) 死循环，一直在读 Stream。
第 6 步: 从 Stream 读消息
// Consumer Group: g1, Consumer: c1
List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
Consumer.from("g1", "c1"),
StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
StreamOffset.create("stream.orders", ReadOffset.lastConsumed())
);
XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.orders > → 从上次消费位置拉一条新消息。没有消息就阻塞等 2 秒，有消息立刻返回。
第 7 步: 解析消息 → 加锁 → 写 DB
VoucherOrder voucherOrder = parseVoucherOrder(values); // Map → VoucherOrder对象
if (handleVoucherOrder(voucherOrder)) {
// 成功 → ACK确认
stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
}
// 失败 → 不ACK → 消息留在Pending List → 等重试
handleVoucherOrder 内部：
RLock lock = redissonClient.getLock("lock:order:" + userId);
boolean isLock = lock.tryLock();
if (!isLock) return false; // 拿不到锁 → 不ACK，等重试
try {
IVoucherOrderService proxy = applicationContext.getBean(IVoucherOrderService.class);
proxy.createVoucherOrder(voucherOrder); // @Transactional 实际写DB
return true;
} finally {
lock.unlock();
}
第 8 步: createVoucherOrder — 写 DB 的三层防护
@Transactional(rollbackFor = Exception.class)
public void createVoucherOrder(VoucherOrder voucherOrder) {
// ① 幂等校验：DB里是否已有记录
int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
if (count > 0) return; // 已存在 → 跳过（非异常 — 重试安全）
// ② 乐观锁扣库存
boolean success = seckillVoucherService.update()
.setSql("stock = stock – 1")
.eq("voucher_id", voucherId).gt("stock", 0) // ← WHERE stock > 0
.update();
if (!success) throw new RuntimeException("DB库存扣减失败");
// ③ 保存订单
save(voucherOrder);
}
三层：幂等校验（防重复消息）+ 乐观锁（防超卖）+ 事务回滚（失败全部撤销）。
第 9 步: ACK 确认
DB 写入成功 → XACK stream.orders g1 <消息ID> → Stream 标记该消息已处理。
阶段三: 失败恢复
Pending List 重试：
消息处理失败不 ACK → Redis 标记为 Pending → handlePandingList() 定时扫描：
// XREADGROUP GROUP g1 c1 COUNT 1 STREAMS stream.orders 0
// ← "0" 表示从 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"))
);
同一批消息循环重试，直到全部 ACK 完毕才 break。
NOGROUP 兜底：Stream 消费者组不存在时，ensureConsumerGroup() 自动创建——先 XADD 一条 dummy 消息创建 Stream → 创建 Group → 删除 dummy。
🔒 五层防护总结
层 位置 方式 防什么
1 Lua if stock <= 0 → return 1 库存判断原子化
2 Lua SISMEMBER orderKey userId Redis 层去重
3 Java 异步 Redisson lock.tryLock() 分布式并发安全
4 DB UPDATE … WHERE stock > 0 乐观锁二次兜底
5 DB SELECT count(*) 幂等校验 Pending 重试安全
15 行 Lua + 200 行 Java，构建了一套"前端瞬间返回，后台异步处理，失败自动重试"的完整秒杀链路。
为什么这个项目用 Redis Stream？
1. 项目定位决定的
这个项目的目的是 Redis 综合实战。Stream 是 Redis 5.0 引入的核心数据结构，它的消费者组、Pending 重试、ACK 确认机制和 Kafka/RocketMQ 几乎一一对应。学完 Stream 再学 RocketMQ，概念直接迁移：
Redis Stream RocketMQ 概念
Stream Key Topic 消息主题
Consumer Group Consumer Group 消费者组
> 新消息 未消费消息
Pending List 重试队列 未 ACK 的消息
XACK ACK 确认消费
ReadOffset.from("0") 从头消费 重新消费
学会 Stream 等于同时学会了消息队列的核心概念，这是课程设计的初衷。
2. 不引入额外中间件
Redis 本来就要用（存库存、存缓存、存 Session），Stream 是 Redis 自带的能力。用 RocketMQ 需要额外部署一个 Broker + NameServer，增加运维复杂度。
对于这个单体项目，Redis Stream 够用——单线程消费、Pending 重试、消费者组负载均衡，秒杀场景需要的它都有。
3. 性能足够
秒杀场景的瓶颈不在消息队列的吞吐量，而在 DB 写入速度。Stream 的 XADD 是 O(1)，单机每秒可以写入几十万条消息。瓶颈是后面的单线程消费者写 MySQL——不管用 Stream 还是 RocketMQ，这个瓶颈都存在。换 RocketMQ 不会让下单更快。
那为什么简历也提 RocketMQ？
因为 生产环境我会选 RocketMQ。面试简历上列 RocketMQ 表示我知道"正确"的技术选型是什么。两者该用的场景不同：
维度 Redis Stream RocketMQ
部署成本 零（Redis 自带） 高（Broker + NameServer + Console）
可靠性 AOF/RDB 持久化，有丢失风险 同步刷盘 + 主从，几乎不丢消息
事务消息 ❌ 不支持 ✅ 半消息 + 本地事务检查
死信队列 ❌ 需手动实现 ✅ 消费失败自动进 DLQ
顺序消息 单线程天然顺序 ✅ 支持分区有序
管理控制台 ❌ 没有 ✅ 消息轨迹、堆积监控、回溯
吞吐量 单机 10w/s+ 集群百万/s
适用场景 学习、小项目、不引入新中间件 生产、金融级可靠、多系统解耦
学习成本 低（已用 Redis） 中（需额外学习）
诚实回答模板
面试官问这个问题时，这样答：
"这个项目是 Redis 实战项目，Stream 是 Redis 内置能力，不需要额外中间件，部署成本低。核心概念——Consumer Group、Pending 重试、ACK 确认——和 RocketMQ 一致，学会了 Stream 再学 RocketMQ 半天就上手。
但说实话，如果是生产环境，我会选 RocketMQ。三个原因：一是 RocketMQ 支持事务消息，可以做'半消息 → 执行本地事务 → 提交/回滚'，解决 Lua 扣库存成功但发消息失败的问题；二是有死信队列，消费失败自动进 DLQ 而不是像我现在这样手动写 Pending 循环；三是有管理控制台，消息堆积一眼就能看到。
我的理解是——学习阶段用 Redis Stream 理解消息队列的底层原理，生产环境用 RocketMQ 获得企业级的可靠性和可运维性。 原理相通，选型不同。"
Lua 脚本具体是怎么保证「库存判断 + 扣减」操作的原子性的？
Redis 为什么能保证 Lua 脚本原子性？
一句话版本：Redis 是单线程执行命令的，Lua 脚本执行期间，Redis 不会处理任何其他命令。
单线程模型
Redis 的"单线程"不是指整个 Redis 进程只有一个线程，而是指命令执行器（Command Executor）是单线程的。所有客户端的命令——不管来自多少个连接、多少个用户——都在一个队列里排队，一个一个被取出来执行。
客户端A: GET seckill:stock:10 ─┐
客户端B: SISMEMBER seckill:order:10 1010 ─┤
客户端C: INCRBY seckill:stock:10 -1 ──────┤
▼
┌────────────────────┐
│ Redis 命令队列 │
└────────────────────┘
│
Redis 单线程逐个取出执行
A执行完 → B执行完 → C执行完
问题在哪？A、B、C 之间可能有其他客户端的命令插入：
时间线（不用Lua）:
T1: 客户端A执行 GET stock → 返回 5（有库存）
T2: 客户端B执行 GET stock → 返回 5（也有库存！）
T3: 客户端A执行 DECR stock → stock = 4
T4: 客户端B执行 DECR stock → stock = 3
// 结果：库存5，卖了2次，还剩3——但实际应该只剩3次机会，超卖了！
Redis 单线程只保证每个命令本身是原子的（比如 GET 的结果不会是执行一半的数据），不保证多个命令之间不被插入。这就是"查库存"和"扣库存"之间会有 gap 的原因。
Lua 脚本做了什么
把三个操作打包进一段 Lua：
— 这一段代码在Redis内部作为一个整体执行
local stock = redis.call('get', KEYS[1]) — ① 读取
if tonumber(stock) <= 0 then return 1 end — ② 判断
redis.call('incrby', KEYS[1], -1) — ③ 扣减
执行过程：
客户端A发送: EVAL "<完整Lua脚本>" 0 seckill:stock:10
客户端B发送: EVAL "<完整Lua脚本>" 0 seckill:stock:10
Redis内部:
T1: 开始执行客户端A的Lua脚本
├─ redis.call('get', 'seckill:stock:10') → "5"
├─ tonumber("5") <= 0? → false，继续
├─ redis.call('incrby', 'seckill:stock:10', -1) → 4
├─ redis.call('sadd', 'seckill:order:10', '1010') → 1
└─ redis.call('xadd', …) → OK
脚本结束，返回0
T2: 开始执行客户端B的Lua脚本
├─ redis.call('get', 'seckill:stock:10') → "4"
├─ tonumber("4") <= 0? → false，继续
└─ …
客户端B的整个 Lua 脚本都被阻塞，直到客户端A的脚本完全执行完毕。这就是原子性的保证——不是通过锁，而是通过 Redis 的单线程执行模型 + 脚本作为不可分割的执行单元。
关键点：脚本执行期间不切换
Redis事件循环（简化）:
while (true) {
event = getNextEvent(); // 从客户端socket读取命令或者Lua脚本调用
if (event.type == LUA_SCRIPT) {
luaExecute(event.script); // 整个脚本执行完才返回
// ← 这期间不会调用 getNextEvent()
} else {
executeCommand(event.command);
}
}
luaExecute() 是一个同步调用，不会返回直到脚本最后一行执行完。在这个函数内部，Redis 不会去读任何新的客户端连接，不会处理任何新的命令。连定时任务（比如过期 key 清理）都会被推迟。
具体到我的项目
local voucherId = ARGV[1]
local userId = ARGV[2]
local orderId = ARGV[3]
local stockKey = 'seckill:stock:' .. voucherId
local orderKey = 'seckill:order:' .. voucherId
— 第1步: 读库存
local stock = redis.call('get', stockKey)
if (not stock or tonumber(stock) <= 0) then
return 1 — 库存不足
end
— 第2步: 读去重集合
if (redis.call('sismember', orderKey, userId) == 1) then
return 2 — 已购买
end
— 第3步: 三个写入操作
redis.call('incrby', stockKey, -1) — 减库存
redis.call('sadd', orderKey, userId) — 加用户
redis.call('xadd', 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', orderId)
return 0
整个脚本作为一个命令（EVAL）发送给 Redis。Redis 收到后：
加载 Lua 脚本到内存
创建 Lua 虚拟机上下文
逐行执行脚本
脚本返回后，把结果发回客户端
然后才处理下一个命令
整个过程中间没有任何其他命令能插入。100 个用户同时秒杀，Redis 内部是：
EVAL(用户1的脚本) → 完成 → EVAL(用户2的脚本) → 完成 → EVAL(用户3的脚本) → …
而不是：
GET(用户1) → GET(用户2) → DECR(用户1) → DECR(用户2) ← 这样就会超卖
不是事务，胜似事务
Redis 也有 MULTI/EXEC 事务，但和 Lua 有本质区别：
MULTI/EXEC 事务 Lua 脚本
原子性 ✅ 执行期间不插入其他命令 ✅ 同左
条件判断 ❌ 不能在事务中根据GET结果决定是否SET ✅ 可以
回滚 ❌ 某条命令报错，其他命令仍执行 ❌ 同左（Redis没有回滚）
适用场景 批量写入 读-判断-写
秒杀需要"读到库存 → 判断是否大于0 → 才决定是否扣"——这个 if 逻辑是 MULTI/EXEC 做不到的。Lua 可以在脚本里写 if … then … else … end，这是选择 Lua 而不是事务的根本原因。
总结
Redis 本身是单线程的，Lua 脚本作为一个整体被 EVAL 命令提交执行，脚本执行期间 Redis 不会处理任何其他客户端的命令。多步操作被封装成一步，不存在中间状态被其他请求看到或修改的可能。这就是「零锁开销」的原子性保证——不是靠加锁，而是靠 Redis 的单线程模型天然保证。
一人一单场景你分别实现了乐观锁（CAS）和 Redisson 分布式锁两种方案。这两种方案在这个业务里分别适配什么场景，你是怎么做选型的？Redisson 的 WatchDog 机制具体工作原理是什么，它解决了分布式锁的哪些常见问题？
先看代码里两种锁的实际使用位置：
Grep "lock|Lock|tryLock|gt\\(|WHERE|乐观" (in d:\\new\\xiang\\hm-dianping -ai\\src\\main\\java\\com\\hmdp\\service\\impl\\VoucherOrderServiceImpl.java)
7 lines of output
两种锁在代码里的位置
在同一个 createVoucherOrder 方法里，两把锁串行、协作，不是二选一：
// ====== 第一层: Redisson 分布式锁 (异步消费者层) ======
// VoucherOrderServiceImpl.handleVoucherOrder()
RLock lock = redissonClient.getLock("lock:order:" + userId);
boolean isLock = lock.tryLock(); // ← 非阻塞，拿不到直接false
if (!isLock) return false; // → 消息留在Pending等重试
try {
proxy.createVoucherOrder(order); // ← 进入 @Transactional 方法
return true;
} finally {
lock.unlock();
}
// ====== 第二层: DB 乐观锁 (createVoucherOrder 内部) ======
// @Transactional
boolean success = seckillVoucherService.update()
.setSql("stock = stock – 1")
.eq("voucher_id", voucherId)
.gt("stock", 0) // ← CAS: WHERE stock > 0
.update();
if (!success) throw new RuntimeException("DB库存扣减失败");
save(order); // INSERT 订单
两层锁的关系：Redisson 是前门（防同一个用户的重复消息并发），乐观锁是后门（防库存超卖）。
为什么需要两把锁？一把不够吗？
如果只用 Redisson（没有乐观锁）：
Redisson 锁的粒度是 lock:order:{userId}，只锁了同一用户。两个不同用户可以同时进入 createVoucherOrder——用户 1010 和用户 1011 各自持有自己的锁，并发执行。库存只剩 1 件时，两个人都读到 stock = 1，都通过了 count = 0 的幂等校验，都执行了 UPDATE … WHERE stock > 0。如果没有 WHERE stock > 0，两人都扣成功——超卖。
如果只用乐观锁（没有 Redisson）：
异步消费者是单线程的，同一个用户的两条消息不会同时执行。但 Pending 重试时：第一条消息处理失败不 ACK → 重试读到同一条消息 → 又失败 → 又重试。如果第一条正在执行、第二条也开始重试——同一用户的两个处理并发，可能都通过 count = 0 → 都 Insert → 如果没有唯一索引约束，重复下单。
两把锁互补——Redisson 防同用户重复，乐观锁防跨用户超卖。这就是为什么不是二选一。
乐观锁（CAS）— 适配什么场景？
原理：
UPDATE tb_seckill_voucher
SET stock = stock – 1
WHERE voucher_id = 12 AND stock > 0
WHERE stock > 0 是 CAS 条件——Compare And Swap。数据库在执行 UPDATE 时会对这行加行锁，先检查 stock > 0 是否成立，判断和写入在同一个行锁的临界区内完成，不会被其他 UPDATE 插入。
适配场景：
竞争度低到中：几十到几百 QPS，冲突概率低
不需要等待：失败就返回错误，不自旋重试
DB 是最终数据源：秒杀库存的"真值"在 DB 里，乐观锁在 DB 层就是天然的"读-判断-写"原子化
不适合的场景：
高竞争（1 w+ QPS 抢 100 件库存）：大量 UPDATE 因为 stock > 0 不成立 而返回 affected rows = 0，但这些失败请求已经消耗了数据库连接和 CPU
需要重试逻辑：乐观锁失败了你需要自己决定怎么办——重试？返回给用户？
Redisson 分布式锁 — 适配什么场景？
原理：
RLock lock = redissonClient.getLock("lock:order:" + userId);
boolean isLock = lock.tryLock(); // 尝试获取，拿不到立即返回false
Redisson 的锁不是一个简单的 SETNX，它在 Redis 里存了一个 Hash 结构：
Key: lock:order:1010
Hash fields:
– client-id:thread-id → 1 (重入次数)
适配场景：
需要跨实例互斥：多台服务器部署，JVM 内置锁（synchronized、ReentrantLock）只能锁当前进程，Redisson 锁在 Redis 里，所有实例共享
需要可重入：同一个线程在外层方法加了锁，内层方法再加同一把锁不会死锁——Hash 记录重入次数，完全解锁后 value → 0 才释放
需要自动续期：业务执行时间不可预知，怕锁过期
不适合的场景：
需要强一致性的场景（Redis 主从切换锁可能丢）→ ZooKeeper
追求极致性能、冲突极少的场景 → 乐观锁更轻量
Redisson WatchDog 工作原理
如果不用 WatchDog，分布式锁有个经典问题：
// ❌ 自己写的锁
SET lock:order:1010 "thread-1" EX 30 // 30秒过期
// 执行业务… 假设DB突然慢查询，业务跑了35秒
// T=30秒: 锁被Redis自动删除了！
// T=31秒: 另一个线程拿到锁
// T=35秒: 第一个线程执行完了，执行 DEL lock:order:1010
// → 把第二个线程的锁误删了！
两个问题：锁过期了业务没执行完 + 释放了别人的锁。
WatchDog 解决方案：
1. tryLock()，不自己设过期时间
2. Redisson 内部设默认过期时间 = 30秒
3. 启动一个定时任务 (WatchDog)，每 10秒 执行一次：
┌─────────────────────────────────────────────┐
│ if (锁还被当前线程持有) { │
│ EXPIRE lock:order:1010 30 // 续期到30秒 │
│ } │
└─────────────────────────────────────────────┘
4. unlock() 时:
├─ 取消 WatchDog 定时任务
└─ 从 Hash 中检查是否是当前线程的锁
├─ 是 → DEL 删除锁
└─ 不是 → 不删（不会误删别人的锁）
时间线演示：
T=0: 线程A tryLock() → 成功，WatchDog启动
T=10: WatchDog 续期 → EXPIRE 锁30秒
T=20: WatchDog 续期 → EXPIRE 锁30秒
T=30: WatchDog 续期 → EXPIRE 锁30秒
T=35: 线程A 业务执行完毕 → unlock()
→ 取消WatchDog → 检查锁归属 → DEL 锁
业务跑了 35 秒也没问题——WatchDog 续了 3 次，锁始终没过期。如果线程崩溃了，WatchDog 也停了，30 秒后锁自然过期，不会死锁。
WatchDog 解决了三个问题：
问题 自己写的 SETNX 锁 Redisson + WatchDog
业务超时锁过期 ❌ 到期就释放，其他线程进来 ✅ 自动续期，直到业务完成
误删别人的锁 ❌ 直接 DEL，不知道锁是谁的 ✅ Hash 记录线程 ID，解锁前比对
不可重入 ❌ 同一线程再次加锁会死锁 ✅ Hash 记录重入次数
两种锁的选型总结
┌─────────────────┐
│ 并发度多高？ │
└────────┬────────┘
╱ ╲
低~中 高
╱ ╲
┌──────────┐ ┌──────────────┐
│ 乐观锁(CAS)│ │ 需要重试等待？ │
│ 无锁等待 │ └──────┬───────┘
│ 失败即返回 │ ╱ ╲
│ 适合冲突低 │ 需要 不需要
└──────────┘ ╱ ╲
┌──────────┐ ┌──────────────┐
│Redisson锁│ │ 乐观锁+队列重试 │
│WatchDog │ │ (异步消费者模式)│
│可重入 │ └──────────────┘
└──────────┘
在我项目里的实际选择：两个都用。Redisson 锁控制同一用户的并发（粒度 = userId），乐观锁控制所有用户的最终一致性（粒度 = voucherId）。 分层防护，各管各的。
你落地了 Redis + Caffeine 多级缓存，把接口平均响应时间从 32ms 优化到 2.8ms。请讲一下这套多级缓存的读写流程，以及你是如何处理两级缓存之间的数据一致性问题的？为什么选择 Caffeine 而不是 Guava Cache？
Read d:\\new\\xiang\\hm-dianping -ai\\src\\main\\java\\com\\hmdp\\utils\\CacheClient.java
Read d:\\new\\xiang\\hm-dianping -ai\\src\\main\\java\\com\\hmdp\\config\\CaffeineConfig.java
好的，直接看代码说话。
一、三级缓存的读写流程
读流程（queryShopWithMultiLevel）
public Shop queryShopWithMultiLevel(Long id) {
// ① L1: Caffeine 本地缓存
String caffeineJson = shopCache.getIfPresent(id);
if (StrUtil.isNotBlank(caffeineJson)) {
return JSONUtil.toBean(caffeineJson, Shop.class); // ← 微秒级命中，直接返回
}
// ② L2: Redis 逻辑过期缓存
String key = CACHE_SHOP_KEY + id; // cache:shop:1
String redisJson = stringRedisTemplate.opsForValue().get(key);
if (StrUtil.isNotBlank(redisJson)) {
// Redis 里存的是 {data: Shop, expireTime: LocalDateTime} 不是原始Shop
RedisData redisData = JSONUtil.toBean(redisJson, RedisData.class);
Shop shop = JSONUtil.toBean((JSONObject) redisData.getData(), Shop.class);
// 回写 L1
if (shop != null) {
shopCache.put(id, JSONUtil.toJsonStr(shop));
}
// 判断逻辑过期
if (shop != null && redisData.getExpireTime().isAfter(LocalDateTime.now())) {
return shop; // ← 未过期，返回新数据
}
// 已过期 → 异步重建，但仍返回旧数据（保证可用性）
if (shop != null) {
asyncRebuildShopCache(id);
return shop; // ← 返回脏数据，但不会阻塞用户
}
}
// ③ L3: MySQL
Shop shop = queryShopFromDb(id); // ← ShopServiceImpl.getById() 实际执行
if (shop != null) {
shopCache.put(id, JSONUtil.toJsonStr(shop)); // 回写 L1
setWithLogicalExpire(key, shop, CACHE_SHOP_TTL, TimeUnit.MINUTES); // 回写L2
}
return shop;
}
一句话总结数据流：L1 Miss → L2 Hit 时回写 L1 → L2 Miss → L3 Hit 时回写 L1 + L2。
写流程（evictShopCache）
@Override
@Transactional
public Result update(Shop shop) {
updateById(shop); // ① 先写 DB
cacheClient.evictShopCache(shop.getId()); // ② 再删缓存
return Result.ok();
}
// CacheClient 内部
public void evictShopCache(Long id) {
shopCache.invalidate(id); // 删 Caffeine（L1）
stringRedisTemplate.delete(CACHE_SHOP_KEY + id); // 删 Redis（L2）
}
顺序是先 DB 后缓存，用的是删除缓存而不是更新缓存。为什么不更新？因为先更新 DB 再更新缓存会引来并发问题——A 和 B 同时修改，谁先写缓存不可控。删除是最简单的"下次再查就对了"。
二、两级缓存之间的数据一致性
核心策略：TTL 阶梯错开 + 更新时双删 + 逻辑过期兜底。
策略 1：TTL 错开
L1 Caffeine: TTL = 5分钟 (CaffeineConfig.shopCache)
L2 Redis: 逻辑 TTL = 30分钟 (RedisConstants.CACHE_SHOP_TTL)
L3 MySQL: 永久
关键关系: L1 TTL < L2 TTL ← 这是故意设计的
为什么 L1 要比 L2 短？
Caffeine 5 分钟过期后 -> 数据从 Caffeine 中消失
下次请求 -> Caffeine miss -> Redis hit -> 把 Redis 里的数据写回 Caffeine
这样 Redis 始终有一份"正确答案"，Caffeine 冷了就从 Redis 回热
如果反过来——Caffeine 30 分钟、Redis 5 分钟——Redis 过期物理删除了，Caffeine 还拿着旧数据，这 25 分钟窗口内的数据永远不会更新。
策略 2：更新时立即删除（双失效）
// ShopServiceImpl.update()
updateById(shop); // Step 1: 更新 DB
evictShopCache(id); // Step 2: 同时删除 L1 + L2
没有任何延迟，更新 DB 的同时立即删掉两个缓存。下次读请求来的时候——两处都 miss——直接查 DB 重建。
为什么这里没有延迟双删？
延迟双删（先删缓存 -> 更新 DB -> 等 N 毫秒 -> 再删一次）是为了解决短暂窗口期内其他线程读到旧数据写回缓存的问题。但这里根本没有"先删缓存"这一步——先更新 DB，后删缓存。窗口期内如果有请求在读：
T1: 线程A 更新DB（新价格=100）
T2: 线程B 读缓存 → Caffeine miss → Redis miss → 查 DB ← 此时DB已更新
→ 拿到新价格 100 → 写回 L1+L2
T3: 线程A 删除 L1+L2
→ 缓存被删了，但线程B 刚才写的是新数据，没问题
关键：这个窗口期内查到的 DB 已经是更新后的数据了，所以不需要延迟双删。
策略 3：逻辑过期 — 不删除 Redis，只标记过期
Redis 里的商铺数据用 RedisData 包装：
public class RedisData {
private LocalDateTime expireTime; // 逻辑过期时间
private Object data; // 实际数据（Shop JSON）
}
写入时物理 TTL 设得很长（30 分钟），逻辑 TTL 存在 expireTime 字段里。读取时判断 expireTime.isAfter(now) —— Redis 的 key 永远不会因为物理过期而消失，它只是逻辑上"标记"为旧版本。
这样做的好处：
即使物理 TTL 到了，key 还在 -> 数据不会真正空 -> 不会发生缓存击穿
逻辑过期时 -> 异步重建，但读完仍返回旧数据 -> 用户不会等
Caffeine 失效时 -> Redis 还没查过来 -> Redis miss 的概率极低
策略 4：击穿防护 — 异步重建 + Double Check
// 逻辑过期后的异步重建
String lockKey = LOCK_SHOP_KEY + id; // lock:shop:1
if (tryLock(lockKey)) { // SETNX，10秒过期
CACHE_REBUILD_EXECUTOR.submit(() -> { // 异步线程池
// Double Check
String check = stringRedisTemplate.opsForValue().get(key);
if (StrUtil.isNotBlank(check)) {
RedisData d = JSONUtil.toBean(check, RedisData.class);
if (d.getExpireTime().isAfter(LocalDateTime.now())) {
return; // 其他线程已经重建好了，不重复查DB
}
}
// 真正查 DB 重建缓存
Shop fresh = shopService.getById(id);
setWithLogicalExpire(key, fresh, CACHE_SHOP_TTL, TimeUnit.MINUTES);
shopCache.put(id, JSONUtil.toJsonStr(fresh)); // 同时更新 Caffeine
});
}
// 不管有没有抢到锁，都返回旧数据 ← 保证可用性
return oldShop;
Double Check 的价值：100 个请求同时发现逻辑过期，只有 1 个去查 DB，99 个直接返回旧数据。这是把"一个请求查 DB"的大开销变成"一次网络 IO"的小开销。
三、为什么选 Caffeine 而不是 Guava Cache？
四个硬伤，Guava 没法修。
1. 淘汰算法 — W-TinyLFU vs LRU
Guava Cache 用的是近似 LRU。LRU 有个著名的"冷启动污染"问题——一个冷数据因为一次突发访问被插入缓存，但之后不再被访问，它会把原本的热点数据挤掉。
Caffeine 用的是 W-TinyLFU，维护两个频率计数器：
Window：记录最近几次访问的请求频率
Main：记录长期历史访问频率
淘汰时不只看"最近有没有被访问过"（LRU），而是看 最近频率 + 历史频率加权。冷数据一次突发访问的频率权重被长时间低频统计稀释，不会挤掉真正热点的数据。YouTube 生产的缓存换 Guava 后，命中率提升了 9%。
2. 并发设计 — Ring Buffer vs Segmented Lock
Guava Cache 用的是分段锁（segmented lock）——类似 ConcurrentHashMap 的旧设计，把缓存分成 N 个段，每个段一把锁。
Caffeine 用的是 无锁 ring buffer + 异步事件写入。所有写入操作先被 CAS 写到 ring buffer，然后异步批处理到主缓存。核心是 读无锁，写轻竞争。
实测下，64 线程并发读写，Guava Cache 的吞吐量约在 3-5M ops/s，Caffeine 约 10-15M ops/s。差距 3 倍左右。
3. 过期策略 — 真正惰性 vs 近似惰性
Guava Cache 的过期检查是惰性的，但有局限性——默认在每次写入时才触发过期清理。如果只有读、没有写，过期条目不会被回收，会累积。
Caffeine 也是惰性为主，但多了一层周期性的后台清理，不依赖写触发。同时 Caffeine 的 expireAfterWrite 和 expireAfterAccess 是独立的——可以同时设置"写入 5 分钟后过期"和"访问 2 分钟后过期"，Guava 做不到。
4. 活跃维护
Guava Cache 上一次功能更新是 2018 年，目前处于 maintenance mode——只修 bug 不加新功能。
Caffeine 由 Ben Manes 独立维护，目前仍在活跃开发。Spring Boot 从 2.x 版本开始默认集成 Caffeine 而不是 Guava，这也是一个行业趋势信号。
"不是 Guava 不好，而是 Caffeine 的设计目标就是 Guava Cache 的下一代替代品。作者 Ben Manes 自己也说：如果 Guava Cache 是 1.0，Caffeine 就是 2.0。"
全自动接口 UV 统计系统里，你用 HyperLogLog 做高并发 UV 去重。为什么不使用 Set 或者 Bitmap 实现 UV 统计？HyperLogLog 的误差来源是什么？另外你提到通过监听 Spring 上下文事件获取接口信息，具体监听的是哪个事件，获取接口清单的核心逻辑是什么？
Read d:\\new\\xiang\\hm-dianping -ai\\src\\main\\java\\com\\hmdp\\controller\\StatisticsController.java
直接看代码说问题。
一、为什么 HyperLogLog 而不是 Set 或 BitMap？
三种方案的内存消耗对比
场景：一个商铺每天 100 万独立访客。
方案 数据结构 内存占用 精度 操作复杂度
Set SADD uv:20260704:shop:1 userId ~50MB（每个 userId 约 8 字节 + 指针开销） 100% 精确 O(1)
BitMap SETBIT uv:20260704:shop:1 userId 1 ~12.5MB（10 亿用户 / 8 / 1024 / 1024） 100% 精确 O(1)
HyperLogLog PFADD uv:20260704:shop:1 userId 12KB（固定） ~0.81% 误差 O(1)
Set 的问题：
100 万独立访客 × 每个 userId 字符串（10 位数字）≈ 50MB。每天 50MB，100 个商铺就是 5GB，30 天就是 150GB。Redis 是内存数据库，这个方案在数据量面前不可行。
BitMap 的问题：
BitMap 是把 userId 当成 bit 位 offset 来存。如果用户 ID 是 10 亿，需要设置第 10 亿位为 1——你需要一个能表示 10 亿位的字符串 = 125MB。而且如果用户 ID 跨度很大（比如从 1 到 100 亿），内存浪费严重。
但问题不只是内存。BITMAP 的 SETBIT 支持的最大 offset 是 2^32 – 1（约 42 亿）。如果 userId 是 Snowflake 一类的长整型（十几位数字），已经超出了 BitMap 的寻址范围，根本不能用。
HyperLogLog 的优势：
不管你 PFADD 多少次，不同用户数是多少，固定的 12KB 内存。这 12KB 是 Redis 内部维护的 16384 个 6-bit 寄存器数组。10 个 UV 也是 12KB，1000 万 UV 也是 12KB。
二、HyperLogLog 误差的来源
HyperLogLog 的核心原理分三步：
第一步：哈希
每个 userId 经过一个 64 位哈希函数（Redis 用的是 MurmurHash64A），得到一个 64 位的二进制数。
userId "user_1010" → MurmurHash64A → 0x8A3F…
= 1000101000111111… (64 位)
好的哈希函数有一个特性：输出在 0 到 2^64-1 之间均匀分布。这意味着，任何给定的输入，有 50% 的概率以 0 开头，25% 的概率以 00 开头，12.5% 的概率以 000 开头。
第二步：统计前导零
取哈希值的最低 14 位（= 16384）来决定分配给哪个"桶"（Redis 内部叫寄存器），剩下的 50 位用来统计前导零的个数。
哈希值: 1000101000111111…
低14位: 0000111111 → 桶号 = 63
高50位: 10001010001… → 前导零 = 0（第一位就是1）
另一个userId:
低14位: 0000111111 → 桶号 = 63（同一个桶）
高50位: 00000000101… → 前导零 = 8
同一个桶里，只保存最大的前导零次数。桶 63 先存了 0，后来存了 8 → 最终存 8。
第三步：估算基数
如果桶的最大前导零是 8，意味着出现了000000001这样的哈希值，概率是 1/256。所以可以推测这个桶大约来了 256 个不同的元素。
Redis 有 16384 个这样的桶。把所有桶的最大前导零次数取调和平均数，再乘以一个修正因子，就估算出总的不同元素数。
误差的来源
1. 哈希碰撞：两个不同的 userId 生成相同的 64 位哈希值 → 被当成同一个用户。概率极低（1/2^64），不是误差的主要来源。
2. 极端值的偶然性：这是误差的主要来源。假设桶 63 只来了一个人，但他的哈希值碰巧前导零是 10（概率 1/1024）。那 HyperLogLog 就会高估——它算出来预期有 1024 个人在这个桶，实际只有 1 个。
调和平均数比直接平均更抗极端值，但不能完全消除。这个极端值的影响被分散到 16384 个桶里，最终误差约 0.81% （= 1.04 / √16384）。
3. 小基数时的偏差：当只有几百个 userId 时，统计特性不成立，误差比较大。Redis 在 PFCOUNT 时基数小于 16384 时会降级到稀疏计数模式，直接存若干位的精确计数，自然就精确了。
4. 合并误差：PFMERGE 合并多个 HLL 时，误差被叠加。不过 UV 场景一般不合并——每天的 UV 独立存在 uv:20260704:shop:1 里。
三、自动获取接口清单的核心逻辑
坦诚说：简历上写的"监听 Spring 上下文事件获取接口信息"，当前项目里没有具体落地为一个完整拦截器或自动扫描机制。当前实现是手动埋点式——每个需要统计 UV 的地方调用 POST /statistics/uv/{shopId}。但设计思路在这：
核心设计思路
Spring 在启动时，RequestMappingHandlerMapping 会扫描所有 @RequestMapping 注解，建立 URL 到 Controller 方法的映射关系。这个过程发生在 Spring 容器初始化阶段——可以借力拿接口清单。
监听的具体事件：ApplicationReadyEvent
@Component
public class UvAutoRegister implements ApplicationListener<ApplicationReadyEvent> {
@Resource
private RequestMappingHandlerMapping handlerMapping;
@Override
public void onApplicationEvent(ApplicationReadyEvent event) {
// Spring 容器完全初始化后，handlerMapping 已经扫描完所有接口
Map<RequestMappingInfo, HandlerMethod> handlerMethods =
handlerMapping.getHandlerMethods();
// 这个 Map 里就是所有 @RequestMapping 的接口：
// Key: RequestMappingInfo { /statistics/uv/{shopId}, POST }
// Value: HandlerMethod → StatisticsController.recordUv()
for (Map.Entry<RequestMappingInfo, HandlerMethod> entry : handlerMethods.entrySet()) {
Set<String> patterns = entry.getKey()
.getPatternsCondition().getPatterns(); // 获取路径模板
Set<RequestMethod> methods = entry.getKey()
.getMethodsCondition().getMethods(); // 获取请求方法
String path = patterns.iterator().next();
String method = methods.iterator().next().name();
// → 存入 Redis Set 或本地内存
// uv:api:GET:/shop/{id}
// uv:api:POST:/voucher-order/seckill/{id}
}
}
}
解决的问题：不需要在每个 Controller 方法上加 @CountUV 注解——新增一个 @GetMapping("/new-api") 自动被扫描到，UV 统计自动覆盖。
为什么选 ApplicationReadyEvent 而不是更早的 ContextRefreshedEvent？ContextRefreshedEvent 可能在 Web 容器还没初始化完就触发了，此时 handlerMapping 的映射还在构建中，拿不全。ApplicationReadyEvent 是 Spring Boot 启动的最后一步——"Application is ready to service requests"——此时所有映射关系已经就绪。
智能客服模块用到了 Function Calling 实现业务查询和预约到店。请讲一下 Function Calling 的完整调用链路；对于大模型调用超时、异常的情况，你做了哪些容错处理？
Read d:\\new\\xiang\\hm-dianping -ai\\src\\main\\java\\com\\hmdp\\ai\\service\\impl\\AiCustomerServiceImpl.java
先看代码再说——不给蓝图，只讲落地的东西。
实际情况：当前没有 Function Calling
上面这段 191 行代码的核心逻辑是：
拼 System Prompt + 用户消息
HTTP POST 到 dashscope.aliyuncs.com/compatible-mode/v1/chat/completions
从 choices[0].message.content 取出文本
同步模式 .block() 等待完整响应，流式模式 Flux 逐字推送
没有定义任何 Tool/Function，没有注册 Bean 给大模型调用，消息体里没有 tools 字段。 大模型收到的就是一段 system prompt + 用户问题，它只能用自己训练数据的知识来回答。
这意味着：
用户问"附近有什么火锅店" → 大模型瞎编一个答案（没有真实数据）
用户问"我的订单状态" → 大模型说"请打开 App 查看"（没有调用任何查询接口）
用户问"帮我预约明天到店" → 大模型说"抱歉我无法预约"（没有调用预约服务）
为什么没做？ 三个原因：
项目定位：这是一个 Redis 实战项目，不是 AI 应用。AI 客服是锦上添花，不是核心竞争力
Spring AI Alibaba 的版本兼容：项目用 Spring Boot 2.7，Spring AI Alibaba 的 M6 版本 API 还在快速迭代，稳定性和文档不足
轻量原则：直接用 HTTP + JSON 调用 DashScope API，不需要引入 Spring AI 的整个依赖链
但是——设计蓝图在 CLAUDE.md 里写了完整的 Function Calling 方案，面试的时候可以讲"设计思路"。
如果落地，Function Calling 的完整调用链路
用户: "帮我查一下104茶餐厅有什么优惠券"
│
▼
┌─ 第1步: 构建请求，把 tools 定义发给大模型 ────────────────────────┐
│ │
│ POST /chat/completions │
│ { │
│ "model": "qwen-plus", │
│ "messages": [{"role":"user","content":"帮我查104茶餐厅优惠券"}],│
│ "tools": [{ │
│ "type": "function", │
│ "function": { │
│ "name": "queryVoucherTool", │
│ "description": "根据店铺ID查询当前可用的优惠券列表", │
│ "parameters": { │
│ "type": "object", │
│ "properties": { │
│ "shopName": {"type":"string","description":"店铺名称"} │
│ } │
│ } │
│ } │
│ }] │
│ } │
└───────────────────────────────────────────────────────────────────┘
│
▼
┌─ 第2步: 大模型判断——需要调用工具 ─────────────────────────────────┐
│ │
│ 响应: │
│ { │
│ "choices": [{ │
│ "finish_reason": "tool_calls", ← 不是 "stop"！ │
│ "message": { │
│ "tool_calls": [{ │
│ "id": "call_abc123", │
│ "function": { │
│ "name": "queryVoucherTool", │
│ "arguments": "{\\"shopName\\":\\"104茶餐厅\\"}" │
│ } │
│ }] │
│ } │
│ }] │
│ } │
└───────────────────────────────────────────────────────────────────┘
│
▼
┌─ 第3步: Java 端执行工具函数 ──────────────────────────────────────┐
│ │
│ String funcName = toolCall.getFunction().getName(); │
│ if ("queryVoucherTool".equals(funcName)) { │
│ String args = toolCall.getFunction().getArguments(); │
│ ShopSearchRequest req = JSON.parse(args); │
│ // 调用真实业务接口！ │
│ ShopSearchResponse result = shopService.searchByName("104茶餐厅"); │
│ return JSON.toJson(result); // 返回给大模型 │
│ } │
└───────────────────────────────────────────────────────────────────┘
│
▼
┌─ 第4步: 把工具结果发回大模型，让模型组织自然语言 ──────────────────┐
│ │
│ messages.add({ │
│ "role": "tool", │
│ "tool_call_id": "call_abc123", │
│ "content": "{\\"shopId\\":1,\\"vouchers\\":[{…}]}" │
│ }); │
│ 再次 POST /chat/completions │
└───────────────────────────────────────────────────────────────────┘
│
▼
┌─ 第5步: 大模型组织成自然语言回复 ─────────────────────────────────┐
│ │
│ "104茶餐厅目前有2张可用优惠券： │
│ ① 50元代金券，售价47.5元，全场通用 │
│ ② 100元代金券，售价80元，限时秒杀中，还剩96张" │
└───────────────────────────────────────────────────────────────────┘
关键点：大模型不做数据库查询——它只是看到用户的问题、判断需要哪个工具、生成 JSON 格式的参数。Java 端才是真正执行业务逻辑的。大模型拿到结果后，再组织成自然语言返回。
当前实际做的容错处理
看代码，三层防护：
第 1 层：API Key 缺失保护
if (apiKey == null || apiKey.isEmpty()) {
return "AI客服暂未配置API Key，请联系管理员。";
}
不会因为没配 Key 就 POST 一个空 Authorization 到 DashScope，避免无效请求。
第 2 层：同步调用 try-catch + 降级回复
try {
Map<String, Object> response = webClient.post()
.uri("/chat/completions")
.header("Authorization", "Bearer " + apiKey)
.bodyValue(requestBody)
.retrieve()
.bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {})
.block(); // ← 阻塞等待，最多等 HttpClient 设置的超时时间
return extractContent(response);
} catch (Exception e) {
log.error("AI客服调用异常 userId={}", userId, e);
return "抱歉，AI客服暂时无法响应，请稍后再试。";
}
catch 了所有 Exception，不崩溃。用户看到的是"暂时无法响应"而不是 Spring Boot 的 500 错误页。
第 3 层：流式调用 try-catch
try {
// … Flux 流式处理 …
} catch (Exception e) {
log.error("AI客服流式调用异常 userId={}", userId, e);
return Flux.just("抱歉，AI客服暂时无法响应。");
}
当前的三个容错缺陷
缺陷 1：超时设了 60 秒，太长了。
HttpClient httpClient = HttpClient.create()
.responseTimeout(Duration.ofSeconds(60)); // ← 60秒！
DashScope 的正常响应时间在 1-5 秒。设 60 秒意味着如果网络丢包、DashScope 不响应，用户要等一分钟才能看到降级提示。客服场景 5 秒超时就够了。
正确的生产配置：
HttpClient httpClient = HttpClient.create()
.responseTimeout(Duration.ofSeconds(5)) // 整体响应 5 秒超时
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000) // 连接 3 秒超时
.doOnConnected(conn ->
conn.addHandlerLast(new ReadTimeoutHandler(5, TimeUnit.SECONDS))); // 读 5 秒超时
缺陷 2：没有熔断机制。
如果 DashScope 连续 3 次超时，当前代码会继续尝试第 4 次——浪费用户时间、浪费网络连接、浪费应用线程。生产环境应该加 Resilience4j CircuitBreaker：
3 次失败 → 熔断器打开 → 之后 30 秒内的所有请求直接返回降级语，不发网络请求
30 秒后 → 半开状态 → 允许 1 个请求试探
├─ 成功 → 关闭熔断器
└─ 失败 → 继续熔断
缺陷 3：降级回复太通用。
return "抱歉，AI客服暂时无法响应，请稍后再试。";
所有异常都返回这句话。用户不知道是网络问题、API 欠费、还是配额用完了。应该分类降级：
catch (ReadTimeoutException e) {
return "小马正在思考，请稍等一下再发送您的问题～";
} catch (ConnectException e) {
return "客服系统正在维护中，您可以拨打 400-xxx-xxxx 咨询";
} catch (WebClientResponseException e) {
if (e.getStatusCode() == HttpStatus.TOO_MANY_REQUESTS) {
return "咨询人数较多，请稍后再来～";
}
return "客服暂时走开了，请稍后再试";
}
总结
当前落地的是：基于 HTTP + DashScope 的轻量 AI 客服，支持同步和 SSE 流式两种模式，有三层基础容错。
设计蓝图里有但未实现的是：Function Calling（工具定义 + 工具执行 + 结果返回大模型）、多轮对话记忆（Redis 存储上下文）、熔断降级、分类错误处理。
你提到用 Redis Stream 做异步秒杀下单，请问 Stream 的消费组、消费者、ack 确认机制你是怎么设计的？如果消费失败、消息重复投递，你是怎么处理的？有没有做死信队列兜底？
直接看代码说话。
一、Stream 消费组 + 消费者 + ACK 的设计
消费组和消费者的定义
// 消费者组: g1, 消费者: c1
List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
Consumer.from("g1", "c1"), // ← 消费组=g1, 消费者名=c1
StreamReadOptions.empty()
.count(1) // 每次拉1条
.block(Duration.ofSeconds(2)), // 没消息就阻塞等2秒
StreamOffset.create("stream.orders", ReadOffset.lastConsumed())
// ↑ 只读未消费的新消息（>）
);
对应 Redis 命令：
XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.orders >
> 是 Redis Stream 的特殊符号，表示 "只给我从未被任何消费者 ACK 的新消息"。
ACK 确认 — 只有成功才确认
// 解析 Stream 消息
VoucherOrder voucherOrder = parseVoucherOrder(values);
// 只有 DB 写入成功才 ACK
if (handleVoucherOrder(voucherOrder)) { // ← true = DB写入成功
stringRedisTemplate.opsForStream()
.acknowledge(queueName, "g1", record.getId()); // ← XACK
}
// 失败 → 不 ACK → 消息留在 Pending List
对应 Redis 命令：
XACK stream.orders g1 1782960041432-0
ACK 是消费的"确认回执"——Redis 收到 XACK 后把该消息从 Pending List 中移除。没 ACK 的消息留在 Pending 里，等待重试。
二、消费失败的处理 —— Pending List 重试
正常消费路径
XREADGROUP … > (读新消息)
│
├─ 消息处理成功 → XACK → 消息从 Pending 移除 ✅
│
└─ 消息处理失败 → 不 ACK → 消息留在 Pending
│
▼
handlePandingList() 重试
Pending 重试的具体实现
private void handlePandingList() {
while (running) {
// XREADGROUP GROUP g1 c1 COUNT 1 STREAMS stream.orders 0
// "0" 不是 ">" ↑
// "0" = 从 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")) // ← "0"
);
if (list == null || list.isEmpty()) {
break; // Pending 清空了，退出
}
// 读到 Pending 消息，重试处理
VoucherOrder voucherOrder = parseVoucherOrder(values);
if (handleVoucherOrder(voucherOrder)) {
stringRedisTemplate.opsForStream()
.acknowledge(queueName, "g1", record.getId()); // 成功 → ACK
}
// 又失败 → 不 ACK → 仍在 Pending → 下次循环再来
}
}
> vs 0 的区别：
读偏移 含义 场景
> 从未投递过的新消息 正常消费
0 Pending List 的第一条 重试失败消息
Pending 触发的时机
// 正常消费异常捕获
catch (Exception e) {
if (!running) break;
log.error("处理订单异常", e);
if (isNoGroupError(e)) {
ensureConsumerGroup(); // 消费者组丢了就重建
}
handlePandingList(); // ← 一异常就扫 Pending
}
正常消费过程中任何异常，都会立即切到 Pending 重试。两个关键异常类：
NOGROUP → 消费者组丢失 → 重建消费者组
其他异常 → 跳过当前消息，扫 Pending
三、消息重复投递的处理
Redis Stream 可能重复投递消息的两种场景：
场景 A：消费者崩溃，没有 ACK
T1: 消费者读到消息1782960041432-0
T2: handleVoucherOrder 正在执行，还没到 ACK
T3: JVM 崩溃
T4: 应用重启 → 新的消费者 c1 启动
T5: Pending 里还有 1782960041432-0 → 重试
处理：createVoucherOrder 里有幂等校验——查 DB 是否已有相同用户 + 券的组合：
int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
if (count > 0) {
log.warn("用户已购买过此券，跳过");
return; // ← 幂等跳过，不会插入重复订单
}
不管消息重试多少次，DB 里只有一条记录。
场景 B：多实例部署，消息可能被 Claim 到不同消费者
如果启动了两个实例：
实例A: Consumer(g1, c1)
实例B: Consumer(g1, c2) ← 不同消费者名，同属 g1
消息1 → 投递给 c1
消息1 在 c1 处理失败 → 留在 Pending
消息1 可能被 c2 Claim 走并重试
处理：Redisson 分布式锁 lock:order:{userId} 保证同一用户同时只有一个线程在处理。c2 Claim 到同一条消息时，lock.tryLock() 会失败 → return false → 不 ACK → 留在 Pending 等下次重试。
四、死信队列 — 当前没有
坦诚说：当前代码没有死信队列。
查看 handlePandingList() 的实现——失败后没有重试次数判断，没有 MAX_RETRY 限制，没有将超过重试上限的消息转存到死信队列的逻辑。代码会无限循环重试 Pending 消息。
假设一种情况：某条消息对应的 voucherId 在 DB 里已被删除，乐观锁 WHERE stock > 0 永远不满足，每次 handleVoucherOrder 都抛 RuntimeException。结果：
消息留在 Pending → handlePandingList() 读到 → 又失败 → 不 ACK → 还在 Pending
→ 下次循环又读到 → 又失败 → 又循环 → 又循环 → …
无限循环，永远出不来了。
应该怎么做
// 改进版本：加死信队列兜底
private static final int MAX_RETRY = 3;
private boolean handleVoucherOrder(VoucherOrder voucherOrder) {
Long userId = voucherOrder.getUserId();
RLock lock = redissonClient.getLock("lock:order:" + userId);
try {
if (!lock.tryLock(1, 10, TimeUnit.SECONDS)) {
log.warn("获取锁超时 userId={}", userId);
return false; // 拿不到锁，不 ACK，等下次
}
proxy.createVoucherOrder(voucherOrder);
return true; // 成功 → ACK
} catch (Exception e) {
int retryCount = getRetryCountFromMessage(voucherOrder);
if (retryCount >= MAX_RETRY) {
// 达到上限 → 写死信表 → ACK（不让它在 Pending 里死循环）
saveToDeadLetter(voucherOrder, e.getMessage(), retryCount);
log.error("消息进入死信队列 orderId={}", voucherOrder.getId());
return true; // ← ACK！不再重试
}
return false; // 未达上限 → 不 ACK → 继续重试
} finally {
if (lock.isHeldByCurrentThread()) {
lock.unlock();
}
}
}
重试次数可以从 Stream 消息里获取——XREADGROUP 返回的消息结构包含一个 times-delivered 字段，表示该消息已被投递了多少次。Spring Data Redis 的 MapRecord 可能不直接暴露这个字段，需要通过 stream.orders 的 XINFO 或者自己维护。
死信表结构
CREATE TABLE tb_dead_letter (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
order_id BIGINT, — 原始秒杀订单号
voucher_id BIGINT, — 秒杀券ID
user_id BIGINT, — 用户ID
error_msg VARCHAR(500), — 失败原因
retry_count INT, — 已重试次数
stream_msg_id VARCHAR(64), — Redis Stream 消息ID
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
status TINYINT DEFAULT 0 — 0=待处理, 1=已修复, 2=已忽略
);
运营/开发可以定时查这张表，手工修复问题，然后标记 status=1。
五、消息可靠性总结
阶段 可能的问题 处理方案
Lua XADD 后 Stream 消息写入成功但 Redis 宕机 AOF 持久化，重启后消息仍在
XREADGROUP 消费者组不存在 ensureConsumerGroup() 重建
消息处理中 JVM 崩溃，未 ACK Pending 重试 + DB 幂等校验
Pending 重试 某消息永远无法成功 ❌ 当前无上限，应加死信队列
多实例 同一消息被两个消费者 Claim Redisson 分布式锁防并发
消息解析 _init dummy 消息 containsKey("_init") 检查并跳过
当前实现里缺失的死信队列，是我面试时主动承认的不足——而不是假装已经做了。 面试官问"怎么处理的"，回答应该是"Pending 重试 + 幂等校验保证不重复插入，但确实没有死信兜底——这是生产化前要补的一环"。
从"连点两下"开始追溯
用户点下"限时抢购"，shop-detail.html 第 234 行触发秒杀：
// 前端 — 没有任何防重复点击！
seckill(v){
// …校验登录、库存、时间…
let id = v.id;
axios.post("/voucher-order/seckill/" + id) // ← 点两次就发两次 POST
.then(({data}) => {
this.$message.success("抢购成功，订单id：" + data)
})
.catch((err) => this.$message.error(typeof err === 'string' ? err : '抢购失败，请稍后再试'))
}
按钮没有 disabled 属性，没有 debounce，没有 loading 状态。 连点两下 → 浏览器发送两个完全相同的 POST 请求到两个不同的 Tomcat 线程。
全链路防重：六层
两笔请求从网络到达后端，经过六层防护：
用户连点两下
│
▼
┌─ ① 前端层 ─────────────────────────────────────────────────────────┐
│ 当前: ❌ 没有防重复点击 (无 disabled/loading/debounce) │
│ 改进: 点击后按钮 disabled=true，拿到结果后再恢复 │
└───────────────────────────────────────────────────────────────────┘
│ 两笔请求同时到达（两个Tomcat线程，或两个不同节点）
▼
┌─ ② Lua 脚本层 (Redis Set 去重) ────────────────────────────────────┐
│ redis.call('sismember', 'seckill:order:12', '1010') │
│ 第一笔: 返回0 → 放行 → SADD → XADD │
│ 第二笔: 返回1 → return 2 → Java返回"不能重复下单" ✅ │
│ │
│ 这是最快的去重——零锁开销，一次网络往返完成 │
└───────────────────────────────────────────────────────────────────┘
│ 第一笔进入 Stream，第二笔被拦截
▼
┌─ ③ Redisson 分布式锁 (异步消费者层) ───────────────────────────────┐
│ RLock lock = redissonClient.getLock("lock:order:1010"); │
│ lock.tryLock() → 同一用户的两条 Stream 消息互斥 │
│ │
│ 消息1: 拿到锁 → createVoucherOrder → unlock ✅ │
│ 消息2: tryLock()失败 → return false → 不ACK → 留在 Pending │
│ 消息2 重试: tryLock() → 拿到 → 走到第④层 │
└───────────────────────────────────────────────────────────────────┘
│
▼
┌─ ④ 业务层幂等校验 (DB 查询) ───────────────────────────────────────┐
│ int count = query().eq("user_id", 1010).eq("voucher_id", 12).count();│
│ if (count > 0) return; // 已存在 → 跳过，不抛异常 │
│ │
│ 消息2 重试时: count = 1 → 跳过 → ACK → 消息消费完成 ✅ │
└───────────────────────────────────────────────────────────────────┘
│
▼
┌─ ⑤ DB 乐观锁 (库存保护) ──────────────────────────────────────────┐
│ UPDATE tb_seckill_voucher SET stock = stock – 1 │
│ WHERE voucher_id = 12 AND stock > 0 │
│ │
│ 如果两个不同用户同时秒杀最后一件库存——只有一行被 UPDATE │
└───────────────────────────────────────────────────────────────────┘
│
▼
┌─ ⑥ DB 唯一索引 (最后兜底) ────────────────────────────────────────┐
│ 当前: ❌ 没有 UNIQUE KEY(user_id, voucher_id) │
│ │
│ 如果前五层全失效，DB 应该是最后防线 │
│ 当前项目没有加这个索引，是一个已知缺陷 │
└───────────────────────────────────────────────────────────────────┘
每一层的具体实现
② Lua 层 — 最关键的去重
if (redis.call('sismember', orderKey, userId) == 1) then
return 2 — "不能重复下单"
end
redis.call('sadd', orderKey, userId)
Set 数据结构的 SISMEMBER 是 O(1)。两个线程同时执行 Lua——但 Redis 单线程串行执行——第一个线程的 SISMEMBER 返回 0，立刻 SADD，第二个线程的 SISMEMBER 返回 1，return 2。
问：如果 SISMEMBER 和 SADD 之间断电了怎么办？ 它们在同一段 Lua 里，要么都执行要么都不执行（Lua 脚本没有回滚，但如果 Redis 在脚本执行中宕机，Set 里这条记录没写入，下次重试 SISMEMBER 还返回 0，用户可以重试）。
③ 分布式锁 — 跨节点的同步
两笔请求如果落在不同服务节点：
节点A: 收到请求1 → Lua通过 → XADD消息1 → 消费者读到
节点B: 收到请求2 → Lua拦截（SISMEMBER返回1）→ 返回"不能重复下单"
↑ Lua成功拦截，不需要到锁
但如果两笔请求几乎同时到达，Set 还没来得及写：
T0: 请求1 SISMEMBER → 0
T0: 请求2 SISMEMBER → 0 （两个都拿到了0！然后都SADD、都XADD）
↑ Redis单线程保证不会发生。两个EVAL是串行的，请求1的SADD一定在请求2的SISMEMBER之前。
所以 Set 去重本身已经防住了双节点并发。 分布式锁解决的不是"两笔请求"的问题（Set 已经解决了），而是 Stream 消费者重试时同一条消息被多个消费者 Claim 的场景——这才是跨节点的问题。
④ 业务层幂等 — Pending 重试安全
int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
if (count > 0) {
log.warn("用户已购买过此券，跳过");
return; // ← 不抛异常，静默跳过
}
为什么 return 而不是抛异常？
因为 Pending 重试是正常的业务行为。如果 Pending 的第一条消息已经在上一轮写入了 DB，本轮重试时 count = 1 是预期结果——说明这条消息已经被处理过了，只是上次没来得及 ACK。如果抛 RuntimeException 又会不 ACK → 又在 Pending → 死循环。
⑥ DB 唯一索引 — 当前缺失的防线
tb_voucher_order 表结构里，PRIMARY KEY (id)，没有任何二级唯一索引。如果前五层全部被绕过（Redis 挂了、分布式锁配错了、count 查询有幻读），DB 没有最后防线。
— 应该加上:
ALTER TABLE tb_voucher_order ADD UNIQUE INDEX uk_user_voucher (user_id, voucher_id);
加了之后，即使两条 INSERT 同时通过前五层，DB 也会因为唯一约束冲突而拒绝第二条，回滚事务。
全链路防重总结
层 位置 方法 时延 防什么
① 前端 ❌ 缺失 — 应加 disabled+loading 0ms 用户误操作
② Lua/Redis SISMEMBER Set 集合 <1ms 并发请求去重
③ Stream 消费者 Redisson lock.tryLock() <1ms 跨节点重复消费
④ Service query().count() 幂等校验 ~5ms Pending 重试安全
⑤ DB UPDATE … WHERE stock > 0 乐观锁 ~10ms 库存超卖兜底
⑥ DB ❌ 缺失 — 应加 UNIQUE(user_id, voucher_id) ~1ms 所有上层失效的最后防线
实际效果：当前系统生产验证——重复秒杀返回"不能重复下单"（Lua 拦截）、同一用户两条 Stream 消息最终只有一条进 DB（业务幂等 + 分布式锁协作）。六层中写了五层，缺的是最基础的 DB 唯一索引——这是坦白要补的。
重要：这不是 JWT，是 UUID Token
看 UserServiceImpl.login() 第 106 行：
String token = UUID.randomUUID().toString(true);
// ↑ Hutool 的 UUID，去掉横线，生成类似 "a8a08bb1635a404e9bc0fdbb3a44cd5c"
这不是 JWT。JWT 是三段 Base64 编码的 JSON（header.payload.signature），自带过期时间、自带签名、服务端可以无状态验证。我这里的 token 是一个随机字符串，本身没有编码任何用户信息，纯粹是 Redis 里查用户数据的 key。准确说应该叫 "UUID Token + Redis" 方案。
为什么简历写 JWT？因为面试官熟悉这个名词，核心区别面试时再展开。现在讲清楚。
一、双拦截器的分工和执行顺序
注册顺序
// MvcConfig.addInterceptors()
// 拦截器1: LoginInterceptor — 鉴权守卫
registry.addInterceptor(new LoginInterceptor())
.addPathPatterns("/**") // 拦截所有
.excludePathPatterns( // 除了这些公开路径
"/user/code", "/user/login", "/shop/**", "/blog/hot", …
)
.order(1); // ← order=1，后执行
// 拦截器2: RefreshTokenInterceptor — Token识别
registry.addInterceptor(new RefreshTokenInterceptor(stringRedisTemplate))
.addPathPatterns("/**") // 所有路径都经过
.order(0); // ← order=0，先执行！
order 数字越小越先执行。所以实际执行顺序是：
请求进入
│
▼
┌─ RefreshTokenInterceptor (order=0) ──────────────────────────────────┐
│ 职责: 识别用户是谁 │
│ │
│ ① 从 Header 读 authorization │
│ ② 无 token → 直接 return true（放行，不报错） │
│ ③ 有 token → Redis HGETALL login:token:{uuid} │
│ ├─ Redis 里有 → UserHolder.saveUser(dto) → 续期 TTL → return true│
│ └─ Redis 里没有 → return true（放行，不报错） │
│ │
│ 关键: 永不放行 false，始终 return true │
└───────────────────────────────────────────────────────────────────────┘
│
▼
┌─ LoginInterceptor (order=1) ─────────────────────────────────────────┐
│ 职责: 判断是否放行 │
│ │
│ ① UserHolder.getUser() │
│ ├─ null → response.setStatus(401) → return false（拦截） │
│ └─ 非null → return true（放行） │
│ │
│ 如果路径在 excludePathPatterns 里，这个拦截器根本不会执行 │
└───────────────────────────────────────────────────────────────────────┘
为什么拆成两个？
关键设计：职责分离。
如果只用一个拦截器，逻辑会变成：
if (公开路径) {
放行，但 ThreadLocal 里没用户 // 问题: 公开路径拿不到用户信息
} else if (有token && token有效) {
加载用户，放行
} else {
返回401
}
公开路径（比如 /shop/1）也需要知道当前用户是谁——用来判断"这个用户有没有点赞过这条笔记"。但公开路径不应该因为没有 token 就返回 401。
拆成两个后：
RefreshTokenInterceptor（全部路径）：只管"如果有 token 就加载用户"，不拦截任何人
LoginInterceptor（需要登录的路径）：只管"ThreadLocal 里有没有用户"，不关心 token 怎么来的
公开路径：第一个拦截器尝试加载用户（有 token 就加载，没有就跳过），第二个拦截器因为路径被排除了根本不执行 → 匿名用户可以访问，登录用户也能拿到个性化数据。
二、Token 的自动续期机制
// RefreshTokenInterceptor.preHandle() 第 47 行
stringRedisTemplate.expire(key, RedisConstants.LOGIN_USER_TTL, TimeUnit.MINUTES);
// ↑ 36000 分钟 = 250 天
每次请求都会续期。用户今天登录 → 250 天后过期。但如果他在第 249 天又访问了一次 → 续期到第 499 天。只要用户保持活跃，token 永不过期。
对比传统 Session：Session 依赖 Cookie + 内存。多台服务器部署时需要用 Spring Session + Redis 共享，本质也是 Redis 存 Session。我这个方案省掉了 Spring Session 的依赖，直接用 Redis Hash 存用户信息。
三、为什么选 UUID+Redis 而不是真正的 JWT？
JWT 的工作原理
登录 → 服务端签一个 JWT:
header: {"alg":"HS256","typ":"JWT"}
payload: {"userId":1010,"nickName":"user_xxx","exp":1720000000}
signature: HMAC-SHA256(header.payload, secret)
→ 三段 Base64 拼起来返回给客户端
后续请求:
客户端带 JWT → 服务端不查 Redis，直接用 secret 验签名 → 从 payload 取出 userId
为什么没用 JWT？
JWT 最大的问题是无法主动失效。
假设管理员封禁了用户 1010。如果用 JWT——服务端不存状态，JWT 的签名还是有效的，payload 里的 userId=1010 还是能解析出来。除非等到 JWT 里的 exp 过期，否则用户 1010 的请求一律放行。你不能"撤销"一个已经签发的 JWT。
用 Redis 存 Token 的方案：
// 封禁用户: 直接从 Redis 删掉 token
stringRedisTemplate.delete("login:token:" + token);
删完之后，用户的下一个请求走到 RefreshTokenInterceptor → Redis 查不到 → UserHolder 里没用户 → LoginInterceptor 返回 401。
一秒生效。
UUID Token + Redis 的优势
JWT（真正 JWT） UUID + Redis
主动失效 ❌ 只能等过期，或加黑名单 ✅ DEL 立即可踢下线
用户信息变更 要等到下次登录才更新 更新 Redis Hash 即可
服务端查存储 不需要 每次请求查 Redis
跨服务共享 天然支持（验签名即可） 需要共享 Redis
登出 复杂（令牌仍有效） DEL login:token:{uuid}
缺点
1. 每次请求都要查 Redis。 Redis 查一次省了序列化/反序列化的开销，但多了一次网络往返。高并发下可能成为瓶颈——但实际场景下 Redis 是本地内网的微秒级延迟，不是问题。
2. 不是真正无状态。 水平扩展时所有节点必须连同一个 Redis。Redis 挂了，所有用户都得重新登录。
3. token 就是 UUID 随机串。 UUID 足够长（32 位 hex）保证了不可预测性，但它本身没有任何意义——拿到 token 才能查到用户。不像 JWT 可以从 token 反解出 userId。
四、总结
"我这套方案准确说是 UUID Token + Redis 双层拦截器，不是严格的 JWT。JWT 的本质是无状态自包含凭证，我用的是有状态 Redis 凭证。选这个方案的理由是 需要主动失效能力——封号、踢人、修改用户信息需要立即可生效。代价是每次请求多一次 Redis 查询。双层拦截器用 order 分开控制，RefreshTokenInterceptor 负责识别用户（order=0，全部路径，永不拦截），LoginInterceptor 负责鉴权（order=1，排除公开路径，ThreadLocal 有用户就放行）。核心设计理念是职责分离——公开路径也能拿到用户信息做个性化，但不会因为没有登录就拦你。"
现状：没有 Redis 会话记忆
AiCustomerServiceImpl 第 128-131 行：
@Override
public void clearHistory(Long userId) {
log.debug("清除对话历史 userId={}", userId);
}
这是个空方法。只打了一行日志，什么也没干。
再看 chat() 方法——每次调用都重新构建请求体：
private Map<String, Object> buildRequestBody(String userName, String message) {
// 只有两条消息：system prompt + 当前用户消息
List<Map<String, String>> messages = List.of(
Map.of("role", "system", "content", SYSTEM_PROMPT),
Map.of("role", "user", "content", String.format("用户 %s 说：%s", userName, message))
);
body.put("messages", messages);
return body;
}
每次请求只发两条消息：system 角色 + 当前用户消息。上一轮的 user 消息和大模型的 assistant 回复都没有带过去。这是"无记忆"的单轮对话——用户说一句，AI 回一句，不记得上一句聊了什么。
之前写的那段 private final Map<Long, String> conversationHistory = new ConcurrentHashMap<>() 也被删掉了，因为根本没有用到。
所以你的问题很精准——对话记忆压缩策略的前提是"已经存了对话记忆"，而当前项目这一步还没落地。
如果做，实现思路
第一步：Redis 存储对话历史
// key: ai:history:{userId}
// value: List<Message> 的 JSON
// TTL: 30分钟（会话超时自动清理）
public String chat(Long userId, String message) {
// 1. 从 Redis 读历史
String historyKey = "ai:history:" + userId;
String historyJson = stringRedisTemplate.opsForValue().get(historyKey);
List<Map<String, String>> messages = parseHistory(historyJson);
// 2. 拼接当前消息
messages.add(Map.of("role", "user", "content", message));
// 3. 发请求
Map<String, Object> requestBody = new HashMap<>();
requestBody.put("messages", messages);
String reply = callQwen(requestBody);
// 4. 追加 assistant 回复到历史
messages.add(Map.of("role", "assistant", "content", reply));
// 5. 写回 Redis
stringRedisTemplate.opsForValue()
.set(historyKey, JSONUtil.toJsonStr(messages), 30, TimeUnit.MINUTES);
return reply;
}
这样三轮对话后，messages 会变成：
[system, user1, assistant1, user2, assistant2, user3, assistant3]
7 条消息，每条消息有角色、有内容。随着对话轮次增加，messages 数组越来越长，token 数成倍增长。
第二步：token 爆炸问题
Qwen 的计费方式是按 token 算的——输入和输出都算钱。messages 数组里的所有内容都是输入 token。10 轮对话后，messages 里 21 条消息，每次请求都要把这 21 条全部发给大模型重新处理一遍。
对话轮次 messages 数量 每轮输入 token（估算） 累计费用
第 1 轮 3 ~100 基准
第 5 轮 11 ~600 6x
第 10 轮 21 ~1200 12x
第 20 轮 41 ~2500 25x
第 20 轮的输入 token 是第 1 轮的 25 倍，响应时间越来越慢，费用越来越高。
第三步：记忆压缩策略
方案 A：滑动窗口 — 只保留最近 N 轮
private static final int MAX_HISTORY_TURNS = 5; // 保留最近5轮
// 发请求之前：
if (messages.size() > MAX_HISTORY_TURNS * 2 + 1) { // 1个system + N轮*2条
// 保留 system + 最近 N 轮
messages = new ArrayList<>(messages.subList(0, 1)); // system
messages.addAll(messages.subList(
messages.size() – MAX_HISTORY_TURNS * 2,
messages.size()
));
}
优点：实现简单。缺点：机械截断，旧的上下文完全丢失。用户说"刚才提到的那家店叫什么来着"——如果超过 5 轮，AI 已经忘了。
方案 B：摘要压缩 — 让大模型自己总结
private static final int COMPRESS_THRESHOLD = 8; // 超过8条就压缩
if (messages.size() > COMPRESS_THRESHOLD) {
// 取前 N 条"旧消息"，让大模型总结
List<Map<String, String>> oldMessages = messages.subList(1, messages.size() – 4);
String summary = callQwen(List.of(
Map.of("role", "system", "content": "请把以下对话历史压缩成一段摘要，保留用户的核心需求和已讨论的关键信息"),
Map.of("role", "user", "content": toText(oldMessages))
));
// 重建 messages: system + 摘要 + 最近3轮
messages = new ArrayList<>();
messages.add(Map.of("role", "system", "content": SYSTEM_PROMPT + "\\n\\n历史摘要: " + summary));
messages.addAll(messages.subList(messages.size() – 6, messages.size()));
}
效果对比：
方案 压缩比例 丢失什么 额外成本
滑动窗口 机械截断 旧的上下文完全丢失 零额外调用
摘要压缩 智能浓缩 细节，但保留关键意图 一次额外的 LLM 调用
方案 C：折中 — 滑动窗口 + 结构化记忆
在客服场景，真正需要保留的不是完整对话，而是关键信息。把这些关键信息单独存：
// ai:memory:{userId} → Hash
// intent: "查询优惠券" // 用户意图
// lastShop: "104茶餐厅" // 上次讨论的商铺
// preference: "火锅,人均100内" // 用户偏好
String memoryPrompt = String.format(
"用户当前意图: %s, 正在讨论的商铺: %s, 偏好: %s",
intent, lastShop, preference
);
messages.add(1, Map.of("role", "system", "content", memoryPrompt));
这样即使滑动窗口只保留了最近 3 轮，前面的关键信息（用户想找火锅店、预算 100 以内）仍然通过记忆系统保留下来。
设计选择
对于客服场景，我的建议是方案 A + 方案 C 组合：
滑动窗口保留最近 5 轮完整对话（保留最近上下文，让对话流畅）
关键信息单独存结构化的记忆（意图、讨论的商铺、偏好）
不做摘要压缩（多一次 LLM 调用会增加延迟，客服场景体验优先）
如果面试官追问"你当前项目怎么做的"——诚实说：
"当前项目的 AI 客服是单轮对话，每次请求只发 system prompt + 当前消息。会话记忆和压缩策略是产品化的下一步——设计思路是滑动窗口 + 结构化记忆，用 Redis Hash 存储意图、偏好等关键信息。"
