凌晨两点被报警电话叫醒，使用者投诉说和 AI 聊着聊着对话历史就丢了半截。我睡眼惺忪地翻日志，发现既没有报错也没有宕机——只是 Redis 里少了几条消息。老实说，直到打开监控看到那个接口的并发曲线。我才意识到：我们用 Redis 存会话历史的方式，在并发下根本不正确。

痛点回顾

• 数据丢失导致使用者体验骤降
• 调试耗时数小时却找不到根本原因
• 单纯加锁又会严重影响吞吐量
• 现有代码难以改动，迁移成本高昂

问题拆解

我们的场景很常见：每个使用者的对话记忆都存在 Redis 里。为了“省事”，早期设计直接用一个 String 键。把整个会话记录序列化成 JSON 数组。每次对话发生，后端就：

1. GET 整个 JSON
2. 反序列化。追加新消息
3. 序列化，SET 回去

这在 QPS 个位数的时候岁月静好。但最近增加使用者，前端又加了自动重试机制——同一个对话请求可能在极短时间内被触发两次。于是经典的 read-modify-write 竞态条件出现了：请求 A 和请求 B 都读到了同一个旧数组。各自加上自己的消息再写回，最终必然有一个请求的消息被覆盖丢失。

这就是「最终不一致」的根因——不是 Redis 不可靠，而是我们对 Redis 的原子性抱有超出它能力的期待。常规的单条 SET/GET 是原子的，但“读取-修改-写回”这种复合操作不是。直接加分布式锁当然能解决，但会严重拉低吞吐。而且引入锁就代表着要处理死锁、锁超时等一系列麻烦。我们必须找一个锁‑free 的原子操作。

方案设计与选择

MVP 备选方案：

• : 每条消息压入列表。原子操作，自然追加，不会覆盖。但读取历史需要 LRange 全量拉取，对大会话不友好。
• : 适合消息队列场景，可持久化；但引入使用者组后读取变复杂，对接成本高。按理说，
• : 在服务端执行一段原子脚本。直接在 JSON 数组上追加元素并写回。单次网络往返，无锁，完全原子； 唯一代价是把 JSON 逻辑写进 Lua，但 Redis 内置 cjson 库支持。

*最终决定： Lua 脚本方案最小改动—仍使用 String 存整个会话；业务层接口保持不变，只把读‑改‑写替换为单条 EVAL 命令；可渐进迁移、老数据兼容，

主要实现细节

第一步先这方面，用 Pytest 复现并发 Bug

*为什么先复现？话说回来，* 因为只有可复现的问题才能谈一致性与修复。

import asyncio
import json
import pytest
import redis.asyncio as aioredis
@pytest.mark.asyncio
async def test_concurrent_append_causes_missing_messages:
"""复现 read-modify-write 导致的消息丢失"""
r = aioredis.from_url
user_key = "user::history"
# 初始化空历史
await r.set)
async def bad_append:
raw = await r.get
history = json.loads if raw else
history.append
await r.set)
tasks =
await asyncio.gar
final_raw = await r.get
final_history = json.loads
# 很可能小于 10。因为发生了覆盖
assert len == 10,f"丢失消息！期望 10 条，实际 {len}"
}

接下来的观点是。编写原子追加的 Lua 脚本

*脚本做什么？话说回来，* 初始化、解析、追加、返回新长度。

-- lua/append_history.lua
local key = KEYS
local new_message = ARGV
local raw = redis.call
local history = {}
if raw and raw ~= '' n
history = cjson.decode
end
table.insert
local new_raw = cjson.encode
redis.call
return #history -- 返回当前总条数，可用于监控或验证
}

：Python 封装原子操作

import redis.asyncio as aioredis
LUA_APPEND_SCRIPT = """
local key = KEYS
local new_message = ARGV
local raw = redis.call
local history = {}
if raw and raw ~= '' n
history = cjson.decode
end
table.insert
local new_raw = cjson.encode
redis.call
return #history
"""
class HistoryStore:
def __init__:
self.r = r
self._script = r.register_script
async def append_message -> int:
"""原子追加一条消息。返回当前总条数"""
return await self._script
}

第四步这方面，验证修复是否有效

@pytest.mark.asyncio
async def test_concurrent_append_with_lua_consistency:
r = aioredis.from_url
store = HistoryStore
user_key ="user::history_v2"
await r.set
tasks =
results=await asyncio.gar
final_raw=await r.get
final_history=json.loads
assert len==10,"Lua 方案不应该丢失任何消息"
assert max==10,"每个调用返回最大长度应为总数"

*结果： 所有并发请求都成功追加，没有任何遗漏；该测试已成为 CI 持续守护。

踩坑记录

• PITFALL: .register_script 在异步集群客户端下抛异常 No way to dispatch this command. 再看原因。RedisCluster 会根据 key 路由到不同节点，而脚本注册是全局挂载，需要在每个节点执行 SCRIPT LOAD。从解决来看，
  • 每次执行直接使用 EVAL 把脚本体传进去。
  • 在启动阶段手动通过每个集群节点底层连接执行 SCRIPT LOAD。我们选择，因为脚本体小且一次性加载足够满足性能需求。
  
  
• PITFALL: .decode_responses=True> Lua 返回字符串解码错误
从原因来看。Python 客户端将 Lua 返回字符串 解码，接下来我们误用了二次 json.loads,导致转义混乱。从解决来看，
• => 在 Lua 中仅返回整数长度或标识符。不返回完整 JSON，
• => Python 一直通过普通 GET+json.loads 获取历史，以避免转义问题。
- **Tip**：只用脚本返回状态码或计数就可以完成验证，无需返回完整数据结构。

效果验证 & 性能指标

P99 延迟增幅仅 .3 ms 与 Lua 原子追加相比，同时彻底消除了数据覆盖风险；旧会话无需迁移，仅代码行变动不到十行。

可直接用的代码/工具 & 快速启动教程

If you want to verify your own session storage against concurrent writes right away:

# conftest.py - pytest fixture for redis client
# ---------------------
import pytestasyncio
import redis.asyncio as aioredis
@pytestasyncio.fixture
async def redisclient:
r=aioredis.fromurl
yield r
await r.close
# ---------------------
# 将上述 Lua 脚本及 HistoryStore 类粘贴到项目中，接下来运行以下测试。# tests/testconcurrency.py
from conftest import redisclient
from yourmodule import HistoryStore
\
async def testluaconsistency:
store=HistoryStore
user=\"user::test\"
await redisclient.set
tasks=
await asyncio.gar
hist=json.loads)
assert len==50
\tprint
\t\t\t\t\t\t\t` ` ` ` ` \t\t` `