1. 项目定位Hermes 到底解决了什么问题做后端开发这些年我一直在找一个平衡点消息中间件要足够轻、足够灵活但又不想每次都为了一套重型的消息队列去维护一堆依赖。看到 Hermes-Agent 这个名字的时候我第一反应就是那个希腊神话里的信使神带翅膀的脚传递消息速度极快。这名字起得很贴切因为它就是一个传消息的框架。不过它不是那种大而全的企业级消息总线而是一个更纯粹、更嵌得进去的轻量级代理系统。简单说Hermes-Agent 是一个面向服务间异步通信与任务编排的轻量级消息代理框架。它的核心职责只有三件事接住上游系统发来的消息、按照规则把消息路由到正确的任务处理器、在处理器失败时执行重试和补偿。听起来很像微服务里常见的消息队列对不对确实功能上有重叠但它的差异化优势在于极简和嵌入友好。你不需要单独部署一套 Broker不需要 ZooKeeper不需要 Kubernetes甚至可以在一台 1 核 2G 的小机器上把它跑起来直接嵌进你自己现有的 Python 服务里把消息处理能力拉到和主服务同一个进程空间里。适合谁来用我个人的感受是如果你正在做一个中小规模的业务系统比如电商订单同步、物联网设备数据上报、内部工单流转消息量一天在几十万到几百万这个量级不想引入 Kafka 那套重基建也不想被 RabbitMQ 的运维细节缠住那 Hermes-Agent 会非常顺手。如果你是学生或者独立开发者想搞明白消息代理到底是怎么回事用它来做教学复盘也很合适——因为它代码量不大核心链路清晰读一遍源码基本能把消息投递、确认、重试这一整套机制串起来。我最早接触这个项目是想解决一个实际问题公司内部的订单服务和库存服务之间需要异步同步数据之前的方案是直接 HTTP 调用加数据库定时任务轮询结果就是接口一抖、库存就对不上日志里全是超时重试。后来我尝试了几个方案最终留下了 Hermes-Agent。这篇博文不打算写成文档翻译我会把架构逻辑、部署步骤、路由策略的配置方式以及我实际运行中踩过的坑全部摊开来讲清楚。就算你之前完全没接触过这类中间件照着这篇文章走一遍也能跑起来。1.1 它和主流消息队列的边界在哪很多人看到消息代理四个字第一反应是可以用 RabbitMQ 或者 Kafka然后就会问一句话已经有了这么多成熟的消息中间件为什么要用一个名不见经传的 Hermes-Agent这个问题我自己也想过很久后来在项目里做了三种方案对比之后才真正想明白不是替代关系是补充关系。RabbitMQ 和 Kafka 解决的是大规模消息流、跨团队、高持久性的公共基础设施问题它们需要独立部署、独立运维还要考虑集群、分区、消费者组的配置对一个小团队来说维护成本并不低。而 Hermes-Agent 解决的是应用进程内消息调度的问题它是嵌入式的跟你的服务一起启动一起死掉一起重生。拿个生活化的例子来类比RabbitMQ 就像一个城市的邮政总局所有信件都汇到总局再分拣派送体系完善但你要是只是想在办公楼里给几个部门传个话把邮政总局搬进来就有点夸张了。Hermes-Agent 更像办公楼里的内部电话系统不需要邮局介入拿起电话就能把事情说清楚。它更适合那些局部消息流转的场景尤其是同一个服务内多个模块间的异步解耦。在选型层面我的建议是这样如果消息量级是百万级以内、对消息顺序没有极端要求、希望快速迭代Hermes-Agent 足够好如果消息量是千万级起、需要跨团队复用、要求消息无限期堆积那还是老老实实上 Kafka。工具没有绝对的好坏只有跟场景是否匹配。选 Hermes-Agent 最大的收获是它把我的部署复杂度从维护一套中间件集群降到了写一段 Python 代码这也是它名字里 Agent 的由来——它是一个替你跑腿的助手而不是一栋需要你天天打扫的大楼。1.2 核心模块与整体架构梳理我先把 Hermes-Agent 的模块结构画个大概的逻辑图方便后面展开讲。它主要由这么几个部分组成接入层负责接收外部消息把 HTTP 请求或者队列输入统一转化成内部事件结构路由层根据消息的 topic 和规则表决定这条消息应该进入哪个处理队列调度层维护多个 worker每个 worker 绑定一个或者多个队列消费消息并调用对应的处理器函数存储层保存消息内容和处理状态方便重试和追踪还有一块是管理接口提供简单的指标查询和开关控制。一开始接触有点懵但把它对应到实际业务里就清晰了。比如用户下单之后订单服务产生一条order.created事件接入层收到事件路由层判断这个事件两个消费者感兴趣——库存服务要扣减库存通知服务要发短信于是事件被复制成两份分别投递到 stock.deduct 队列和 notification.send 队列。两个队列由不同的 worker 消费互不干扰。库存扣减成功消息就走完了通知服务临时不可用消息进入重试队列等恢复之后继续尝试。整个过程不会阻塞用户的 HTTP 请求因为是异步的。这就是我说它结构简洁的原因一条消息从进来到处理完只经过四次传递没有多余的 hop。代码层面每个模块都是独立类接口定义清晰你要是想替换存储后端或者改路由策略不用动其他模块的代码。这点对二次开发特别友好我就是在这个基础上加了自定义的路由规则扩展成本很低。2. 配置与核心机制解析2.1 配置文件结构与关键参数Hermes-Agent 使用 YAML 作为配置文件格式整个配置围绕连接入口 队列规则 处理策略三条线展开。初次配置的时候不用贪多先把最核心的几十个参数调好就能跑后续再按业务情况增加规则。这是我的一个基础配置示例去掉敏感信息后可以直接参考server: host: 0.0.0.0 port: 8090 storage: type: sqlite dsn: ./hermes_agent.db queues: order.created: handler: handlers.order_handler:on_order_created max_retry: 3 retry_delay: 5 workers: 2 prefetch: 10 stock.deduct: handler: handlers.stock_handler:on_stock_deduct max_retry: 5 retry_delay: 10 workers: 1 prefetch: 5 routers: - topic: order.* target: order.created - topic: stock.* target: stock.deduct monitor: enabled: true port: 9100逐个解释一下关键参数。storage.type是可插拔存储后端项目默认支持 sqlite 和 redis 两种。sqlite 适合单机小规模部署零依赖一个文件搞定重启之后消息状态还能保留redis 适合需要多个实例共享状态、做简单水平扩展的场景但部署成本高一点。queues下面是具体的队列定义每个队列有独立的 handler、重试次数和 worker 数量。特别要注意max_retry和workers这两个参数。max_retry是消息处理失败后的最大重试次数设太大会导致死信消息反复占用资源设太小又会误杀一些偶发失败的正常请求通常区间在 3 到 5 次比较合理。workers是消费该队列的并发 worker 数它不是越大越好因为 worker 多了会同时抢占 CPU 和数据库连接当前我实践下来大部分业务队列设 2 就够了只有那种 I/O 密集型的通知类任务可以拉到 5 到 8。配置里面还有一个容易忽略的东西是prefetch意思是每个 worker 一次最多从队列里拉多少条消息放在本地缓冲区。这个参数对吞吐量和消息堆积有直接影响。prefetch 太大会导致某个 worker 手里压着一大批消息却处理不动其他 worker 闲置太小又会让每次消费都要走一次存储查询增加延迟。经验值大概是 workers * prefetch 的结果控制在 30 以内。2.2 消息路由与队列策略详解路由层是最值得花时间研究的部分因为这决定了你的消息能不能准确送到该去的处理器。Hermes-Agent 的路由是基于 topic 通配符匹配的支持*匹配单级和#匹配多级两种模糊匹配熟悉 MQTT 的读者应该一看就明白。举个例子。业务里我定义了三类事件order.created、order.paid、order.shipped。我只写一条路由规则order.*这样三个事件就都进入了order.created队列。接着我再用一个order.#规则把这些事件复制到日志队列做全量记录。这里有个细节要特别留意路由规则是有序的Agent 按照配置文件里的顺序逐条匹配命中的第一条规则生效。如果你把#这种宽泛规则写在前面后面的窄规则可能永远匹配不到。所以配置顺序上要遵循具体规则在前泛化规则在后的原则。队列策略上有个模式我非常推荐一个 topic 对应一个队列队列对应一个处理函数。这种一线性结构最大的好处是排查问题的时候非常直观收到一条消息看它 topic 就能推断出它走了哪条链路不需要在全局的 handler 注册表里翻来翻去。等到业务复杂了再按延时敏感型和可靠性敏感型把队列做二级拆分及时推送类走内存队列交易对账类走持久化队列这样轻重分离资源利用率更高。还需要解释一下广播和点对点的区别。Hermes-Agent 默认是点对点消费模型一条消息只会被同一个队列下的一个 worker 消费掉这保证了对库存扣减这种必须只执行一次的操作是安全的。但如果你想做发布订阅也就是一条消息被多个消费者各自处理一遍就需要配置多个队列或者使用路由复制规则把消息投递给多个目标。我上面日志同步的例子就是这么用的同一份order.created消息一份给业务处理队列一份给日志队列两边消费互不影响。2.3 任务调度与失败补偿机制说到可靠投递这可能是使用任何消息代理时最容易闹心的问题消息发出去了接收方崩了怎么办处理函数里抛了个异常怎么办进程重启了队列里没处理完的消息会不会丢Hermes-Agent 的答案有三层持久化存储、确认机制、重试补偿。存储层解决消息会丢的问题。投递进系统的消息会先落库状态是 pending只有处理成功之后才会标记为 done否则一直保留在库里。所以哪怕服务中途宕机重启之后 Agent 会自动扫描 pending 状态的消息重新派发给对应的 worker。这就是我把storage.type设置成 sqlite 的原因单文件持久化丢失风险接近于零。确认机制解决重复处理的问题。消息从队列里被取出来交给 handler 之后并不会立即删除而是进入处理中状态并有一个处理超时时间。如果 handler 在超时时间内没有显式地 ack 确认Agent 会判定处理失败并把消息恢复到队列里交给下一个 worker 重试。这样就防止了进程处理到一半卡住了消息却已经被认为消费完成的问题。重试补偿解决失败怎么办的问题。处理函数抛出异常之后Agent 不会立刻执行重试而是把消息放进一个带延迟的待重试队列。retry_delay参数控制两次重试之间的间隔避免失败原因还没恢复就疯狂重试把下游服务打挂。我经历过一次比较典型的情况通知服务因为数据库连接池满了所以持续报错如果没有重试延迟Agent 会每秒重试一次直接加剧下游压力那场面太恐怖了。加上延迟之后5 秒一次、10 秒一次给下游留出恢复时间系统逐步自愈了。当重试次数耗尽仍然失败消息会进入死信队列DLQ不会继续无限循环。实际业务中我建议对死信队列做人工巡检因为它们通常意味着代码 bug 或者数据格式异常需要真正的人工介入。我自己的做法是每天都看一遍死信队列的长度如果突然涨上来就去查这几条消息触发的 handler 日志往往能提前发现潜在故障。3. 从零部署 Hermes-Agent 的完整实操3.1 环境准备与依赖安装部署 Hermes-Agent 对环境的要求不高我测试过 Python 3.9 到 3.12 都能正常跑。需要说明的是它是一个标准的 Python 包安装之前最好先建一个干净的虚拟环境避免跟系统环境里的其他包产生版本冲突。下面是安装步骤我按顺序记录下来# 创建并激活虚拟环境 python3 -m venv hermes-env source hermes-env/bin/activate # 安装核心包 pip install hermes-agent # 可选如果使用 redis 作为存储后端需要额外安装驱动 pip install hermes-agent[redis] # 验证安装是否成功 hermes-agent --version如果你是从源码安装流程稍微多一点但很直接git clone https://github.com/your-source/hermes-agent.git cd hermes-agent pip install -e .安装完成之后项目会生成一个hermes-agent命令行工具。这个工具提供三个常用命令start用来启动代理服务check用来校验配置文件语法list用来查看当前已经注册的队列和处理器。我每次改完配置都会先跑一遍check能少踩很多低级错误。依赖方面核心依赖其实只有几个PyYAML用于解析配置pydantic用于参数校验requests或aiohttp用于 HTTP 接入层。其他的都是可选项按需安装就好。这种轻量的依赖设计是我喜欢它的原因之一——它可以安安静静地待在你的服务旁边不会把整个虚拟环境撑得乱七八糟。3.2 第一个消息收发 Demo装好之后我们要做的第一件事往往是验证一条消息能不能顺利从生产者走到消费者。我会带你写一个最简单的场景生产者向demo.queue发送一条消息消费者收到后打印出来。先定义一个处理器文件handlers.pyimport logging logger logging.getLogger(hermes) def handle_demo(payload: dict) - dict: logger.info(收到消息: %s, payload) # 模拟业务处理耗时 time.sleep(0.1) return {status: ok, data: payload}注意handler 函数的签名是统一的接收一个 dict 类型的 payload返回一个 dict。返回的内容会被 Agent 记录下来方便追踪。如果处理失败直接抛异常就行Agent 会自动捕获并按重试策略重新调度。对应的配置文件demo_config.yamlserver: host: 127.0.0.1 port: 8090 storage: type: sqlite dsn: ./demo.db queues: demo.queue: handler: handlers:handle_demo max_retry: 3 retry_delay: 1 workers: 1启动服务hermes-agent start -c demo_config.yaml然后另开一个终端用 Python 脚本或者 curl 发送消息。一个最简单的 Python 发送端代码是import requests resp requests.post( http://127.0.0.1:8090/publish, json{ topic: demo.queue, payload: {hello: hermes} } ) print(resp.status_code)如果一切正常你会看到服务端日志输出收到消息: {hello: hermes}。到这一步一个完整的消息链路就已经通了。你已经用不到 10 行业务代码搭起了一个可用的异步消息系统这个成就感还是很足的。3.3 接入真实业务订单状态同步案例Demo 通了之后我建议直接拿一个贴近实际的场景来跑通整个流程这样才能理解配置参数之间是如何协同的。我选的是订单状态同步这个常见场景因为它覆盖了消息投递、失败重试、死信处理三个核心要点。业务需求是这样的订单服务收到用户下单请求后需要在异步流程中同时做三件事给用户发下单成功通知、通知仓库系统备货、把一个落库操作写入数据分析表。这正好对应三个独立的处理器。处理器handlers.py扩展一下def on_notification(payload: dict) - dict: # 调用短信或推送服务 return {sent: True} def on_warehouse(payload: dict) - dict: # 调用仓库系统接口如果仓库系统 5xx这里抛异常触发重试 resp warehouse_api.stock_in(payload[sku], payload[qty]) if not resp.ok: raise RuntimeError(warehouse api error) return {warehouse: resp.ok} def on_analytics(payload: dict) - dict: # 写分析库通常不会失败但有概率失败 write_analytic_record(payload[order_id]) return {db: True}配置文件里把它们注册到三个队列queues: order.notify: handler: handlers:on_notification max_retry: 3 workers: 2 order.warehouse: handler: handlers:on_warehouse max_retry: 5 retry_delay: 10 workers: 2 order.analytics: handler: handlers:on_analytics max_retry: 2 retry_delay: 2 workers: 1配置完后重启服务。每次用户下单订单服务只需要向三个 topic 各发一条消息剩下的都由 Hermes-Agent 去异步处理。仓库服务临时不可用怎么办on_warehouse抛出异常消息进入重试队列等 10 秒再试最多试 5 次。如果仓库服务连续挂了超过 1 分钟消息进入死信队列等待人工介入。整个过程订单接口的响应时间不受任何影响。这段实操里我踩过一个很关键的坑就是在handler返回的 dict 里放了一些无法序列化的对象比如一个数据库连接对象结果 Agent 在记录返回结果时直接序列化失败导致消息处理失败、进入重试循环。解决办法是 handler 的返回值只放基础类型让生产者和消费者用提前约定好的协议来传递业务信息——我在真实项目里普遍的做法是消息体就是 JSON所有字段都是字符串数字布尔值所有状态都靠约定而非对象引用。4. 踩坑实录与性能调优4.1 高频踩坑消息丢失、重复消费、死锁这几个问题几乎伴随着每一个初用消息代理的人。我把我遇到的高频问题整理成一张速查表每个问题都附上排查思路和解决方案。问题现象可能原因排查方法解决方案消息丢了存储后端未启用持久化检查 storage 配置确保 type 为 sqlite 或 redis不要用内存模式消息重复处理处理超时导致 ack 延迟查看 handler 耗时日志增加超时时间或者让 handler 具备幂等性队列堆积、消费变慢workers 数量不足监控队列长度调大 workers同时注意 prefetch 值服务重启后消息状态异常崩溃瞬间消息处于处理中状态数据库查看消息状态启用启动时自动恢复扫描处理中消息死信队列快速增长handler 代码 bug检查异常堆栈修复代码死信消息规范化记录我特别要强调的是重复消费这个问题的处理思路。因为 Hermes-Agent 的确认机制是超时确认如果 handler 处理时间超过了超时阈值消息会被重新投递给同队列里的其他 worker而原来那个 worker 可能还在处理中于是同一条消息被并发处理了两次。对于扣库存、转账这类敏感操作重复处理是不可接受的。这不能靠中间件层面完全解决必须业务处理器自己做到幂等。幂等的实现方式有很多种可以用数据库唯一约束比如订单号加类型做联合唯一索引也可以在消息体里带一个唯一消息 ID处理前查一下这个消息 ID 是否处理过还可以用 Redis SETNX 做一次性标记。我的实践经验是数据库唯一约束最可靠因为它是持久化的不会因为进程重启就丢掉标记。任何消费敏感业务消息的 handler都必须考虑幂等这不只是理论建议而是真实事故换来的教训。4.2 性能调优参数说明如果只是 demo 级别的消息量默认参数完全够用。但一旦进入生产环境每天百万级消息的话几个调优参数就非常重要了。第一个是 worker 并发数的调整。这里有个基本判断方法先看你的 handler 是 CPU 密集型还是 I/O 密集型。如果是 I/O 密集型比如调用外部 API、读写数据库worker 数可以调大因为大部分时间都在等待 I/O如果是 CPU 密集型比如大量计算或者编码转换worker 数不要超过 CPU 核心数太多否则频繁切换上下文反而降低吞吐。我自己的规则是I/O 型业务worker 数从 2 开始逐步增大观察 CPU 使用率CPU 型业务worker 数直接设为 CPU 核心数减 1。第二个是批量消费。Hermes-Agent 支持一次拉取多条消息批量处理配置项叫batch_size。这个参数对吞吐量提升非常明显因为它减少了消息处理的上下文切换和数据库状态更新的次数。我做过一次对比测试batch_size 从 1 调到 20同一批 10 万条消息的处理耗时从 8 分钟下降到了不到 3 分钟。代价是单条消息处理延迟变高因为要等凑齐一整批才会开始处理。对实时性要求不高的场景可以放心用但通知推送这种要求秒级延时的batch 就不要开了。第三个是存储性能。sqlite 在单机场景下很稳但写入量一大就会遇到锁竞争。我建议开启 WAL 模式它能让读和写并行显著改善并发处理能力。同时把synchronous设为NORMAL在每次写操作之间减少 fsync 次数换取更高的吞吐。再一个实用的做法是把消息内容和消息状态分开存消息内容字段可以很大状态字段很小拆成两张表之后状态更新的频率和速度都会好很多。4.3 监控与可观测性配置无论系统多小我都坚持从一开始就配置监控否则出了事故你连从哪排查都不知道。Hermes-Agent 自带一个监控端口暴露了一些指标默认在 9100 端口上配置项monitor.enabled打开之后就能用。核心指标有这么几个hermes_queue_depth队列当前积压消息数、hermes_processing_total正在处理中的消息数、hermes_retry_total累计重试次数、hermes_dlq_total死信队列累计数。用 Prometheus 定时抓取这些指标配合 Grafana 画一个简单的看板系统状态就一目了然。我个人的习惯是重点盯两个指标一个是队列深度另一个是死信队列变化率。队列深度代表当前积压的压力如果长期不降说明消费者速度跟不上生产速度需要扩 worker 或者优化 handler。死信队列变化率则更像一个紧急程度信号平稳时期几乎为零一旦突然上升基本可以认定有代码层面的问题。日志方面Hermes-Agent 默认输出到 stdout也支持配置日志文件。我建议在生产环境至少把日志级别调到 INFO并且按照消息 ID 做一个关联字段这样看日志的时候可以一条链路追到底。排查问题时我通常先按消息 ID 过滤出这条消息的完整生命周期——什么时候进来、被哪个 worker 领走、处理结果是什么——基本能快速定位问题出在哪一环。日志是排障的第一工具别省。5. 最后一点实际体会项目做完之后我最大的体会是中间件的价值不在功能堆砌而在于把工程复杂度藏好。Hermes-Agent 把消息投递、确认、重试、死信这一整套机制封装了起来留给你的是几个简单的配置项和处理函数这让业务代码能保持非常干净。但它也不是万能的遇到超大流量或者复杂路由需求该上重型消息队列的时候不要犹豫。选型之前想清楚自己的真实场景和资源约束比盲目追新技术重要得多。一个消息代理轻便顺手、出了问题你能看懂就比什么都强。
轻量级消息代理框架Hermes-Agent实战：架构、配置与踩坑指南
1. 项目定位Hermes 到底解决了什么问题做后端开发这些年我一直在找一个平衡点消息中间件要足够轻、足够灵活但又不想每次都为了一套重型的消息队列去维护一堆依赖。看到 Hermes-Agent 这个名字的时候我第一反应就是那个希腊神话里的信使神带翅膀的脚传递消息速度极快。这名字起得很贴切因为它就是一个传消息的框架。不过它不是那种大而全的企业级消息总线而是一个更纯粹、更嵌得进去的轻量级代理系统。简单说Hermes-Agent 是一个面向服务间异步通信与任务编排的轻量级消息代理框架。它的核心职责只有三件事接住上游系统发来的消息、按照规则把消息路由到正确的任务处理器、在处理器失败时执行重试和补偿。听起来很像微服务里常见的消息队列对不对确实功能上有重叠但它的差异化优势在于极简和嵌入友好。你不需要单独部署一套 Broker不需要 ZooKeeper不需要 Kubernetes甚至可以在一台 1 核 2G 的小机器上把它跑起来直接嵌进你自己现有的 Python 服务里把消息处理能力拉到和主服务同一个进程空间里。适合谁来用我个人的感受是如果你正在做一个中小规模的业务系统比如电商订单同步、物联网设备数据上报、内部工单流转消息量一天在几十万到几百万这个量级不想引入 Kafka 那套重基建也不想被 RabbitMQ 的运维细节缠住那 Hermes-Agent 会非常顺手。如果你是学生或者独立开发者想搞明白消息代理到底是怎么回事用它来做教学复盘也很合适——因为它代码量不大核心链路清晰读一遍源码基本能把消息投递、确认、重试这一整套机制串起来。我最早接触这个项目是想解决一个实际问题公司内部的订单服务和库存服务之间需要异步同步数据之前的方案是直接 HTTP 调用加数据库定时任务轮询结果就是接口一抖、库存就对不上日志里全是超时重试。后来我尝试了几个方案最终留下了 Hermes-Agent。这篇博文不打算写成文档翻译我会把架构逻辑、部署步骤、路由策略的配置方式以及我实际运行中踩过的坑全部摊开来讲清楚。就算你之前完全没接触过这类中间件照着这篇文章走一遍也能跑起来。1.1 它和主流消息队列的边界在哪很多人看到消息代理四个字第一反应是可以用 RabbitMQ 或者 Kafka然后就会问一句话已经有了这么多成熟的消息中间件为什么要用一个名不见经传的 Hermes-Agent这个问题我自己也想过很久后来在项目里做了三种方案对比之后才真正想明白不是替代关系是补充关系。RabbitMQ 和 Kafka 解决的是大规模消息流、跨团队、高持久性的公共基础设施问题它们需要独立部署、独立运维还要考虑集群、分区、消费者组的配置对一个小团队来说维护成本并不低。而 Hermes-Agent 解决的是应用进程内消息调度的问题它是嵌入式的跟你的服务一起启动一起死掉一起重生。拿个生活化的例子来类比RabbitMQ 就像一个城市的邮政总局所有信件都汇到总局再分拣派送体系完善但你要是只是想在办公楼里给几个部门传个话把邮政总局搬进来就有点夸张了。Hermes-Agent 更像办公楼里的内部电话系统不需要邮局介入拿起电话就能把事情说清楚。它更适合那些局部消息流转的场景尤其是同一个服务内多个模块间的异步解耦。在选型层面我的建议是这样如果消息量级是百万级以内、对消息顺序没有极端要求、希望快速迭代Hermes-Agent 足够好如果消息量是千万级起、需要跨团队复用、要求消息无限期堆积那还是老老实实上 Kafka。工具没有绝对的好坏只有跟场景是否匹配。选 Hermes-Agent 最大的收获是它把我的部署复杂度从维护一套中间件集群降到了写一段 Python 代码这也是它名字里 Agent 的由来——它是一个替你跑腿的助手而不是一栋需要你天天打扫的大楼。1.2 核心模块与整体架构梳理我先把 Hermes-Agent 的模块结构画个大概的逻辑图方便后面展开讲。它主要由这么几个部分组成接入层负责接收外部消息把 HTTP 请求或者队列输入统一转化成内部事件结构路由层根据消息的 topic 和规则表决定这条消息应该进入哪个处理队列调度层维护多个 worker每个 worker 绑定一个或者多个队列消费消息并调用对应的处理器函数存储层保存消息内容和处理状态方便重试和追踪还有一块是管理接口提供简单的指标查询和开关控制。一开始接触有点懵但把它对应到实际业务里就清晰了。比如用户下单之后订单服务产生一条order.created事件接入层收到事件路由层判断这个事件两个消费者感兴趣——库存服务要扣减库存通知服务要发短信于是事件被复制成两份分别投递到 stock.deduct 队列和 notification.send 队列。两个队列由不同的 worker 消费互不干扰。库存扣减成功消息就走完了通知服务临时不可用消息进入重试队列等恢复之后继续尝试。整个过程不会阻塞用户的 HTTP 请求因为是异步的。这就是我说它结构简洁的原因一条消息从进来到处理完只经过四次传递没有多余的 hop。代码层面每个模块都是独立类接口定义清晰你要是想替换存储后端或者改路由策略不用动其他模块的代码。这点对二次开发特别友好我就是在这个基础上加了自定义的路由规则扩展成本很低。2. 配置与核心机制解析2.1 配置文件结构与关键参数Hermes-Agent 使用 YAML 作为配置文件格式整个配置围绕连接入口 队列规则 处理策略三条线展开。初次配置的时候不用贪多先把最核心的几十个参数调好就能跑后续再按业务情况增加规则。这是我的一个基础配置示例去掉敏感信息后可以直接参考server: host: 0.0.0.0 port: 8090 storage: type: sqlite dsn: ./hermes_agent.db queues: order.created: handler: handlers.order_handler:on_order_created max_retry: 3 retry_delay: 5 workers: 2 prefetch: 10 stock.deduct: handler: handlers.stock_handler:on_stock_deduct max_retry: 5 retry_delay: 10 workers: 1 prefetch: 5 routers: - topic: order.* target: order.created - topic: stock.* target: stock.deduct monitor: enabled: true port: 9100逐个解释一下关键参数。storage.type是可插拔存储后端项目默认支持 sqlite 和 redis 两种。sqlite 适合单机小规模部署零依赖一个文件搞定重启之后消息状态还能保留redis 适合需要多个实例共享状态、做简单水平扩展的场景但部署成本高一点。queues下面是具体的队列定义每个队列有独立的 handler、重试次数和 worker 数量。特别要注意max_retry和workers这两个参数。max_retry是消息处理失败后的最大重试次数设太大会导致死信消息反复占用资源设太小又会误杀一些偶发失败的正常请求通常区间在 3 到 5 次比较合理。workers是消费该队列的并发 worker 数它不是越大越好因为 worker 多了会同时抢占 CPU 和数据库连接当前我实践下来大部分业务队列设 2 就够了只有那种 I/O 密集型的通知类任务可以拉到 5 到 8。配置里面还有一个容易忽略的东西是prefetch意思是每个 worker 一次最多从队列里拉多少条消息放在本地缓冲区。这个参数对吞吐量和消息堆积有直接影响。prefetch 太大会导致某个 worker 手里压着一大批消息却处理不动其他 worker 闲置太小又会让每次消费都要走一次存储查询增加延迟。经验值大概是 workers * prefetch 的结果控制在 30 以内。2.2 消息路由与队列策略详解路由层是最值得花时间研究的部分因为这决定了你的消息能不能准确送到该去的处理器。Hermes-Agent 的路由是基于 topic 通配符匹配的支持*匹配单级和#匹配多级两种模糊匹配熟悉 MQTT 的读者应该一看就明白。举个例子。业务里我定义了三类事件order.created、order.paid、order.shipped。我只写一条路由规则order.*这样三个事件就都进入了order.created队列。接着我再用一个order.#规则把这些事件复制到日志队列做全量记录。这里有个细节要特别留意路由规则是有序的Agent 按照配置文件里的顺序逐条匹配命中的第一条规则生效。如果你把#这种宽泛规则写在前面后面的窄规则可能永远匹配不到。所以配置顺序上要遵循具体规则在前泛化规则在后的原则。队列策略上有个模式我非常推荐一个 topic 对应一个队列队列对应一个处理函数。这种一线性结构最大的好处是排查问题的时候非常直观收到一条消息看它 topic 就能推断出它走了哪条链路不需要在全局的 handler 注册表里翻来翻去。等到业务复杂了再按延时敏感型和可靠性敏感型把队列做二级拆分及时推送类走内存队列交易对账类走持久化队列这样轻重分离资源利用率更高。还需要解释一下广播和点对点的区别。Hermes-Agent 默认是点对点消费模型一条消息只会被同一个队列下的一个 worker 消费掉这保证了对库存扣减这种必须只执行一次的操作是安全的。但如果你想做发布订阅也就是一条消息被多个消费者各自处理一遍就需要配置多个队列或者使用路由复制规则把消息投递给多个目标。我上面日志同步的例子就是这么用的同一份order.created消息一份给业务处理队列一份给日志队列两边消费互不影响。2.3 任务调度与失败补偿机制说到可靠投递这可能是使用任何消息代理时最容易闹心的问题消息发出去了接收方崩了怎么办处理函数里抛了个异常怎么办进程重启了队列里没处理完的消息会不会丢Hermes-Agent 的答案有三层持久化存储、确认机制、重试补偿。存储层解决消息会丢的问题。投递进系统的消息会先落库状态是 pending只有处理成功之后才会标记为 done否则一直保留在库里。所以哪怕服务中途宕机重启之后 Agent 会自动扫描 pending 状态的消息重新派发给对应的 worker。这就是我把storage.type设置成 sqlite 的原因单文件持久化丢失风险接近于零。确认机制解决重复处理的问题。消息从队列里被取出来交给 handler 之后并不会立即删除而是进入处理中状态并有一个处理超时时间。如果 handler 在超时时间内没有显式地 ack 确认Agent 会判定处理失败并把消息恢复到队列里交给下一个 worker 重试。这样就防止了进程处理到一半卡住了消息却已经被认为消费完成的问题。重试补偿解决失败怎么办的问题。处理函数抛出异常之后Agent 不会立刻执行重试而是把消息放进一个带延迟的待重试队列。retry_delay参数控制两次重试之间的间隔避免失败原因还没恢复就疯狂重试把下游服务打挂。我经历过一次比较典型的情况通知服务因为数据库连接池满了所以持续报错如果没有重试延迟Agent 会每秒重试一次直接加剧下游压力那场面太恐怖了。加上延迟之后5 秒一次、10 秒一次给下游留出恢复时间系统逐步自愈了。当重试次数耗尽仍然失败消息会进入死信队列DLQ不会继续无限循环。实际业务中我建议对死信队列做人工巡检因为它们通常意味着代码 bug 或者数据格式异常需要真正的人工介入。我自己的做法是每天都看一遍死信队列的长度如果突然涨上来就去查这几条消息触发的 handler 日志往往能提前发现潜在故障。3. 从零部署 Hermes-Agent 的完整实操3.1 环境准备与依赖安装部署 Hermes-Agent 对环境的要求不高我测试过 Python 3.9 到 3.12 都能正常跑。需要说明的是它是一个标准的 Python 包安装之前最好先建一个干净的虚拟环境避免跟系统环境里的其他包产生版本冲突。下面是安装步骤我按顺序记录下来# 创建并激活虚拟环境 python3 -m venv hermes-env source hermes-env/bin/activate # 安装核心包 pip install hermes-agent # 可选如果使用 redis 作为存储后端需要额外安装驱动 pip install hermes-agent[redis] # 验证安装是否成功 hermes-agent --version如果你是从源码安装流程稍微多一点但很直接git clone https://github.com/your-source/hermes-agent.git cd hermes-agent pip install -e .安装完成之后项目会生成一个hermes-agent命令行工具。这个工具提供三个常用命令start用来启动代理服务check用来校验配置文件语法list用来查看当前已经注册的队列和处理器。我每次改完配置都会先跑一遍check能少踩很多低级错误。依赖方面核心依赖其实只有几个PyYAML用于解析配置pydantic用于参数校验requests或aiohttp用于 HTTP 接入层。其他的都是可选项按需安装就好。这种轻量的依赖设计是我喜欢它的原因之一——它可以安安静静地待在你的服务旁边不会把整个虚拟环境撑得乱七八糟。3.2 第一个消息收发 Demo装好之后我们要做的第一件事往往是验证一条消息能不能顺利从生产者走到消费者。我会带你写一个最简单的场景生产者向demo.queue发送一条消息消费者收到后打印出来。先定义一个处理器文件handlers.pyimport logging logger logging.getLogger(hermes) def handle_demo(payload: dict) - dict: logger.info(收到消息: %s, payload) # 模拟业务处理耗时 time.sleep(0.1) return {status: ok, data: payload}注意handler 函数的签名是统一的接收一个 dict 类型的 payload返回一个 dict。返回的内容会被 Agent 记录下来方便追踪。如果处理失败直接抛异常就行Agent 会自动捕获并按重试策略重新调度。对应的配置文件demo_config.yamlserver: host: 127.0.0.1 port: 8090 storage: type: sqlite dsn: ./demo.db queues: demo.queue: handler: handlers:handle_demo max_retry: 3 retry_delay: 1 workers: 1启动服务hermes-agent start -c demo_config.yaml然后另开一个终端用 Python 脚本或者 curl 发送消息。一个最简单的 Python 发送端代码是import requests resp requests.post( http://127.0.0.1:8090/publish, json{ topic: demo.queue, payload: {hello: hermes} } ) print(resp.status_code)如果一切正常你会看到服务端日志输出收到消息: {hello: hermes}。到这一步一个完整的消息链路就已经通了。你已经用不到 10 行业务代码搭起了一个可用的异步消息系统这个成就感还是很足的。3.3 接入真实业务订单状态同步案例Demo 通了之后我建议直接拿一个贴近实际的场景来跑通整个流程这样才能理解配置参数之间是如何协同的。我选的是订单状态同步这个常见场景因为它覆盖了消息投递、失败重试、死信处理三个核心要点。业务需求是这样的订单服务收到用户下单请求后需要在异步流程中同时做三件事给用户发下单成功通知、通知仓库系统备货、把一个落库操作写入数据分析表。这正好对应三个独立的处理器。处理器handlers.py扩展一下def on_notification(payload: dict) - dict: # 调用短信或推送服务 return {sent: True} def on_warehouse(payload: dict) - dict: # 调用仓库系统接口如果仓库系统 5xx这里抛异常触发重试 resp warehouse_api.stock_in(payload[sku], payload[qty]) if not resp.ok: raise RuntimeError(warehouse api error) return {warehouse: resp.ok} def on_analytics(payload: dict) - dict: # 写分析库通常不会失败但有概率失败 write_analytic_record(payload[order_id]) return {db: True}配置文件里把它们注册到三个队列queues: order.notify: handler: handlers:on_notification max_retry: 3 workers: 2 order.warehouse: handler: handlers:on_warehouse max_retry: 5 retry_delay: 10 workers: 2 order.analytics: handler: handlers:on_analytics max_retry: 2 retry_delay: 2 workers: 1配置完后重启服务。每次用户下单订单服务只需要向三个 topic 各发一条消息剩下的都由 Hermes-Agent 去异步处理。仓库服务临时不可用怎么办on_warehouse抛出异常消息进入重试队列等 10 秒再试最多试 5 次。如果仓库服务连续挂了超过 1 分钟消息进入死信队列等待人工介入。整个过程订单接口的响应时间不受任何影响。这段实操里我踩过一个很关键的坑就是在handler返回的 dict 里放了一些无法序列化的对象比如一个数据库连接对象结果 Agent 在记录返回结果时直接序列化失败导致消息处理失败、进入重试循环。解决办法是 handler 的返回值只放基础类型让生产者和消费者用提前约定好的协议来传递业务信息——我在真实项目里普遍的做法是消息体就是 JSON所有字段都是字符串数字布尔值所有状态都靠约定而非对象引用。4. 踩坑实录与性能调优4.1 高频踩坑消息丢失、重复消费、死锁这几个问题几乎伴随着每一个初用消息代理的人。我把我遇到的高频问题整理成一张速查表每个问题都附上排查思路和解决方案。问题现象可能原因排查方法解决方案消息丢了存储后端未启用持久化检查 storage 配置确保 type 为 sqlite 或 redis不要用内存模式消息重复处理处理超时导致 ack 延迟查看 handler 耗时日志增加超时时间或者让 handler 具备幂等性队列堆积、消费变慢workers 数量不足监控队列长度调大 workers同时注意 prefetch 值服务重启后消息状态异常崩溃瞬间消息处于处理中状态数据库查看消息状态启用启动时自动恢复扫描处理中消息死信队列快速增长handler 代码 bug检查异常堆栈修复代码死信消息规范化记录我特别要强调的是重复消费这个问题的处理思路。因为 Hermes-Agent 的确认机制是超时确认如果 handler 处理时间超过了超时阈值消息会被重新投递给同队列里的其他 worker而原来那个 worker 可能还在处理中于是同一条消息被并发处理了两次。对于扣库存、转账这类敏感操作重复处理是不可接受的。这不能靠中间件层面完全解决必须业务处理器自己做到幂等。幂等的实现方式有很多种可以用数据库唯一约束比如订单号加类型做联合唯一索引也可以在消息体里带一个唯一消息 ID处理前查一下这个消息 ID 是否处理过还可以用 Redis SETNX 做一次性标记。我的实践经验是数据库唯一约束最可靠因为它是持久化的不会因为进程重启就丢掉标记。任何消费敏感业务消息的 handler都必须考虑幂等这不只是理论建议而是真实事故换来的教训。4.2 性能调优参数说明如果只是 demo 级别的消息量默认参数完全够用。但一旦进入生产环境每天百万级消息的话几个调优参数就非常重要了。第一个是 worker 并发数的调整。这里有个基本判断方法先看你的 handler 是 CPU 密集型还是 I/O 密集型。如果是 I/O 密集型比如调用外部 API、读写数据库worker 数可以调大因为大部分时间都在等待 I/O如果是 CPU 密集型比如大量计算或者编码转换worker 数不要超过 CPU 核心数太多否则频繁切换上下文反而降低吞吐。我自己的规则是I/O 型业务worker 数从 2 开始逐步增大观察 CPU 使用率CPU 型业务worker 数直接设为 CPU 核心数减 1。第二个是批量消费。Hermes-Agent 支持一次拉取多条消息批量处理配置项叫batch_size。这个参数对吞吐量提升非常明显因为它减少了消息处理的上下文切换和数据库状态更新的次数。我做过一次对比测试batch_size 从 1 调到 20同一批 10 万条消息的处理耗时从 8 分钟下降到了不到 3 分钟。代价是单条消息处理延迟变高因为要等凑齐一整批才会开始处理。对实时性要求不高的场景可以放心用但通知推送这种要求秒级延时的batch 就不要开了。第三个是存储性能。sqlite 在单机场景下很稳但写入量一大就会遇到锁竞争。我建议开启 WAL 模式它能让读和写并行显著改善并发处理能力。同时把synchronous设为NORMAL在每次写操作之间减少 fsync 次数换取更高的吞吐。再一个实用的做法是把消息内容和消息状态分开存消息内容字段可以很大状态字段很小拆成两张表之后状态更新的频率和速度都会好很多。4.3 监控与可观测性配置无论系统多小我都坚持从一开始就配置监控否则出了事故你连从哪排查都不知道。Hermes-Agent 自带一个监控端口暴露了一些指标默认在 9100 端口上配置项monitor.enabled打开之后就能用。核心指标有这么几个hermes_queue_depth队列当前积压消息数、hermes_processing_total正在处理中的消息数、hermes_retry_total累计重试次数、hermes_dlq_total死信队列累计数。用 Prometheus 定时抓取这些指标配合 Grafana 画一个简单的看板系统状态就一目了然。我个人的习惯是重点盯两个指标一个是队列深度另一个是死信队列变化率。队列深度代表当前积压的压力如果长期不降说明消费者速度跟不上生产速度需要扩 worker 或者优化 handler。死信队列变化率则更像一个紧急程度信号平稳时期几乎为零一旦突然上升基本可以认定有代码层面的问题。日志方面Hermes-Agent 默认输出到 stdout也支持配置日志文件。我建议在生产环境至少把日志级别调到 INFO并且按照消息 ID 做一个关联字段这样看日志的时候可以一条链路追到底。排查问题时我通常先按消息 ID 过滤出这条消息的完整生命周期——什么时候进来、被哪个 worker 领走、处理结果是什么——基本能快速定位问题出在哪一环。日志是排障的第一工具别省。5. 最后一点实际体会项目做完之后我最大的体会是中间件的价值不在功能堆砌而在于把工程复杂度藏好。Hermes-Agent 把消息投递、确认、重试、死信这一整套机制封装了起来留给你的是几个简单的配置项和处理函数这让业务代码能保持非常干净。但它也不是万能的遇到超大流量或者复杂路由需求该上重型消息队列的时候不要犹豫。选型之前想清楚自己的真实场景和资源约束比盲目追新技术重要得多。一个消息代理轻便顺手、出了问题你能看懂就比什么都强。
