企业官网建设流程全解析

1. 项目概述：Agent-Reach 到底在做什么

1.1 “任务永远 pending”这种痛，你可能也遇到过

Agent-Reach 这个名字你一眼看不出来它干嘛的，“Agent”是代理，“Reach”是触达，我把它当成一个“代理触达与调度管理”的轻量级项目来设计。背景是这样的：我管理的十几台服务器上跑着各种各样的小代理，有抓网页的、有处理队列的、有定期调外部接口做分析统计的，反正在不同宿主上各干各的。问题在于，当我把任务通过 Redis 直接乱发一通的时候，根本不知道哪些任务被哪个代理接走了，也不知道某些代理是不是早就断连了，于是系统里堆了一堆永远处于 pending 状态的任务。日志里天天飘着 timeout，但要定位是网络问题还是节点挂了，只能一台台上去敲命令，效率极低。

这个问题的本质不是消息队列该用哪个，而是缺少一层“代理触达状态”的管理。我需要的不是把消息塞进某个队列就算完事，而是搞清楚每一个任务能不能真的被一个活着的、有余力的代理接住，并且这个接住的过程要有记录、有反馈。Agent-Reach 就是从这个问题里长出来的，它把心跳感知、可达性评分、路由决策、事件记录这四件事拧成了一根完整的链路。

1.2 Agent-Reach 能做什么、适合谁

Agent-Reach 核心能力可以概括成四点：第一，持续感知代理存活状态，每个代理每隔几秒上报一次心跳，调度端据此判断在线离线；第二，为每个代理计算实时的可达性评分，评分综合考虑延迟、任务积压、历史成功率；第三，任务提交时按评分把任务路由给最合适的代理，避免“发给已经挂了或忙到飞起的节点”；第四，把整个调度过程的当前状态推送到可视化面板上，让你一眼看清任务现在在哪、卡在哪。整个系统不绑定具体业务，是一个独立的调度触达层，你可以把它当成多代理编排场景里的“中间件”来用。

这套东西适合几类人：在多 Agent 系统里跑了很多独立代理，但是管不住它们的学生和开发者；在边缘设备或内网环境里需要做轻量级调度的工程师；以及想把“消息投递 + 状态追踪”做成一个透明可观测模块的团队。如果你是第一次接触这个概念，也不用慌，下面的章节我会从选型、数据结构、代码实现一路讲下来，所有关键参数都是实测之后得出的，可以直接抄去用。

2. 整体方案设计与技术选型

2.1 为什么不用现成消息队列，而是自己搭一个触达层

先讲选型前的心态。做之前我翻了不少现成方案，包括 Celery、RQ、NATS，甚至考虑过直接用 Redis 的 Pub/Sub 开干，但梳理完需求之后发现，通用消息队列解决的只是“消息怎么从 A 到 B”的问题，而我要解决的其实是“从 A 到哪个 B、B 是不是还在、选 B 的理由是什么”这一串决策链的问题，这两者表面相似，本质完全不同。

Celery 确实是很多人的第一反应，任务重试、worker 心跳、定时任务全都有，但它对代理状态的抽象非常弱。Celery 里你关心的是 queue 和 worker，它虽然有心跳，但 worker 的健康状态并不会参与路由决策，任务投递后由 broker 按抢占或预取机制分配，你无法提前指定“只有这台机器上的 agent_07 可以接这个任务”。要达到 Agent-Reach 的效果，你只能硬生生在业务代码里把 route 逻辑层叠出来，结果你的业务层跟调度层越缠越紧，改路由逻辑还要动业务代码，维护成本直线上升。

Redis Pub/Sub 就更不用说了，消息发出去之后不落盘，消费者不在线就消失，这跟重试和审计需求天生冲突。Redis Streams 的出现解决了一部分问题，它是一条结构化的消息日志，消息可以持久化，消费者可以用消费组来读，读完之后要手动 ack，这基本就是轻量版 Kafka 的语义。不过 Streams 自己也不懂“代理状态”，它只负责存储，所以我还是得在上层实现心跳和评分。最后我采用的方案是：Redis Streams 当作任务的核心投递通道，路由和状态感知完全自己写，两边各管一摊，职责清晰。

2.2 技术栈组成与各模块分工

Agent-Reach 的技术栈相当收敛，就四个组件：Python 3.11、FastAPI、Redis 7、SQLite。前端控制台是纯静态 HTML 加原生 JavaScript，不引框架。下面把每个组件为什么出现在这里说清楚。

FastAPI 是所有 HTTP 入口的容器：代理注册接口、心跳接口、任务提交接口、WebSocket 推送接口都挂在它上面。选它而不是 Flask，原因是心跳和 WebSocket 并发场景下异步模型有天然优势，服务端可以对多个代理的连接做事件循环复用。尤其是一堆代理同时打心跳的时候，Flask 的同步线程模型容易出现可预见的延迟峰值，FastAPI 在这种情况下表现稳定得多。开发时自带 swagger 文档也顺手，调试接口不用额外工具。

Redis Streams 承担任务投递。它跟普通 List 最大的区别是结构上自带消息 ID、消费者组、ack 语义、pending 列表，正好覆盖“任务不能被丢、消费要确认、失败要能重新投递”这一套需求。配合XREADGROUP、XAUTOCLAIM、XPENDING几个命令，可以做很成熟的消费管理。需要说明的是这不是唯一选择，如果你已经在用 Kafka，直接用 Kafka 也没问题；如果只有一台机器，Redis Streams 轻得刚刚好。

SQLite 存储代理状态。因为数据量真的不大：几十个代理，每个代理一条记录，外加每条心跳一行事件，容量完全够用。SQLite 没有独立进程，随应用进程启动，天然适合嵌入式部署，不需要额外维护一个数据库服务。如果未来代理数量超过几百，可以平移到 PostgreSQL，表结构基本不用改。

WebSocket 是一条实时推送通道，把心跳变化、路由决策、任务状态这些事件一条不漏地推给控制台。没有选轮询，因为调度事件往往集中在几秒内发生，轮询既浪费带宽又做不到真正的实时。我把推送设计成“全量快照 + 增量事件”，连接建立时先推一次全量，之后只推增量，前端维护一个状态对象持续更新。

2.3 数据模型：调度状态如何抽象

讲抽象之前，先定义 Agent-Reach 里几个不可再拆的原子概念，它们是整个系统的地基。

第一个是代理（Agent），它代表一个可以接受任务的执行单元，有自己的 agent_id、注册时带的元信息、最近一次心跳时间、当前积压任务数、历史成功率统计。第二个是心跳事件（HeartbeatEvent），代理每次上报心跳都会生成一条记录，里面包含时间戳、健康指标，它是判断存活和计算评分的原始数据。第三个是任务事件（TaskEvent），它记录一个任务从提交、路由、触达、执行到完成的全过程，每经过一个环节就追加一条事件，事件带上当时的打分快照。第四个是任务实体（Task），它就是业务层的任务对象，携带唯一的 task_id 和业务参数。

数据表的结构在实现上可以精简为两张表。

agents表存代理概要信息：

task_events表存事件流，每条事件包含 task_id、agent_id、event_type、score_snapshot、timestamp。后续查“这个任务当时是发给谁的，为什么发给他”时，直接按 task_id 查这张表，一目了然。

这里有个设计经验要重点说：路由决策里唯一不可事后推演的信息就是“当时的评分快照”。如果只存最终选中的代理 ID，出了问题你根本不知道当时打分排序长什么样，也不知道是哪个因素让某个代理排在前面。因此每一条 routed 事件我都会把 agent 列表和各自的评分一起存进去，代价是数据量大几倍，但排障效率成倍提高，这钱花得值。

3. 核心细节解析与实操要点

3.1 心跳与超时判定：参数怎么定才不会误杀

心跳机制是整个触达管理的生命线。我用的是最朴素、最稳定的模型：代理每隔固定间隔上报一次心跳，服务端记录最近一次心跳时间，然后由一个扫描过程周期性检查所有代理的最近心跳时间，如果超过某个阈值还没更新，就认为这个代理失联。

这里的核心参数有三个：心跳间隔、失联阈值、扫描周期。我在系统里最终定的值是：心跳间隔 5 秒、失联阈值 15 秒、扫描周期 1 秒。为什么心跳间隔是 5 秒而不是更快或者更慢，这是实践权衡出来的：太短的话代理数量一多，服务端压力上升很快，而且网络抖动导致的迟到心跳会让系统误以为节点失联；太长的话失联发现时间会拉长。5 秒加 3 次心跳周期作为阈值，既能容忍偶发的一两次心跳丢失，又能在一分钟内完成故障发现和路由切换。

要特别说明的是失联阈值千万不要设成 2 倍或者 1 倍心跳间隔。我在早期测试中把阈值设成 10 秒，结果代理稍一阻塞超过 10 秒就被判定失联，任务被路由到别的节点，原代理恢复后发现自己已经没有任务可做，出现了大量重复调度。后来换到 3 倍间隔加 15 秒阈值后，误杀率几乎降到了零。

代理端心跳上报还有一个容易被忽略的坑：心跳接口和业务执行必须放在不同的线程，否则代理在跑一个耗时任务时，心跳会阻塞在业务代码后面，服务端会误判失联。我在代理端用 Python 的 ThreadPoolExecutor 单独跑了一个心跳线程，和主任务循环完全隔离，实测下来即使主循环处理超大任务卡住几十秒，心跳线程也能正常上报，服务端不会产生误判。

关于超时阈值计算，给一个可以直接用的伪代码思路，扫描的时候比较now - last_heartbeat_at和heartbeat_interval * 3 + 允许的时钟偏移即可。时钟偏移一般取 500ms，避免多机时间不同步造成的边缘误杀。

3.2 可达性评分与路由策略：让任务去找状态最好的代理

心跳只能告诉你代理“活着”，但“活着”不等于“适合接任务”。一个代理可能在线，但队列里已经积压了 200 个任务，把新任务再塞给它只会让整体延迟变大。所以路由前必须对每个在线代理算一个可达性分数，分数越高，表示这个任务交给它之后被快速执行的可能性越大。

我用的评分公式可以简单写成四个因子的加权组合：

• 新鲜度因子(A)：根据最近心跳时间计算，心跳越近越接近 1，超过失联阈值则直接归零。这里用一个指数衰减函数，刚收到心跳时约为 1，到第 15 秒降到接近 0。
• 延迟因子(B)：1 / (1 + 平均代理处理时间)。平均处理时间越短，分数越高。
• 积压因子(C)：1 / (1 + 当前队列深度)。队列越空，分数越高。
• 成功率因子(D)：最近 50 次任务的成功比例。

最终分数是score = 0.3 * A + 0.2 * B + 0.3 * C + 0.2 * D，权重可以根据实际场景调，如果你更关注低延迟，可以放大 B 的权重；如果你更关注稳定，就放大 D。

路由策略听起来很复杂，但我建议从一个足够简单且合理的规则开始：先按可达性分数过滤掉低于阈值的代理，再从剩余代理中挑选分数最高且当前队列深度最小的那个。不加这个过滤直接选分数最高的，容易被“看着在线但其实最近才开始恢复”的代理带偏。分数过滤阈值我在测试中定在 0.3，低于这个值不投递，任务进入延迟队列等待下一轮重试。

这个评分机制的另一个好处是可解释。路由时我不仅记下最终选了谁，还会把所有候选代理的评分存进事件日志。这样当线上反馈“这个任务为什么发给了 agent_03，明明 agent_04 更快”时，我可以直接把当时的评分排行拉出来，看清是延迟因子还是积压因子拉低了分数，再针对性调整权重。

3.3 幂等与重试：触达成功不等于只执行一次

分布式系统里最尴尬的一种情况：消息确实发到了代理手上，代理也执行了，但回执丢了或者超时，调度端以为没成功，于是把任务重新投递了一次，结果同一任务被执行了两次。Agent-Reach 里我用三层机制来防御这个问题。

第一层是任务级幂等。每个任务带全局唯一的 task_id，代理端在执行前用这个 task_id 在本地 SQLite 表里做一个“执行状态”的插入或更新，状态字段是 pending 到 running 再到 done，靠唯一约束保证同一个 task_id 在同一时刻只有一个 running 状态。如果插入失败，说明这个任务已经在执行或已经完成，直接忽略新到达的投递。

第二层是消费组的 ack 机制。代理从 Redis Streams 的消费组里读出任务后，必须等业务执行完成才调用XACK确认。如果代理崩溃，任务会留在 pending 列表里，由扫描进程用XAUTOCLAIM把它重新投递给别的代理。这里有个细节：重新投递之后，新代理的第一件事依然是查幂等表，而不是直接执行业务，这样才能把“触达成功”和“执行成功”两个概念彻底分开。

第三层是重试间隔的指数退避加抖动。重试间隔按 2 秒、4 秒、8 秒、16 秒递增，同时每次加 10% 到 20% 的随机抖动，防止多个失败任务同时重试时在 Redis 上造成惊群效应。重试超过 5 次的任务放进死信队列，由人工或上层系统介入处理。这里强调一下：别把所有失败任务都不分青红皂白地重试，有些失败是永久性的，重试只会白白消耗资源，识别永久性失败并把它们推入死信是分布式任务系统必须养成的习惯。

3.4 实时触达状态推送：全量快照 + 增量事件

控制台面板要实时显示“哪些代理在线、每个代理当前打分、任务事件流”，我最初用的是前端轮询，每 2 秒拉一次状态接口。这么做最大的问题是：状态接口返回的是一个聚合视图，后端每次都要把所有代理状态重新计算一遍，前端拿到数据后还要全量重绘，当时觉得能用，但一旦代理数量超过二十个，监控页面就肉眼可见地卡顿。

后来我把推送改成 WebSocket，并在接口设计上采用“全量快照 + 增量事件”的组合：面板连接建立后，服务端先把当前所有代理状态、任务状态汇总成一棵快照树发给前端；此后只推送各类事件，事件包括 HeartbeatEvent、RouteEvent、TaskStateChange，前端收到增量事件后，更新本地状态树中对应的节点，而不是整棵树重画。

这个设计带来的直接收益是带宽消耗降了两个数量级，一个心跳事件大约只有 200 字节，不再需要每 2 秒把整个状态表拖一遍。实时性也从 2 秒延迟提升到小于 100 毫秒。WebSocket 断线重连后，前端重新走一次全量快照流程，确保状态收敛。这套模式跟现在主流前端框架的 state 管理天然契合，不依赖任何重型的前端轮子。

4. 实操过程与核心环节实现

4.1 环境准备与基础依赖

为了让你能照着复现，这部分讲的是真实的运行环境。主服务跑在一台 2 核 4G 的 Linux 机器上，Python 版本 3.11，Redis 版本 7.0，其余依赖全部用 pip 安装，没有跨平台的坑。

安装依赖时我用的是一个干净的虚拟环境，依赖列表如下：

pip install fastapi uvicorn redis pydantic websockets

Redis 只需要默认配置启动即可，不需要给 Streams 做特殊的配置文件。注意如果代理和调度端跨机器部署，Redis 的protected-mode要记得关闭，或者把bind改成内网地址，同时设一个访问密码，否则代理连不上会白白浪费大半天排障时间。

项目的目录结构非常简单，我实际部署时的目录长这样：

agent-reach/ ├── server.py # FastAPI 入口，HTTP + WebSocket ├── router.py # 可达性评分与路由决策 ├── registry.py # 代理状态表管理 ├── consumer.py # Stream 转投递循环 ├── agent_demo.py # 代理端模拟进程 └── static/ └── panel.html # 纯静态控制台页面

接下来我会分别拆解代理端和调度端的代码实现。代理端单独用了一个文件刻意把所有上报逻辑放在一起，方便你看清楚心跳循环、任务读取循环之间的线程模型。

4.2 代理端实现：注册、心跳、任务执行

代理端承担三类工作：启动时注册、周期心跳、消费任务。核心代码我用一个简化版展示，标注了关键逻辑。

import requests import threading import time AGENT_ID = "agent_07" SERVER_URL = "http://127.0.0.1:8000" def register(): requests.post(f"{SERVER_URL}/agent/register", json={ "agent_id": AGENT_ID, "meta": {"labels": ["crawler", "zone-a"]} }, timeout=5) def heartbeat_loop(): while True: try: requests.post(f"{SERVER_URL}/agent/heartbeat", json={ "agent_id": AGENT_ID, "queue_depth": get_current_queue_depth() # 真实积压数 }, timeout=3) except Exception: pass # 心跳失败不阻塞主循环 time.sleep(5) def work_loop(): while True: task = fetch_next_task() if task is None: time.sleep(1) continue mark_running(task.task_id) try: result = run_task(task) ack_task(task.task_id, result) except Exception as exc: mark_failed(task.task_id, str(exc)) nack_task(task.task_id) def main(): threading.Thread(target=heartbeat_loop, daemon=True).start() work_loop() if __name__ == "__main__": main()

这里专挑了三个容易踩坑的点提醒你。第一，queue_depth上报的是真实运行中的任务数，而不是等待中的数量，否则调度端会把一个执行很慢但排队很少的代理误判为“很闲”。第二，心跳循环一定要包一层异常捕获，心跳失败不能影响主业务循环，很多代理挂掉的表象其实是心跳线程先死掉了。第三，任务执行时先mark_running再做业务，顺序不能反，否则会重复执行。

4.3 调度端实现：状态注册、评分与路由

调度端是系统的核心部分，我用 FastAPI 实现了三个 HTTP 接口，分别是注册、心跳、任务提交。注册是元信息写入，心跳负责实时状态更新，任务提交是整个路由决策的入口。

from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class RegisterData(BaseModel): agent_id: str meta: dict = {} class HeartbeatData(BaseModel): agent_id: str queue_depth: int class TaskSubmit(BaseModel): task_id: str payload: dict @app.post("/agent/register") async def post_register(data: RegisterData): registry.upsert(data.agent_id, data.meta) return {"ok": True} @app.post("/agent/heartbeat") async def post_heartbeat(data: HeartbeatData): registry.heartbeat(data.agent_id, data.queue_depth) push_event({"type": "heartbeat", "agent_id": data.agent_id, "queue_depth": data.queue_depth}) return {"ok": True} @app.post("/task/submit") async def post_task(data: TaskSubmit): candidates = registry.online_agents() chosen = router.pick(candidates, data.task_id) if chosen is None: return {"task_id": data.task_id, "status": "deferred"} produce_task_to_stream(data.task_id, chosen.agent_id, data.payload) push_event({"type": "routed", "task_id": data.task_id, "agent_id": chosen.agent_id}) return {"task_id": data.task_id, "status": "routed", "agent": chosen.agent_id}

router.pick 的实现就是评分排序加过滤，伪代码如下：

def pick(candidates, task_id): alive = [a for a in candidates if a.is_alive()] scored = [(calc_reach_score(a), a) for a in alive] qualified = [(s, a) for s, a in scored if s >= SCORE_THRESHOLD] if not qualified: return None qualified.sort(key=lambda x: (-x[0], x[1].queue_depth)) return qualified[0][1]

这里排序的 key 设计是两个维度：主键是评分降序，副键是队列深度升序，这样能在同等评分下优先选更闲的代理，避免任务全部挤向了同一台在线但负荷高的机器。实际上线后，这种双重排序比单纯评分排序更均衡，值得参考。

4.4 控制台面板与实时数据展示

控制台部分我没有引任何前端框架，就一个 HTML 页面加一段原生 JavaScript。页面分四块：顶部一组全局统计卡，当前在线代理数量、离线代理数量、平均可达性分数、待处理任务数；中间一张代理状态表格，列出每个代理的 ID、状态、最近心跳距离、队列深度、实时评分；底部是一条滚动的事件流，每条事件有类型和时间的标签。

WebSocket 建立之后，前端先接收一个snapshot消息，里面是完整的{agents: [...], tasks: [...], stats: {...}}结构；之后所有event消息都在本地状态树上做增量更新。这个模式最关键的一点是：前端永远不发起全量刷新，只在断线重连时才请求一次快照。响应式更新用最简单的render()函数做整表局部重绘即可，没必要引入虚拟 DOM。

我把一段连接逻辑贴在下面，注意看断线重连的部分：

let ws; function connect() { ws = new WebSocket("ws://127.0.0.1:8000/ws"); ws.onopen = () => ws.send("sync"); ws.onmessage = (msg) => applyEvent(JSON.parse(msg.data)); ws.onclose = () => setTimeout(connect, 1000); } connect();

断线后 1 秒重试，重连成功后onopen里发送sync，服务端收到sync就会回发全量快照，能在一个连接周期内把状态收敛回来。这个模式很朴素，但非常可靠，我在三个月运行里面板没再出现过“卡在旧状态”的情况。

4.5 压测结果与关键参数调优记录

做完主流程后，我做了一轮基础压测，数据供你配置时参考。测试环境就是上面说的 2 核 4G 机器，20 个模拟代理进程，每个代理进程的队列深度都不同，任务按固定速率持续提交。

压测中最有价值的调参发现是：路由切换时间基本由失联阈值决定。如果你把失联阈值从 15 秒改成 10 秒，切换时间可以降到 7 秒左右，但误杀率会上升到接近 1%。对大多数业务来说，12 秒的切换时间完全可以接受，没必要为了缩短几秒而担上误杀风险。所以最后我把阈值锁定在 15 秒，误杀率几乎为 0，故障切换也稳定。

5. 常见问题与排查技巧实录

5.1 代理频繁被误判失联，怎么定位

这是我遇到过最烦的问题，没有之一。表现是调度端日志里某个代理反复出现“offline”，但登录到代理机器上看，业务进程活得好好的。排查顺序建议你按照下面这条线来。

第一，查代理端代码里的心跳是不是被某个全局锁阻塞了。如果心跳函数和业务函数共用了同一个锁，业务一旦进入长时间等待，心跳就跟着阻塞，服务端收不到心跳自然判失联。我的解决方式是给心跳分配独立线程，并且在代码里保证心跳请求不跟任何业务锁产生关联。

第二，检查网络链路上的 NAT 超时。如果代理部署在远程机房，中间经过 NAT 设备，长连接超时空闲后，服务端发送的探测请求可能被丢弃，而代理还认为连接是通的。心跳请求用 HTTP 短连接没这个问题，但如果你用 WebSocket 做心跳推送，就得格外注意这个坑。

第三，看看代理机器的系统负载和内存。我发现有一次误判是代理机器内存不足，内核开始 OOM，业务进程被杀了但是心跳线程因为是守护线程，在进程退出前还强行发了最后一次心跳，服务端把这次心跳当作“活着”依据，之后再无心跳，最终导致判失联。这种场景不能用阈值调整解决，必须先在底层把资源问题排查完，再来质疑阈值设置。

5.2 任务被重复执行，幂等为什么没生效

重复执行是分布式系统里最头疼的线上事故之一，Agent-Reach 早期我也翻过车。最经典的现象是：一个任务在“已完成”状态下又被代理跑了一次。逐层排查后，根因往往是下面这三个之一。

第一个原因是状态写入顺序错了。代理收到任务后先执行，再更新状态，结果在执行期间调度端重试，另一个代理解析到同一条任务，于是两边同时运行。正确的顺序是收到任务先mark_running再执行业务，并用唯一约束保证同一个 task_id 只能有一个 running 记录。

第二个原因是 ack 逻辑不严谨。消费者从 Stream 读到消息后，没有等业务完成就立刻XACK，如果执行中抛异常，消息已经不在 pending 列表里，调度端无法感知失败，自然也就不会把任务重新投递给其他代理。但是如果你手动在排障时重新投递了这条消息，而业务又执行了一遍，重复就发生了。所以要严格遵守“先完成业务，再 ack 消息”的顺序。

第三个原因是幂等记录的 TTL。我在本地 SQLite 里维护 dedup 表，原本给每条记录设了 24 小时过期，结果遇到一个要跑 30 小时的超长任务，任务还没结束记录就被清掉了。后来我把 TTL 提升了三倍并跟任务超时时间联动，这个问题才消失。设计 dedup 表时，TTL 一定要留足余量。

5.3 消息积压导致触达延迟飙升怎么办

触达延迟飙高大多数时候不是调度端的问题，而是消费侧的某个代理处理太慢，把 Stream 里的消息堵住了。因为 Redis Streams 的消费组是“消息顺序分发给不同消费者”的，虽然多个消费者可以并行处理不同消息，但如果某一个消费者迟迟不 ack，它读过的消息会停留在 pending 列表里，消费组本身不被阻塞，但单个分区的滞后时间会不断增长。

我的处理办法是给不同业务类型的任务建不同的 Stream 分区，或者干脆按 agent 维度建 Stream，而不是让所有任务共享同一个 Stream。这样某个代理变慢只会影响自己的那条流，其他代理的任务互不干扰。同时监控XLEN（消息长度）和消费组 lag，设置一个预警线，比如 lag 超过 200 就告警，避免等到用户感知到延迟才去补救。

还有一次积压的根因在 Redis 本身：预取逻辑我没限制，每次消费都预取 100 条消息到本地队列，结果内存直接顶爆，Redis 触发缓存淘汰，把一些 Stream 数据也淘汰掉了。后来我把预取数量改成 10，并给消息设置合理 TTL，情况立刻就好了。这个教训提醒我，Streams 虽然在 Redis 里持久化，但它不是无限消息日志，所有消息都得按时间设置合理的生命周期。

5.4 WebSocket 推送中断后如何兜底

WebSocket 中断本身不影响调度主流程，因为调度的事件都会写入 SQLite，不会丢，但面板会冻住，导致值班的人误判系统状态。我维护面板时遇到过三次断连，每次表象都是“任务还在跑，但面板不动了”。

根因基本都是网络链路问题：要么是代理或者监控机器长时间空闲被中间网络设备断开，要么是浏览器休眠时把 WebSocket 连接回收了。兜底方案我用了两层：第一层是前端断线自动重连，重连成功后重新拉全量快照；第二层是即便 WebSocket 处于连通状态，我也让前端每隔 30 秒向后端发一个 ping，后端起一个定时器检查连接，如果超过 45 秒没收到心跳，就主动断开让前端重连。

如果你也在做类似的面板，建议把这两层都加上，别只依赖前端断线事件。因为 WebSocket 断线事件在很多浏览器环境下并不可靠，尤其是长时间后台运行的页面，连接可能已经失效但 onclose 一直没触发。

最后再分享一个我个人体会最深的经验：Agent-Reach 这套东西做下来，真正难的不是路由算法本身，而是把每一个调度决策变成可追溯的状态记录。给代理打分很简单，让打了分的过程能被事后审查就难得多。所以我在所有关键路径上都加了快照和事件存储，这会让开发期多写不少代码，但运行期省下的排障时间远超开发的成本。如果你也准备在自己的多代理系统里加一层触达调度，我建议你一上来就考虑清楚“每个决策是否都有记录”，这件事比选哪个消息中间件重要得多。