企业官网建设流程全解析

1. 项目概述：Redis 并没有“接入 AI”，但它的生态正在被 AI 彻底重构

最近刷到“Redis 已正式接入 AI！”这个标题，第一反应是点进去看——结果发现不是 Redis 官方发布了带 AI 模块的 8.0 版本，也不是 Salvatore Sanfilippo（Redis 之父）在博客里宣布“今天起 Redis 开始思考”。事实上，Redis 官网、GitHub Release Notes、官方博客和 RedisConf 2024 议题列表中，没有任何一条关于“AI 内置”或“原生大模型支持”的公告。那这个标题到底在说什么？它背后的真实信号，比字面意思重要十倍。

简单说：这不是 Redis 变成了 AI，而是 AI 工程师们正把 Redis 当作构建 AI 系统的“隐形脊椎”——从记忆存储、工具调用调度、状态持久化，到多 Agent 协作的协调中枢。热搜词里反复出现的MCP（Model Control Protocol）、agent-skills、Python，恰恰指向一个正在快速落地的技术范式：以 Redis 为底层状态总线的轻量级 AI Agent 架构。它不依赖昂贵的向量数据库集群，也不需要定制化消息中间件，而是用 Redis 的 Pub/Sub、Stream、JSON、Lua 原子操作和极低延迟，把 LLM 的“思考流”和“动作流”稳稳托住。

我过去三年做过 7 个生产级 AI Agent 项目，其中 5 个都选了 Redis 作为核心状态层。不是因为“它支持 AI”，而是因为它解决了 AI 工程中最痛的三个问题：状态一致性难保证、多步任务断点续跑难、工具调用链路不可追溯。比如一个客服 Agent 需要查订单 → 调物流 API → 生成摘要 → 发短信，这四步如果全靠内存或文件暂存，一次进程崩溃就前功尽弃；而用 Redis Stream 存每一步的输入/输出+时间戳+trace_id，重启后直接从失败点 resume，实测平均恢复耗时 < 80ms。这才是标题里“接入 AI”的真实含义——Redis 成为了 AI 工作流的“数字骨骼”。

适合谁读？如果你正在用 LangChain/LlamaIndex 做 Agent，却卡在“为什么每次重试结果都不一样”“怎么让多个 Agent 共享上下文”“调试时根本不知道哪一步挂了”，那你不是缺模型，而是缺一套可靠的状态基座。本文不讲大模型原理，只拆解：如何用 Redis 的 4 类数据结构（String/Hash/List/Stream），配合 Python 和 MCP 协议，搭出一个可审计、可回滚、可水平扩展的 AI Agent 运行时环境。所有代码、配置、压测数据均来自我们团队在电商客服、专利辅助、金融研报三个场景的线上系统。

2. 技术架构解析：为什么是 Redis，而不是 PostgreSQL 或 Kafka？

2.1 核心矛盾：AI Agent 的“状态饥渴” vs 传统存储的“响应迟滞”

先说结论：AI Agent 对存储的需求，和 Web 应用截然不同。Web 应用要的是 ACID、复杂查询、事务回滚；AI Agent 要的是毫秒级读写、原子性状态更新、天然的消息队列能力、以及极简的运维成本。我们曾对比过三种方案：

关键差异在“状态一致性”。举个真实例子：Agent 要执行“查库存→扣减→发通知”三步，若用 PostgreSQL，第二步扣减失败，第一步查的库存值已过期，必须回滚并重查；而用 Redis Stream，我们把三步封装成一个XADD agent:workflow:order_12345 * step:check_stock value:120→step:deduct_stock value:115→step:send_notify status:success，每条消息带唯一 ID 和时间戳，失败时只需XREAD COUNT 1 STREAMS agent:workflow:order_12345 $就能精准定位断点，无需任何锁或事务。

提示：别被“Redis 是缓存”这个旧认知绑架。Redis 6.0+ 的 AOF + RDB 持久化策略，配合appendfsync everysec，在我们的压测中（10K QPS 持续写入），数据丢失概率 < 0.0001%，远低于 Kafka 的默认配置。它早已不是“掉电即丢”，而是“高性能持久化引擎”。

2.2 Redis 的四大核心能力如何匹配 AI Agent 需求

（1）Stream：天然的工作流消息总线

Stream 不是简单的队列，而是带时间序、可回溯、可分组消费的持久化日志。AI Agent 的每一步动作（LLM 输出、Tool 调用、人工干预）都是一条 Stream 消息，格式如下：

{ "task_id": "order_12345", "step": "generate_summary", "input": {"order_id": "12345", "items": ["iPhone15", "AirPods"]}, "output": "用户订单含2件商品，总价¥12,999，预计3天内发货", "timestamp": 1717023456.789, "trace_id": "tr-8a3b9c1d" }

优势在于：

• 消费组（Consumer Group）：多个 Agent 实例可组成 group 并行处理不同 task_id，避免重复消费；
• Pending Entries：自动记录未确认消息，宕机后自动重投；
• 历史回溯：XRANGE agent:steps - + COUNT 100直接拉取最近 100 步完整链路，调试时比翻日志快 10 倍。

（2）JSON 数据类型：动态 Schema 的 Agent 状态存储

传统 Hash 表对嵌套结构支持差，而JSON.SET/JSON.GET让 Agent 的运行时状态（如 memory、tool_history、user_context）可直接存为树形结构：

# 存储 Agent 的长期记忆（类似 RAG 的 context） JSON.SET agent:memory:user_789 '{"last_order_date":"2024-05-20","preferred_language":"zh-CN","loyalty_level":"gold"}' # 更新某次会话的临时状态 JSON.SET agent:session:sess_abcd '{"current_step":"awaiting_payment","retry_count":2,"timeout_at":1717024567}'

实测对比：同样存 10K 条嵌套 JSON，Redis JSON 比 PostgreSQL JSONB 查询快 3.7 倍（因内存索引 + 无 SQL 解析开销），且无需建表。

（3）Pub/Sub：实时事件广播与跨 Agent 协作

当 Agent A 完成“风控审核”，需通知 Agent B “开始发货”，用PUBLISH channel:risk_approved order_12345比轮询数据库高效得多。我们甚至用它实现“人工接管”：客服人员在后台点击“接管对话”，前端 WebSocket 直接收到{"event":"handover","session_id":"sess_abcd"}，Agent 自动暂停并移交控制权。

（4）Lua 脚本：原子性状态变更的终极保险

AI Agent 最怕“检查-执行”竞态。例如判断库存是否充足再扣减，用两条命令可能被并发请求打穿。而 Lua 脚本保证整个逻辑在服务端原子执行：

-- inventory_check_and_deduct.lua local stock = redis.call('JSON.GET', KEYS[1], '$.stock') if tonumber(stock) >= tonumber(ARGV[1]) then redis.call('JSON.SET', KEYS[1], '$.stock', tostring(tonumber(stock) - tonumber(ARGV[1]))) return 1 else return 0 end

调用：redis-cli --eval inventory_check_and_deduct.lua agent:product:iphone15 , 5。全程无网络往返，P99 延迟 < 1.2ms。

2.3 为什么不是其他 NoSQL？MongoDB / Elasticsearch 的短板在哪？

• MongoDB：文档更新需 BSON 解析，复杂查询（如“找出所有 retry_count > 3 的 session”）在 100W+ 文档时延迟飙升至 200ms+，而 RedisSCAN+JSON.GET组合可在 5ms 内完成同条件过滤；
• Elasticsearch：擅长全文检索，但对精确匹配（如task_id == "order_12345"）反而不如 Redis 的 O(1) Hash 查找；且其近实时（NRT）特性导致状态“最终一致”，AI Agent 需要“立刻一致”；
• DynamoDB：AWS 生态友好，但冷启动延迟高（首次请求 100~300ms），且按读写容量计费，在 Agent 场景下成本比自建 Redis 高 4~6 倍。

注意：这不是贬低其他数据库，而是强调场景适配。就像螺丝刀不能代替扳手——Redis 在 AI Agent 的“状态流”场景中，就是那把最趁手的螺丝刀。

3. 实操部署：从零搭建 Redis + Python + MCP 的 AI Agent 运行时

3.1 环境准备：最小可行集群与安全加固

别用docker run redis就完事。生产环境必须考虑三点：数据持久化、访问控制、资源隔离。我们采用 Redis 7.2（2024 年最新稳定版），配置如下：

# redis.conf 关键参数（非默认值） port 6379 bind 0.0.0.0 # 仅内网访问，前置 Nginx 或 Security Group 限制 requirepass your_strong_password_here # 必须设密码！ maxmemory 4gb maxmemory-policy allkeys-lru # 避免 OOM，LRU 自动淘汰 appendonly yes # 启用 AOF 持久化 appendfilename "appendonly.aof" appendfsync everysec # 平衡性能与安全性 save "" # 关闭 RDB，AOF 更可靠 protected-mode yes

部署步骤（以 Ubuntu 22.04 为例）：

# 1. 下载编译（比 apt 安装新 2 个大版本） wget https://download.redis.io/releases/redis-7.2.5.tar.gz tar xzf redis-7.2.5.tar.gz && cd redis-7.2.5 && make && sudo make install # 2. 创建配置目录 sudo mkdir -p /etc/redis /var/lib/redis /var/log/redis sudo cp redis.conf /etc/redis/redis.conf sudo chown -R redis:redis /var/lib/redis /var/log/redis # 3. 创建 systemd 服务 cat > /etc/systemd/system/redis-server.service << 'EOF' [Unit] Description=Redis In-Memory Data Store After=network.target [Service] Type=notify User=redis Group=redis ExecStart=/usr/local/bin/redis-server /etc/redis/redis.conf ExecStop=/usr/local/bin/redis-cli -h 127.0.0.1 -p 6379 -a your_strong_password_here shutdown Restart=always RestartSec=10 LimitNOFILE=65535 [Install] WantedBy=multi-user.target EOF sudo systemctl daemon-reload && sudo systemctl enable redis-server && sudo systemctl start redis-server

验证是否生效：

# 检查状态 systemctl status redis-server # 应显示 active (running) # 测试连接与密码 redis-cli -h 127.0.0.1 -p 6379 -a your_strong_password_here ping # 返回 PONG # 查看内存使用 redis-cli -h 127.0.0.1 -p 6379 -a your_strong_password_here info memory | grep -E "(used_memory_human|maxmemory_human)"

实操心得：很多团队跳过密码设置，结果被扫描器爆破后挖矿。Redis 默认端口 6379 是黑客最爱的目标之一。我们曾监控到某测试环境 24 小时内被尝试登录 17,321 次——设密码是最小成本的安全底线。

3.2 Python SDK 选型：redis-py 还是 aredis？异步支持怎么选？

AI Agent 天然是 I/O 密集型（频繁调用 LLM API、外部工具），必须用异步。redis-py4.0+ 已原生支持 async，无需aredis（已归档）。安装与初始化：

pip install redis==4.6.0 aiohttp python-dotenv

核心连接池配置（避免连接泄漏）：

import redis.asyncio as redis from redis.asyncio.connection import ConnectionPool # 生产环境必须用连接池，而非每次 new client pool = ConnectionPool( host="127.0.0.1", port=6379, db=0, password="your_strong_password_here", max_connections=100, # 根据并发量调整 decode_responses=True, # 自动 decode bytes to str health_check_interval=30, # 每30秒ping一次保活 ) redis_client = redis.Redis(connection_pool=pool)

为什么不用redis.from_url()？因为无法精细控制health_check_interval和max_connections，高并发下易出现ConnectionError: Error 113 connecting to 127.0.0.1:6379. No route to host.—— 这是连接池耗尽的典型表现。

3.3 MCP 协议对接：用 Redis 实现 Model Control Protocol 的核心通道

MCP（Model Control Protocol）是近期兴起的 AI Agent 通信标准，目标是让不同厂商的 LLM、Tool、Memory 模块像 USB 设备一样即插即用。其核心是定义了一套 JSON-RPC 风格的 message schema，而 Redis Stream 正是承载这些 message 的理想载体。

我们定义两个关键 Stream：

• mcp:requests：存放所有 LLM 请求（由 Agent 发起）
• mcp:responses：存放所有 LLM 响应（由推理服务写入）

Agent 发送请求的 Python 代码：

import json import uuid from datetime import datetime async def send_mcp_request( model_name: str, prompt: str, tools: list = None, session_id: str = None ): request_id = str(uuid.uuid4()) message = { "jsonrpc": "2.0", "method": "call_llm", "params": { "model": model_name, "prompt": prompt, "tools": tools or [], "session_id": session_id or f"sess_{int(datetime.now().timestamp())}" }, "id": request_id } # 写入 Stream，自动带时间戳 await redis_client.xadd( "mcp:requests", {"data": json.dumps(message)}, id="*" # 服务端生成 ID ) return request_id # 示例调用 req_id = await send_mcp_request( model_name="qwen2-7b", prompt="总结以下订单：订单号 ORD-7890，商品 iPhone15 1台，AirPods 2副，总价 ¥12,999", tools=["get_order_status", "get_shipping_info"] )

推理服务（如 vLLM 或 Ollama）监听并消费：

# 推理服务端伪代码 async def mcp_consumer(): last_id = "$" # 从最新开始 while True: # 阻塞式读取，超时 5 秒 messages = await redis_client.xread( streams={"mcp:requests": last_id}, count=1, block=5000 ) if not messages: continue stream_name, entries = messages[0] for entry_id, entry_data in entries: try: req = json.loads(entry_data[b"data"]) # 调用本地 LLM response = await call_local_llm(req["params"]["prompt"], req["params"]["tools"]) # 写入响应 Stream await redis_client.xadd( "mcp:responses", { "request_id": req["id"], "response": json.dumps(response), "timestamp": str(datetime.now().timestamp()) } ) # 确认消费 await redis_client.xack("mcp:requests", "mcp_group", entry_id) last_id = entry_id except Exception as e: logging.error(f"MCP processing failed: {e}") # 失败消息进入 dead-letter queue await redis_client.xadd("mcp:dlq", {"error": str(e), "entry": str(entry_data)})

关键设计点：

• 消费组mcp_group：允许多个推理服务实例负载均衡；
• Dead-Letter Queue（DLQ）：失败消息不丢弃，便于人工排查；
• request_id 双向绑定：Agent 通过XREAD监听mcp:responses中request_id == req_id的消息，实现请求-响应匹配。

3.4 Agent 核心状态管理：用 JSON + Stream 构建可回滚的执行引擎

一个健壮的 Agent 必须支持：断点续跑、人工干预、状态快照、多步回退。我们用 Redis 的组合拳实现：

（1）状态存储结构设计

# Agent 状态存于 JSON，key 格式：agent:state:{session_id} # 示例： """ { "session_id": "sess_abcd", "user_id": "user_789", "current_task": "process_return_request", "steps": [ {"step": "parse_request", "status": "success", "ts": 1717023456}, {"step": "verify_eligibility", "status": "failed", "ts": 1717023462, "error": "order_not_found"} ], "memory": {"last_order_id": "ORD-7890", "return_reason": "damaged"}, "retry_count": 2, "timeout_at": 1717024567 } """

（2）原子性状态更新函数

async def update_agent_state(session_id: str, updates: dict): """ 原子更新 Agent 状态，避免并发覆盖 updates: {"steps": [...], "memory": {...}, "retry_count": 3} """ lua_script = """ local state = redis.call('JSON.GET', KEYS[1]) if not state then return 0 end local new_state = cjson.decode(state) for k,v in pairs(ARGV[1]) do if type(v) == 'table' then new_state[k] = cjson.decode(v) else new_state[k] = v end end redis.call('JSON.SET', KEYS[1], '$', cjson.encode(new_state)) return 1 """ # 将 dict 转为 JSON 字符串传入 Lua await redis_client.eval( lua_script, 1, f"agent:state:{session_id}", json.dumps(updates) ) # 使用示例：添加失败步骤 await update_agent_state( session_id="sess_abcd", updates={ "steps": [ {"step": "verify_eligibility", "status": "failed", "ts": int(time.time()), "error": "order_not_found"} ] } )

（3）断点续跑逻辑

async def resume_from_failure(session_id: str): """从最后失败的步骤继续执行""" state = await redis_client.json().get(f"agent:state:{session_id}") if not state or "steps" not in state: return None # 找到最后一个 failed 步骤 failed_steps = [s for s in state["steps"] if s.get("status") == "failed"] if not failed_steps: return None last_failed = failed_steps[-1] # 根据 step 名称决定重试逻辑 if last_failed["step"] == "verify_eligibility": # 重试前先刷新订单数据 order_data = await fetch_order_from_db(last_failed.get("order_id", "")) await update_agent_state(session_id, {"memory": {"order_data": order_data}}) return "verify_eligibility" elif last_failed["step"] == "call_api": # 检查是否超重试次数 if state.get("retry_count", 0) >= 3: await notify_human_review(session_id) return None return "call_api" return None

这套设计让 Agent 在 Kubernetes Pod 重启后，3 秒内就能恢复到故障前状态，比基于数据库的方案快一个数量级。

4. 场景实战：电商客服 Agent 的 Redis 状态流全链路解析

4.1 业务需求与技术挑战还原

客户投诉：“我昨天申请退货，到现在没回复！”
客服 Agent 需完成：

1. 解析用户消息，提取订单号（NLU）
2. 查询订单状态（调用 ERP API）
3. 判断是否符合退货政策（规则引擎）
4. 若符合，生成退货单（调用 WMS API）
5. 向用户发送确认消息（短信/APP Push）

传统方案痛点：

• 步骤 2 失败（ERP 临时不可用），整个流程中断，用户需重新发起；
• 步骤 4 调用 WMS 超时，但步骤 3 已修改本地状态，导致“政策通过但未生成单”；
• 无法向主管展示“这个投诉处理了哪几步，卡在哪”，只能查分散的日志。

4.2 Redis 驱动的全链路状态流设计

我们为该 Agent 定义 5 个 Redis Key：

• agent:state:sess_xyz：JSON 存储当前会话状态
• agent:steps:sess_xyz：Stream 记录每步详细输入输出
• agent:cache:order_ORD-12345：Hash 缓存订单详情（TTL 5min）
• agent:lock:order_ORD-12345：String 实现订单级锁（防止并发处理同一订单）
• agent:metrics:hourly：Sorted Set 统计每小时成功率（score=timestamp）

第一步：消息解析与锁抢占

async def handle_user_message(session_id: str, user_text: str): # 1. 尝试获取订单锁（避免同一订单被多个会话同时处理） lock_key = f"agent:lock:order_{extract_order_id(user_text)}" lock_acquired = await redis_client.set(lock_key, "locked", ex=300, nx=True) # 5分钟锁 if not lock_acquired: await send_to_queue("high_priority", session_id, user_text) # 排队 return # 2. 初始化状态 initial_state = { "session_id": session_id, "user_text": user_text, "steps": [], "memory": {}, "start_time": time.time() } await redis_client.json().set(f"agent:state:{session_id}", "$", initial_state) # 3. 写入解析步骤 await log_step(session_id, "parse_message", {"input": user_text}, {"order_id": "ORD-12345"})

第二步：带缓存的订单查询

async def get_order_details(order_id: str) -> dict: cache_key = f"agent:cache:order_{order_id}" cached = await redis_client.hgetall(cache_key) if cached: return {k.decode(): v.decode() for k, v in cached.items()} # 缓存未命中，调用 ERP order_data = await call_erp_api(order_id) # 写入缓存，Hash 结构便于部分更新 await redis_client.hset(cache_key, mapping={k: str(v) for k, v in order_data.items()}) await redis_client.expire(cache_key, 300) # 5分钟 return order_data

第三步：原子性政策校验与状态更新

# Lua 脚本确保“读-判-写”原子性 policy_check_lua = """ local order = redis.call('HGETALL', KEYS[1]) local status = order['status'] local amount = tonumber(order['amount']) local days_since_order = tonumber(ARGV[1]) if status == 'shipped' and amount <= 10000 and days_since_order <= 7 then redis.call('JSON.SET', KEYS[2], '$.policy_result', '"eligible"') redis.call('JSON.SET', KEYS[2], '$.steps', cjson.encode({{step='check_policy', status='success', ts=ARGV[2]}})) return 1 else redis.call('JSON.SET', KEYS[2], '$.policy_result', '"ineligible"') redis.call('JSON.SET', KEYS[2], '$.steps', cjson.encode({{step='check_policy', status='failed', ts=ARGV[2], reason='policy_violation'}})) return 0 end """ # 调用 result = await redis_client.eval( policy_check_lua, 2, f"agent:cache:order_{order_id}", f"agent:state:{session_id}", int(time.time()) - order_data["created_at_ts"], # 天数计算 int(time.time()) )

第四步：失败自动降级与人工介入

async def on_step_failure(session_id: str, step_name: str, error: str): # 1. 更新状态 await update_agent_state(session_id, { "steps": [{"step": step_name, "status": "failed", "error": error, "ts": int(time.time())}] }) # 2. 若是关键步骤（如 call_wms），触发人工 if step_name in ["call_wms", "send_sms"]: await redis_client.lpush("agent:human_queue", json.dumps({ "session_id": session_id, "step": step_name, "error": error, "state_snapshot": await redis_client.json().get(f"agent:state:{session_id}") })) # 3. 发送企业微信告警 await send_work_wechat_alert(f"Agent 失败：{session_id}，步骤 {step_name}，错误 {error}")

4.3 性能压测与线上监控数据

我们在阿里云 4C8G ECS 上部署 Redis 7.2，用redis-benchmark和自研压测脚本测试：

线上真实数据（电商客服系统，日均 2.3M 次 Agent 调用）：

• 平均端到端延迟：1.8s（含 LLM 调用 1.2s，Redis 操作 0.3s，网络 0.3s）
• 断点续跑成功率：99.97%（全年仅 3 次因 AOF 写满磁盘导致状态丢失）
• 人工介入率：从传统方案的 12.3% 降至 1.8%，主要集中在“WMS 接口超时”场景
• 运维成本：Redis 集群 3 节点，月均费用 ¥860，仅为同等 Kafka 集群的 1/5

实操心得：压测时最容易忽略的是AOF rewrite 的阻塞影响。Redis 在 AOF 文件过大时会 fork 子进程重写，此时主线程可能卡顿 100~300ms。解决方案：

1. 设置auto-aof-rewrite-percentage 100（增长 100% 才重写）；
2. auto-aof-rewrite-min-size 64mb（至少 64MB 才触发）；
3. 监控aof_current_size和aof_base_size，在低峰期手动BGREWRITEAOF。

5. 常见问题与避坑指南：那些只有踩过才懂的 Redis + AI 组合陷阱

5.1 “Redis 内存暴涨”问题：不是内存泄漏，而是数据结构误用

现象：Agent 运行 2 小时后，Redis 内存从 1GB 涨到 8GB，INFO memory显示used_memory_dataset_perc达 95%。
根因：误用 List 存储无限增长的会话历史。例如：

# 错误示范：用 LPUSH 无限追加 redis_client.lpush(f"agent:history:{session_id}", json.dumps({"role": "user", "content": "hello"})) # 1000 次后，List 有 1000 个元素，且无法自动过期

正确方案：

• 用 Stream 替代 List：Stream 支持XTRIM自动裁剪，XTRIM agent:steps MAXLEN 1000保留最近 1000 条；
• 用 TTL 控制 JSON 状态：redis_client.json().set(key, "$", state, ex=3600)设 1 小时过期；
• 定期清理死信队列：redis_client.xtrim("mcp:dlq", maxlen=10000)。

5.2 “Stream 消费延迟高”问题：消费组堆积与心跳机制缺失

现象：XINFO GROUPS mcp:requests显示pending消息达 5000+，idle时间超过 1 小时。
原因：消费组成员（consumer）宕机后未主动退出，Redis 仍认为它在线，导致消息无法分配给其他实例。

解决方案：

1. 客户端主动 heartbeat：消费端每 30 秒发XGROUP SETID mcp:requests mcp_group consumer_1 0-0重置 idle；
2. 服务端自动清理：XINFO CONSUMERS mcp:requests mcp_group查看 idle > 300s 的 consumer，用XGROUP DELCONSUMER删除；
3. 监控告警：redis-cli --stat监控xpending，> 1000 时触发 PagerDuty。

5.3 “JSON 查询慢”问题：别在 JSON 字段上做范围查询

现象：JSON.GET agent:state:sess_abcd $.memory.last_order_id很快，但JSON.GET agent:state:* $.memory.last_order_id却超时。
原因：Redis JSON 不支持全局索引，SCAN扫描所有 key 再逐个JSON.GET，O(n) 复杂度。

正确做法：

• 建立二级索引：用 Sorted Set 存order_id → session_id映射，ZADD index:order_id 0 sess_abcd；
• 用 Hash 分片：agent:state:ORD-12345:sess_abcd，通过订单号路由，避免全量扫描；
• 接受最终一致性：对非实时场景，用 Redis + Elasticsearch 组合，Redis 做热数据，ES 做全量检索。

5.4 “Lua 脚本超时”问题：Redis 的 5 秒硬限制与规避策略

现象：复杂 Lua 脚本执行超时，Redis 返回BUSY Redis is busy running a script。
原因：Redis 单线程执行 Lua，且默认lua-time-limit 5000（5 秒），超时则中断。

规避方法：

• 拆分逻辑：将长脚本拆为多个短脚本，用EVALSHA复用 SHA1；
• 异步化：耗时操作（如 HTTP