做后端这一行早晚会被同一个问题找上门各种各样的系统都要发通知。监控要告警、订单要提醒、用户操作要通知管理员甚至连定时任务失败都要发一条消息。刚开始每个系统各搞各的邮件几封几封地进垃圾箱钉钉群里机器人多了之后除了刷屏什么都看不清真正重要的告警反而没人注意到。我给这套内部用的实时事件推送服务起了个代号叫 Buzz想法很简单所有需要“嗡嗡一声”提醒到人的消息通过它统一收口按规则路由到对的人手里。这篇文章就是记录 Buzz 从需求分析到落地实现的全过程包括关键选型、订阅规则设计、通道适配、部署参数还有几个月运行下来遇到的各种坑。如果你正在搭消息中心、告警平台或者只是想在系统里加一个统一的通知出口这份记录应该能帮你省掉不少弯路。1. 整体设计与思路拆解1.1 Buzz 想解决的就是消息到人的“最后一公里”从技术角度压缩成一句话Buzz 接收任意业务系统发来的事件根据订阅规则找到应该收到的人再通过邮件、Webhook、WebSocket 站内信、App 推送这些通道把内容送出去。它不关心事件具体是什么含义只关心事件该不该触达用户、通过什么方式触达、重不重要。很多团队一开始不觉得这是个需要单独建设的事。订单系统自己发邮件监控系统自己拉钉钉机器人用户系统自己写企业微信应用等系统多起来问题就暴露了消息格式不统一有的带 Markdown有的纯文本有的把日志堆栈整个贴过来没有统一的订阅和开关规则想停掉某类消息得去改业务代码没有运营视角通知发出去了既不知道哪些通道在悄悄失败也查不到某条消息到底有没有送达告警消息和普通消息混在一起真正要紧的事件被淹没Buzz 本质上是一个分发平台不是聊天软件也不是告警引擎它更像是快递中转站不管包裹来自哪家电商到了中转站就重新贴单按目的地分拣最后送到对应的快递员手里。边界一开始就必须划清楚否则后面很容易把业务逻辑塞进来越做越重最后变成一个什么都能干、什么都干不好的大泥球。1.2 三个影响全局的设计取舍第一个取舍是“推还是拉”。提醒类场景天然要求低延迟监控告警晚几秒都可能让事故扩大所以主链路必须推。但客户端已读状态、离线消息补拉这些逻辑可以做成拉取避免把推送平台做成强一致系统。Buzz 的主链路只保证“尽力送达”不保证“一定已读”已读状态由业务方自己管理。第二个取舍是“事件先落地还是先路由”。最开始我图省事想着事件进来直接改 WebSocket 并顺手发邮件就完事了后来发现通道一抖动消息全丢没有任何记录。最后改成所有事件先写 Redis Stream消费组再做规则匹配和通道分发。哪怕某个通道挂了事件还在队列里恢复以后可以重放。代价是延迟增加几毫秒但换来了可追踪性和可恢复性这笔账非常划算。第三个取舍是“统一框架还是各通道自治”。我的做法是每一个通道做成独立适配器统一传入一个结构体返回成功或失败以及重试信息通道内部怎么实现平台不管。这样新增一个通道只需要写一个新适配器完全不碰主流程。之前考虑过用消息队列自带的各种 connector 插件但大部分通道都涉及登录态、签名和限流自研适配器反而更可控。2. 核心细节解析与实操要点2.1 订阅规则设计用层级主题不要用正则订阅规则是 Buzz 的核心。业务系统发布事件时带一个主题比如monitor.api.latency_high订阅者则声明自己关心哪些主题。我采用的是类似 MQTT 的层级通配符方案而不是正则表达式。原因是正则匹配灵活但没法高效索引主题数量一多全量正则扫描很快就扛不住层级通配符可以用前缀树或按段匹配性能稳定而且规则清晰可读。具体规则分两级。第一级是主题匹配决定这条消息能不能路由给某个订阅者第二级是事件级别门槛比如只有fatal和error才触达info默认不推。这样配置的人不需要理解复杂的规则引擎运维成本低很多。一个子系统的订阅配置大概是这样的subscribers: - name: oncall-backend topics: - monitor.api.* - monitor.redis.* min_level: error channels: [webhook, email] - name: order-ops topics: - order.* min_level: warn channels: [webhook]有一点很容易被忽略同一个订阅者命中多个主题时要合并成一条消息还是逐条推送我默认逐条推但又加了聚合窗口。同一订阅者在一分钟内的相似消息会被聚合成一条摘要避免告警轰炸。比如 Redis 连接抖动时可能一次性上报 30 条异常聚合后只发一条“Redis 连接异常 30 次最近一条: xxx”。告警系统最忌讳的就是刷屏刷屏会导致真正重要的消息被忽略。2.2 通道适配器与重试机制通道适配器是消息送出去的最后一环也是最不可控的一环。邮件可能拥堵Webhook 目标服务可能超时WebSocket 客户端可能断线。所以我把每个适配器都封装成“提交任务”模式主流程只把消息投递给适配器的内部队列适配器自己处理发送、超时和重试。重试必须用指数退避加抖动这里解释一下为什么。如果对方系统已经出问题固定频率的重试只会加重故障最常见的就是邮件网关和 Webhook 目标同时被搞挂。指数退避能把重试压力摊开加抖动是为了避免多个 worker 在同一时刻重试形成定时共振。我用的公式是min(30秒, 2^n * 5秒 random(0, 3)秒)总共重试 5 次之后进入死信队列。跨通道幂等同样重要。发送动作不是天然幂等的网络超时的时候消息其实已经发出去了但适配器没收到响应如果重发就会造成用户收到两条。解决办法是每条消息带一个全局唯一requestId在发送目标里附带上对方可以做幂等对于邮件、企微机器人这种不归自己控制的通道只能在平台层面做去重窗口。我在 Redis 里用setnx存requestId窗口 60 秒保证异常重投时重叠时间内只放行一次。2.3 WebSocket 在线推送容易踩的隐性坑WebSocket 是 Buzz 给站内提示用的主通道看起来简单维护起来问题不少。最常见的是客户端担心自己在线但服务端其实早就不知道它在线了。Nginx 默认 60 秒没有读写就断开连接客户端如果只是挂在页面上不操作连接会被切断客户端自己也不一定立刻感知。所以必须在协议层加心跳服务端和客户端都要做服务端 30 秒 ping 一次客户端收到 ping 后回 pong超过两个周期没等到 pong 就主动关闭。心不只是为了保活更重要的是让双方都能快速发现连接已死及时清理资源和重建连接。第二个坑是 WebSocket 连接状态不能只记在单机内存里。一个用户可能同时开着两个标签页也可能断线重连之后旧连接还没回收如果状态不集中管理消息就会重复推送或推给废弃连接。Buzz 里加了一层轻量级的连接注册表放在 Redis hash 里只记录userId、连接 ID、节点 ID、最后心跳时间。推送时先查注册表再通过对应节点转发。这样节点重启后连接可以快速重建不必要的残留会话也不会一直占内存。3. 实操过程与核心环节实现3.1 最小可运行版本的技术栈选择选技术栈时我给自己定了几个约束不引入太重的基础设施、团队成员都很熟、IO 密集场景性能足够。最后敲定的是 Node.js NestJS 作为 API 层Redis Stream 作为事件缓冲和消费者组载体EMQX 作为 MQTT brokerWebSocket 网关和服务跑在同一套服务里用nestjs/websockets暴露。有人可能问API 层为什么不选 Go坦白说纯性能要求下 Go 确实更稳但当时团队的迁移成本、NestJS 的模块化结构、与现有代码生态的衔接更重要。而且 Buzz 这种系统瓶颈基本都在外部通道和网络带宽上API 逻辑本身没有特别重的计算任务Node.js 在单机十万级连接下完全够用。架构选型永远要结合团队现实不能只看排行榜。目录结构大致如下src/ events/ # 事件接收与校验 rules/ # 订阅规则匹配 channels/ # 邮件/webhook/ws 适配器 streams/ # Redis Stream 生产与消费 ws/ # WebSocket 网关与注册表 config/ # 规则与通用配置3.2 事件接收、规则匹配、分发主流程代码事件接收接口要做得足够简单任何系统拿到一个 URL 和一对 token 就能接入不需要安装 SDK也不需要理解内部细节。核心逻辑如下// events/receive.controller.ts 片段 Post(event) async receive(Headers(x-access-token) token: string, Body() payload: EventPayload) { const app await this.appRepo.findByToken(token); if (!app) throw new UnauthorizedException(); this.validate(payload); // 必须包含 topic, level, content await this.stream.produce(buzz:events, { eventId: randomUUID(), appId: app.id, topic: payload.topic, level: payload.level, content: payload.content, timestamp: Date.now(), }); return { accepted: true, eventId }; }规则匹配是在消费组里做的。把事件丢进 Redis Stream 后多个 worker 可以并发消费互不干扰。匹配时用事件主题逐段与订阅者规则通配符对比命中之后按该订阅者配置的channels列表投递到对应队列。// worker.ts 消费逻辑片段 async function consume(streamEntry) { const matches ruleEngine.match(event.topic, event.level); for (const sub of matches) { for (const ch of sub.channels) { await channelQueues[ch].enqueue({ event, subscriber: sub }); } } await ack(streamEntry); }这里有几个细节值得单独说。一是 ack 的位置必须等所有通道都入队成功之后再 ack否则数据会丢。入队成功不代表发送成功但它已经进入适配器的可靠队列适配器自己负责重试。二是 worker 数量不是越多越好因为 Redis Stream 的读吞吐和规则匹配耗时决定合理并发数。我实测单机 16 个消费者进程效果最好加到 32 个反而会因为上下文切换和 Redis 连接争用导致吞吐量下降。3.3 docker-compose 部署与关键参数为了让整套 Buzz 能一键拉起我写了一个 docker-compose 文件。它主要面向开发和快速验证生产环境至少要把 Redis 做成集群或哨兵模式MQTT broker 也要做 HA否则 broker 挂掉后所有在线推送都会断。version: 3.9 services: redis: image: redis:7-alpine command: redis-server --appendonly yes ports: [6379:6379] mqtt: image: emqx/emqx:5.4 ports: [1883:1883, 18083:18083] buzz-api: build: . ports: [3000:3000] environment: REDIS_URL: redis://redis:6379 MQTT_URL: mqtt://mqtt:1883 BUZZ_WORKERS: 16 depends_on: [redis, mqtt]Redis 参数里appendonly必须开否则 Redis 进程重启会丢事件整个可靠性设计就白做了。Stream 的消息保留时间我设置为 7 天因为很多告警需要事后查询超过 7 天意义不大。消费者组的 pending entries 要定时清理否则积压太多会导致XREADGROUP变慢。Node.js 服务的生产者写入 Stream 失败时要重试但次数不宜过多。我以前设过无限重试结果 Redis 短暂抖动时生产者进程里堆积了大量 Promise事件循环被打满最后连健康检查都过不去。现在的做法是生产失败重试 3 次仍然失败就写本地日志并返回非 200 状态码让调用方决定稍后重试还是丢弃。3.4 简单压测结果与容量规划我在本地用 wrk 压过 API 层单机 4 核 8G 的容器里事件接收接口稳定跑到 1.5 万 QPS规则匹配和入队占用的 CPU 大约 30%。真正的瓶颈不在 API 层而在 WebSocket 推送和邮件发送。WebSocket 单节点大约能撑住 8 万活跃连接再往上会明显感受到 Node.js 事件循环和内存占用不太稳定。容量规划上我建议按事件量而不是用户量来预估。一天 1000 万事件和 1 万人在线是完全不同的压力模型。每个人会产生多少条触达消息、多少条需要立即推送这些指标才决定系统要不要分片。Buzz 目前的扩展方式是 API 层无状态随便扩Redis Stream 做分片WebSocket 节点通过注册表做水平扩展。真到单 Redis 扛不住的时候可以按appId对 Stream 做分片这个方案已经在测试中但不建议一开始就上运维复杂度会增加不少。4. 常见问题与排查技巧实录4.1 消息重复推送一个从 ack 位置开始的坑做过消息系统的人十有八九都见过重复消息。我第一次把消费逻辑写完后发现邮件经常同一封发两次。查了半天发现是 ack 写得太早——worker 从 Stream 里读出来就 ack但后续分发抛了异常消息自然就丢了。于是我把 ack 挪到最后结果又出现新问题入队成功后、ack 前 worker 崩了重启后消息会被重新消费于是重复。这就是消费系统的本质at-least-once 语义下无法完全避免重复只能靠幂等和去重。Buzz 的处理思路是平台层不做“绝对不重复”这种不可能实现的承诺而是做到“尽量少重复重复了也不伤”。事件入队时带上eventId各适配器发送前先检查 Redis 里的requestId60 秒内重复的不发。邮件通道没法让邮件服务器替我们做幂等所以也只能靠这个窗口来兜底。排查这类问题时最有效的工具是全链路日志都带eventId没有它你很难定位是生产端重复还是消费端重试导致的。4.2 在线用户收不到消息心跳与 Nginx 超时的双重陷阱有一次客户反馈在线提示经常收不到但是刷新页面后又好了。这就是典型的 WebSocket“僵尸连接”。客户端页面挂着Nginx 在 60 秒后切断了连接但页面里的 JS 没有触发onclose服务端也没感知到。后来在两处做了修正一是 Nginx 针对 WebSocket upgrade 单独设置了更长的超时时间不在全局配置里一刀切二是应用层强制心跳30 秒 ping、45 秒没收到 pong 就主动关闭并清理连接注册表。修完之后这类问题基本绝迹。排查技巧是如果客户端状态和服务端连接状态对不上先查 Nginx 错误日志里有没有upstream timed out有的话八成是负载均衡层超时。然后再看服务端有没有收到close事件如果服务端一直认为连接还在那就是应用层心跳缺失。按这个顺序查五分钟内通常能定位。4.3 消费堆积当慢通道拖垮整个事件流健康状态下 Stream 里的消息进出是平衡的一旦某个外部通道变慢比如对方 Webhook 连续超时适配器的内部队列就会积压消费组占用的时间变长最后 Redis Stream 里未消费消息急速增加。我遇到过邮件通道在高峰期堵了半小时消费者进程跑满API 响应时间跟着飙到 5 秒。这种问题不能只靠加 worker因为瓶颈在外部通道的吞吐上。我当时做了两个调整一是给每个通道单独设置并发上限邮件并发 10、Webhook 并发 50超过就排队让慢通道不影响其他通道二是给通道增加熔断连续失败率达到 50% 就暂停该通道 60 秒消息直接进入死信等外部恢复后再手动重放。效果很明显慢通道被隔离后主要消息分发的延迟恢复了正常。这些指标最好提前观察起来不能等堆积发生了才去止损。4.4 问题排查速查表下面这个表是 Buzz 运行几个月中遇到的主要问题、排查路径和解决办法可以直接拿去做参考。现象常见原因排查手段解决办法消息重复推送ack 时机错误、通道超时重试日志里追 eventId全局 requestId 去重窗口 60 秒在线用户收不到提示Nginx 超时断开、僵尸连接查 Nginx 日志、心跳日志应用层 30 秒 ping 清理失效连接事件堆积延迟上升慢通道拖住消费组看 stream 长度和各通道队列长度通道并发隔离 熔断死信 手动重放WebSocket 断线重连频繁客户端网络切换、服务器超时抓心跳包、看 close code指数退避重连不要固定 1 秒重连某个应用全部消息没有分发token 失效、规则没刷新查应用认证日志、规则版本token 校验失败时返回明确错误码并告警关于重连策略再补一句很多客户端只在检测到断线后立刻重连服务器重启时所有客户端瞬间涌入很容易把服务打崩。正确的做法是第一次重连延迟 1 秒之后每次翻倍最多到 30 秒再加上随机抖动。所谓重连风暴的应对不加抖动的重连策略在生产环境迟早会踩一次。我实际做下来最大的体会是这种系统难的不是“实时推送”这个词而是把各种异常情况都接住。消息可能重复、通道可能变慢、连接可能变成僵尸、Redis 也可能抖动。Buzz 本身的代码并不复杂真正的复杂度都在这些边界里。如果你也想搭一套类似的东西建议先把消息模型、订阅规则和通道抽象定义清楚再考虑技术选型顺序反了会走很多弯路。最后再分享一个小技巧全链路日志里一定要带上eventId它是排查消息类系统最省时间的东西没有之一。