做Web实时消息推送这几年我最大的感受是广播很容易人人都会私发才是真正考验设计功底的地方。很多人以为私发就是“给某个session发一条消息”做出来之后一上集群就凉了session在A机器消息却发到了B机器用户啥都没收到。这篇文章就围绕私发和广播这个主题把我实际落地过的设计方案、踩过的坑、以及一些常规文档里不会写的细节一次性讲清楚。1. 为什么私发比广播难本质是寻址与扇出的差异先问一个问题广播消息服务端只需要往所有连接推一遍逻辑上无脑但代价高私发消息服务端必须精确找到“这个用户当前在哪台机器、哪条连接上”然后只推给那一条。所以私发的核心不是“发”而是“找”。找错了消息就丢了找慢了消息就延迟找到了但连接已经断了消息还是丢。广播的核心则是“量”连接越多扇出压力越大怎么控制广播范围、怎么避免重复投递这才是广播的难点。在我参与过的几个消息推送类项目里第一版几乎都是“一个全局Map存session”的单机方案用户量和连接数上来之后Map本身没问题问题出在横向扩展。你不能把Map分给多台机器因为用户可能连到任意一台。这时候就需要引入一个“路由层”让每台机器知道自己持有哪些连接同时让“消息入口”知道该把私发消息投到哪台机器。私发和广播的架构差异可以这样理解广播像在广场上喊话谁都能听见你只需要一个大喇叭私发像快递员送件你必须知道收件人住在哪栋楼哪个房间手里还得有钥匙。这个“住址信息”就是会话路由表而这个“钥匙”就是连接的唯一标识。整套设计就是围绕这张路由表展开的。2. 选型思路为什么我最终选了WebSocket加消息中间件2.1 传输层选型浏览器端的实时推送现阶段基本没有比WebSocket更合适的方案。HTTP短轮询延迟高、浪费带宽SSE虽然轻但只能服务端单向推双向通信还得再开一条通道。WebSocket一次握手建立长连接客户端和服务端都能主动发消息天然适合聊天、通知、协同编辑这类场景。如果你的客户端是移动App可以考虑Netty自研协议或者直接用MQTT如果是纯浏览器端我的建议是别自己造协议轮子直接用WebSocket协议Java后端用Spring的WebSocket支持或者Netty都行。Spring的WebSocket封得比较薄简单场景够用要精细控制心跳、半包、粘包、流量控制就用Netty。我实际项目里用的是Netty原因后面会讲。2.2 消息投递层的选型私发消息必须做跨节点路由跨节点通信最稳妥的方式就是消息中间件。选型的核心指标有三个延迟、有序性、吞吐。聊天类场景对有序性要求高不能用户发了两条消息后发的那条先到通知类场景对延迟敏感消息最好毫秒级触达推送量大时吞吐必须扛得住。我比较常用的方案是RocketMQ原因是它支持有序消息和事务消息而且消费者端可以做到精准的Tag过滤。如果项目里已经用了Kafka也可以用但Kafka在有序性和重复消费方面的表现不如RocketMQ需要自己做去重和排序。RabbitMQ适合轻量级场景延迟低但集群模式下消息堆积能力不如RocketMQ和Kafka。2.3 为什么不直接用分布式缓存做路由表有人会问路由表能不能放到Redis里用户ID对应机器地址每次私发先查Redis再转发不是更简单吗理论可行但实际用下来有两个问题一是多一跳延迟每次私发都多一次Redis查询二是连接状态变化太快连接建立、断开、心跳超时都要同步更新Redis一致性很难保证。所以我最终采用的是“节点内存持有路由消息中间件做路由分发”的方案。每个推送节点自己维护一份连接注册表只记录连到本机的session私发消息进入消息中间件后根据目标用户找到对应节点转发到那个节点上由节点从本地注册表里查出session并下发。这样查路由是本地内存操作速度极快也不会出现Redis缓存和实际连接不一致的问题。3. 连接接入与本地Session管理一切的地基3.1 WebSocket握手与鉴权绑定私发和广播的前提是连接必须“认识你”。用户建立WebSocket连接时不能只把连接存起来就完事必须完成一步关键操作鉴权并绑定用户身份。WebSocket握手本质上还是HTTP请求所以可以在握手阶段拿到Token。我的做法是在URL参数里带Token形如ws://gateway.example.com/ws?tokenxxxx服务端在握手拦截器里解析Token、校验有效性然后取出userId放到WebSocketSession的属性里。这一步不要省手写协议也要做类似的身份绑定否则后面私发无从谈起。握手成功之后连接就和一个userId建立了映射。一个用户可能有多个连接PC端、手机端、多标签页所以映射关系不能是一对一而是一对多也就是一个userId对应一个连接集合。3.2 本地Session注册表的结构设计本地注册表用什么样的数据结构直接决定并发读写性能。我最终用的是ConcurrentHashMap加嵌套Set的结构userId - SetSession用于私发知道用户找出他所有的连接。sessionId - userId用于反向查找收到断连事件时知道这个连接属于谁。sessionId - Session用于关闭连接时快速拿到Session对象。代码结构大致如下public class LocalSessionRegistry { private final ConcurrentMapString, SetWebSocketSession userSessions new ConcurrentHashMap(); private final ConcurrentMapString, String sessionUserMap new ConcurrentHashMap(); public void addSession(String userId, WebSocketSession session) { sessionUserMap.put(session.getId(), userId); userSessions.computeIfAbsent(userId, k - ConcurrentHashMap.newKeySet()) .add(session); } public void removeSession(String sessionId) { String userId sessionUserMap.remove(sessionId); if (userId ! null) { SetWebSocketSession sessions userSessions.get(userId); if (sessions ! null) { sessions.removeIf(s - s.getId().equals(sessionId)); if (sessions.isEmpty()) { userSessions.remove(userId); } } } } public SetWebSocketSession getSessions(String userId) { return userSessions.getOrDefault(userId, Collections.emptySet()); } }这里有一个细节userSessions的value是SetWebSocketSession为什么用Set因为同一个Session断线重连、心跳超时再恢复时可能出现重复注册Set天然去重。另一个细节是remove的时候要同时更新两个Map顺序上先删索引再删集合避免并发下查到半条脏数据。3.3 全连接注册表与在线状态维护除了按用户索引通常还需要一个“全连接注册表”用于广播场景。广播必须遍历所有连接如果只维护userId - SetSession广播时要先遍历所有userId再遍历每个Set效率低而且可能重复。我另外维护一个ConcurrentMapString, WebSocketSessionkey是sessionIdvalue是Session对象广播直接遍历这个Map的values。在线状态也是围绕这两张表算出来的。用户是否在线就看userSessions里有没有这个userId且Set非空。用户在线数就是userSessions.size()。判断连接是否健康靠心跳机制不是靠连接有没有close。这一点很重要TCP连接可能半开服务端不知道客户端已经走了必须靠Ping/Pong来探活。建连和断连的关键事件都要对外广播。我是通过一个本地事件总线发布UserOnlineEvent和UserOfflineEvent让业务方订阅这样“用户上线”“用户离线”这种信息可以复用到在线状态展示、消息补推、离线推送等等。不要把所有逻辑都堆在连接管理器里事件化之后扩展起来舒服很多。4. 私发链路的完整流程与关键细节4.1 从业务消息到下行消息的消息结构私发消息不能直接把业务对象往连接里塞需要封装成统一的“下行消息结构”。我常用的结构包含这些字段消息ID、消息类型、发送者、接收者、消息内容、时间戳、消息序号。public class PushMessage { private String msgId; private String msgType; private String fromUser; private String toUser; private String payload; private long timestamp; private long seq; }msgId是全局唯一ID用来做幂等seq是同一个发送者内的消息序号用来做排序和去重。消息进入推送系统后至少要保证投递一次所以msgId必须生成否则消费端无法识别重复消息。消息体不建议太大。推送网关只负责搬运不负责业务解析所以payload建议是序列化后的JSON字符串。如果消息超过一定阈值比如64KB应拆分或走对象存储通道别硬塞进消息中间件。4.2 单机私发的处理路径单机场景下的私发流程相对简单但却是理解整体链路的基础。以“用户A给用户B发消息”为例完整路径是这样的业务服务调用推送API构造PushMessage调用推送网关的私发接口。推送网关根据toUser查询本地路由表。如果B在本机的注册表中存在直接取出其Session集合调用sendMessage下发。如果B不在本机说明B连接在其他节点此时需要把消息投递到消息中间件的私发Topic。这个分支是私发链路里最关键的一点先查本地本地没有再发中间件。不要一上来就把所有消息发给中间件让中间件再路由回本机那样白白多一跳。我见过不少实现是“所有消息都走MQ”结果同机消息延迟白白多了几十毫秒集群规模一大人均延迟就上去了。4.3 私发时多连接的处理策略一个用户多个连接时私发消息应该发给哪几条我的默认策略是全部下发也就是同一个用户的所有活跃连接都收到。这样做的好处是实现简单、不会漏消息用户切设备时体验也一致。但代价是多端会重复展示比如PC和手机同时收到同一条聊天消息。有些场景只想推给“最近活跃的那一个连接”比如登录提醒、安全验证码。这时可以给Session增加一个lastActiveTime字段每次收到客户端消息或者心跳时更新私发时按时间取最近的一个连接。这种策略适合一次性通知但要做容错如果最近连接下发失败要自动降级到其他连接。私发下发时还有一个常见问题连接虽然活着但消息发送缓冲区已满sendMessage抛出异常。Netty的channel.writeAndFlush是异步的不能只看返回值要监听ChannelFuture的完成回调失败时记录日志并触发重试或补偿。Spring的WebSocketSession.sendMessage虽然是同步阻塞式的但底层也有缓冲区限制并发下发送过快会抛IllegalStateException。4.4 私发失败时的补偿策略私发失败的场景很多目标用户离线、连接断开、发送超时、消息中间件消费失败。我在设计时统一走“失败进重试队列”的方案。连接不可达用户离线如果业务允许离线消息把消息转存到离线消息表用户下次上线时补推如果业务不允许离线消息直接丢弃并记录日志。连接存在但发送失败将消息放入本地重试队列退避重试重试超过次数后转离线或者丢弃。消息中间件消费异常利用MQ本身的重试机制消费失败时返回RECONSUME_LATER让Broker稍后再投。重试一定要有次数上限和退避策略不能无限重试。我一般设定最多重试3次退避间隔为1秒、5秒、30秒超过则进入死信Topic人工介入排查。这个机制很笨但很可靠能挡住大部分偶发故障。5. 广播的实现方案从全量推送到按需收敛5.1 广播的正确姿势每个节点各扫门前雪广播的核心思想是消息不是从一台机器推给所有连接而是让每个节点消费同一条广播消息然后推送给自己节点上的连接。这样就把“一个中心对N个连接”的压力分摊成了“每台机器对自己本地连接”的并行任务。具体做法是广播发起方往消息中间件的广播Topic发一条消息所有推送节点都订阅这个Topic各自消费遍历本地注册表向自己持有的连接下发。节点数越多广播总吞吐越大扩展性远好于单机遍历。这里需要特别提醒广播Topic的消费组所有推送节点必须属于同一个消费组这样每条消息才能被每个节点各消费一次。很多初学者把每个节点配成独立消费组结果同一节点集群里的多台机器抢消息广播变成了一台机器独播。5.2 广播范围控制别让一条消息炸穿所有连接不是所有广播消息都需要推给全量用户。系统公告、全员通知通常是全量广播但运营活动、定向通知可能只推给部分用户。这就需要给广播消息增加范围属性。我常用的做法是在广播消息里增加一个targetFilter字段携带过滤条件比如人群标签、用户等级、在线状态。消费者端先判断目标连接是否满足过滤条件再决定是否下发。注意这类过滤必须在各个节点本地做不能在发件方做否则又要回到“中心化遍历”的陷阱里。范围控制还有一个重要场景消息太多时广播所有用户会把网关打爆。比如秒杀、大促场景几百万连接同时推送即使每台节点只推本地连接网卡和内核中断也可能先撑不住。这种时候要做“推送速率控制”和“推送量分级”要么限制每秒推送的连接数要么按批次推送、隔几秒推一批。全量广播最怕瞬间峰值峰值一来业务没挂网关先挂了。5.3 广播和私发共用一个下行通道广播和私发的下发动作最终都是“向一条WebSocket连接写入消息”区别只在上游消息来源不同。所以我在设计时把下行通道抽象成了一个Sender接口提供sendToUser和sendToAll两个方法内部都走同一个Session写入逻辑。public interface PushSender { boolean sendToUser(String userId, PushMessage message); int sendToAll(PushMessage message, Filter filter); }这样广播和私发共用一套序列化、压缩、流量控制、失败处理逻辑。不要各写一套否则后期维护成本翻倍。下行通道里我加了一个ChannelGroupNetty里的概念来管理所有连接广播时直接往ChannelGroup写入并发安全私发时先查userSessions拿到Channel再单独写入。5.4 广播消息的幂等与顺序广播消息经过消息中间件理论上存在重复投递可能。消费者端要做幂等同一个msgId最多处理一次。实现上可以引入一个本地去重Map记录近期处理过的msgId也可以用Redis的SETNX做全局去重。我倾向于本地去重因为广播消息量很大全部打到Redis成本高而且广播消息的重复影响相对可控本地去重窗口设置为几分钟就够了。广播消息的顺序同一业务场景下要求高的要保证有序。比如股票行情、赛事比分后发的比分必须覆盖先发的。消息中间件里可以用“消息Key 分区有序”来保证同一业务类型的广播消息有序。这个细节容易忽略等出了问题再补救很麻烦。6. 集群场景下私发与广播的路由分发实践6.1 节点注册与心跳上报集群里每个推送节点必须让其他模块知道自己的存在。我采用的方式是节点启动时在注册中心登记自己的节点ID、IP、端口、负载信息并定期上报心跳。这里的注册中心可以是一套分布式协调服务也可以是一张数据库表加定时刷新关键是要能让“消息入口”找到“目标节点”。私发消息进入消息中间件后消费端需要知道“目标用户连在哪个节点上”。我在实践中用的是“用户最近活跃节点”的映射也就是在用户建立连接时把userId和节点ID的关系写到一个路由表里。路由表存储在分布式缓存中但只做“定位用”不做“实时状态用”因为实时状态以节点本地的session注册表为准。这样即使路由表稍微滞后也不会导致消息完全丢失最多是投递到旧节点后转发一次。6.2 消息中间件的Topic规划Topic规划直接影响私发和广播的性能与运维复杂度。我的推荐规划是私发Topic按userId哈希取模分桶保证同一个userId的消息进同一个分区从而实现有序消费。广播Topic独立Topic不分区或少量分区因为广播消息不需要强顺序每个节点消费所有分区即可。系统内部事件Topic用于节点上下线通知、路由变更、连接状态变更等这类消息量小但要保证可靠投递。不要把所有消息都塞进一个Topic。混用之后广播高峰会拖慢私发私发高频会淹没广播问题排查时也无法区分消息类型。分Topic不仅是为了隔离流量更是给运维留了独立的扩缩容和监控维度。6.3 私发跨节点转发消息从一台机器到另一台机器跨节点私发是最容易被忽视的实现细节。举一个具体场景用户A连接在节点1用户B连接在节点2。业务服务调用推送API时请求可能被负载均衡到任意一台节点。如果请求落在节点1节点1发现B不在本地就把消息投递给私发Topic如果请求落在节点2节点2发现B就在本地直接下发。但这里有个问题如果消息投递到私发Topic后被消费的节点恰好不是目标用户所在的节点呢比如消息投递到Topic的某个分区这个分区的消费者是节点3而用户B在节点2——消息就会被节点3消费但它发现B不在本地怎么办答案是继续转发而不是丢弃。节点3消费到不属于本节点的私发消息时根据路由表查到B在节点2然后把消息转投到节点2的本地队列或者重新投递到私发Topic的对应分区。这个“二次路由”逻辑是集群私发能正确工作的关键。如果不做二次路由私发消息在负载均衡下会出现严重的丢失概率。6.4 路由表的一致性维护消息短暂丢失可接受长期错乱不可接受路由表不能保证100%实时因为节点可能在消息投递瞬间刚刚重启或者用户刚刚断开连接。设计目标是消息不能长期错乱短暂的重试和转发可以接受。所以路由表更新时机是三个连接建立时写入、连接关闭时删除、节点心跳异常时批量清理该节点下所有用户。路由表删除要小心节点宕机时来不及清理连接会导致路由表指向死节点。我的方案是消费端发现目标节点不可达时把该节点所有用户的路由记录标记为过期同时触发一次全集群的路由刷新。这个兜底逻辑比单纯依赖“正常断连”可靠得多毕竟宕机是常态优雅退出才是意外。7. 广播风暴、消息挤压和重复下发的三个真实坑7.1 广播风暴一条广播消息触发N条私发推送广播风暴是我实际遇到过的第一个线上事故。当时做一个全员通知功能运营发了一条“所有用户”的公告推送网关接到的广播消息消费后每个节点往下推这本来没问题。问题出在业务侧回调客户端收到公告后自动请求了某个业务接口这个接口又触发了多条私发通知私发通知再触发客户端请求……等于广播引爆了成倍的私发流量几分钟内网关负载翻了好几倍。这类问题很难在测试环境复现因为量级不够。规避方式有几个一是广播消息的下行推送不要携带过多的业务动作客户端收到后只展示不自动请求业务接口二是推送网关做全局流量监控私发QPS突增时触发限流三是对广播消息加“生效时间”和“撤回时间”防止意外重复推送。技术之外这其实是个业务设计问题但网关侧一定要留限流和熔断的开关。7.2 消息挤压消费速度跟不上推送速度推送场景的消息挤压经常不是因为MQ消费慢而是因为下行通道慢。每个节点消费消息后要遍历本地session逐个下发每写一个session都有网络IO和锁开销。如果一条广播消息有10万连接节点要写10万次这个耗时可能长达几十秒。期间如果MQ消息堆积消费者线程会把消息拉到本地内存内存暴涨GC频繁节点进入恶性循环。解决思路是把“消费消息”和“写入session”解耦。消费者只负责把消息放入本地队列后台线程池负责实际写入。线程池大小根据节点连接数动态配置一般建议核心线程数不超过CPU核数的两倍。还要注意背压本地队列不能无限增长超过阈值就丢弃或者降级。不然消息中间件没堆积推送节点自己倒是先OOM了。7.3 重复下发网络重试和消费重试叠加重复下发最隐蔽的来源是“客户端没收到ACK”引发的服务端重试和“MQ消费失败”引发的消息重投两者叠加。假设一条私发消息下发时网络超时服务端重试了一次同时MQ也重投了一次客户端就收到了两条。除非业务天然允许重复否则必须在客户端做去重。客户端去重的最简单办法是根据消息的msgId做本地去重已经处理过的消息ID直接丢弃。服务端能做的是尽量保证语义上的“至少一次投递”而不是“精确一次”。在这个前提下客户端去重就是最后的防重复防线一定要加。哪怕服务端已经做了很多幂等处理客户端去重仍然值得写这不是浪费是保险。8. 在真实项目中测试这套方案的性能与稳定性测试环境只能验证功能是否正确性能与稳定性必须用压测和故障演练来验证。我搭建了一套三节点的推送集群模拟5万在线连接做私发与广播混合压测重点观察三个指标私发P999延迟、广播全量推送完成时间、节点内存和GC情况。私发压测结论是单节点每秒可以处理约2万条私发下行P999延迟控制在20毫秒以内。瓶颈不在业务代码而在Channel写入的锁竞争和系统调用的开销。广播压测结论是5万连接全量推送一条1KB消息三节点并行大约需要2到3秒完成随着节点增加完成时间基本线性下降。GC方面频繁创建PushMessage对象会产生较大压力优化方式是复用对象池或者用堆外内存封装实测GC耗时降低了约一半。故障演练我做了两组一组是随机杀掉一个节点观察私发消息是否丢失另一组是消息中间件消费延迟30秒观察推送节点是否堆积。第一组的结果印证了“二次路由”的必要性如果杀掉的是用户所在节点路由表未及时清理前私发消息会先到死节点再转发靠转发容错最终消息仍然能到。第二组的结果表明消费侧本地队列的背压机制是必须的不加背压节点内存会快速上涨恢复时间会显著变长。这里也顺带说明一下以上数据来自我自己的测试环境不同机器配置、不同网络环境结果会有差异。但趋势是稳定的方案能支撑“万级连接到十万级连接”的私发和广播场景超过十万级连接或者更高的QPS要求需要进一步做分片、压缩和内核参数调优。9. 容易忽略的细节与常见误区9.1 不要把WebSocketSession等同于用户连接Spring的WebSocketSession代表一次WebSocket握手会话但一个用户可能开多个标签页所以userId和session是一对多的。我之前见过有人用userId - WebSocketSession的Map多标签页用户互踢这是最常见的私发错误之一。所有的设计都以“一对多”为前提才能支持多端登录。9.2 连接关闭不一定是主动关闭心跳必须独立设计WebSocket的close事件只在正常关闭或对端发close帧时触发。网络闪断、客户端崩溃、App切后台被系统回收这些场景下服务端不会立刻感知连接已经不可用。必须设计心跳客户端定时发Ping帧或业务心跳服务端记录最近活跃时间超过阈值判定失活并清理。清理时要同步注销路由表否则私发消息会往死连接上打。9.3 不要依赖线程睡眠做限流做广播限流时我见过有人用Thread.sleep控制写入频率这是非常糟糕的写法。它会占用线程资源拖垮整个节点。正确做法是使用令牌桶或者信号量控制并发写数超出的消息直接进等待队列或丢弃。Netty本身有写缓冲区的水位控制设置WRITE_BUFFER_WATER_MARK可以避免单个慢客户端拖垮整个节点。9.4 别忽略消息大小的限制单条消息过大会导致WebSocket帧超过最大限制客户端解析失败。Netty的默认最大帧大小是64KB超过会抛出异常。广播场景里一条全员消息很容易因为包含大段富文本或者图片Base64而超限。建议在下行前统一做消息大小检查超限消息单独走文件或对象存储通道推送时只推一个下载链接。10. 这套方案的完整代码骨架为了便于大家直接上手我给出一个精简可扩展的代码骨架核心点都包含在内。完整的可用代码还要结合具体的项目框架和业务场景但骨架的逻辑可以作为参考起点。// 消息推送网关入口 Controller RestController RequestMapping(/push) public class PushController { private final PushService pushService; public PushController(PushService pushService) { this.pushService pushService; } PostMapping(/private) public ResponseEntityVoid sendPrivate(RequestBody PrivatePushRequest request) { pushService.sendToUser(request.getUserId(), request.getMessage()); return ResponseEntity.ok().build(); } PostMapping(/broadcast) public ResponseEntityVoid sendBroadcast(RequestBody BroadcastPushRequest request) { pushService.sendToAll(request.getMessage(), request.getFilter()); return ResponseEntity.ok().build(); } }// 推送服务核心逻辑 Service public class PushService { private final LocalSessionRegistry registry; private final MqProducer mqProducer; public PushService(LocalSessionRegistry registry, MqProducer mqProducer) { this.registry registry; this.mqProducer mqProducer; } public void sendToUser(String userId, PushMessage message) { SetWebSocketSession sessions registry.getSessions(userId); if (!sessions.isEmpty()) { // 本地直发不走MQ for (WebSocketSession session : sessions) { try { session.sendMessage(new TextMessage(message.getPayload())); } catch (IOException e) { // 发送失败进入重试或补偿逻辑 } } } else { // 本地无连接投递到MQ等待跨节点转发 mqProducer.sendPrivateMessage(message); } } public void sendToAll(PushMessage message, Filter filter) { // 广播消息只发到MQ每个节点消费后再本地扇出 mqProducer.sendBroadcastMessage(message); } }// MQ消费者私发消息的跨节点路由消费 Component public class PrivateMessageConsumer { private final LocalSessionRegistry registry; private final UserRouteTable routeTable; private final PushService pushService; public void onMessage(PushMessage message) { String userId message.getToUser(); SetWebSocketSession sessions registry.getSessions(userId); if (!sessions.isEmpty()) { // 目标用户在本节点直接下发 pushService.sendToUser(userId, message); } else { // 目标用户不在本节点查询路由表并转发 String targetNode routeTable.lookup(userId); if (targetNode ! null) { mqProducer.forwardToNode(targetNode, message); } } } }// MQ消费者广播消息的本地扇出 Component public class BroadcastMessageConsumer { private final LocalSessionRegistry registry; private final MessageSender sender; public void onMessage(PushMessage message) { // 本地遍历所有session按过滤器决定是否下发 for (WebSocketSession session : registry.getAllSessions()) { if (filter.accept(session)) { sender.send(session, message); } } } }骨架里最需要留意的就是私发消费者的转发分支。初学者往往只写“本地有session就发没有就丢弃”结果集群模式下消息丢了都查不出来。实现的时候请一定把“路由表查询”和“跨节点转发”补齐。11. 把方案落地到现有项目时需要做哪些裁剪如果是在现有单体项目里加推送功能不必一上来就上集群和消息中间件。可以先在单机模式下实现WebSocket连接管理、本地Session注册表、私发和广播的下行逻辑。这套代码是全分布式的第一步单机验证通过后再把消息投递改造成走MQ引入路由表节点可以水平扩展。这样分两步走风险小很多也方便定位问题。如果项目本身已经用了消息中间件强烈建议推送消息的Topic单独规划不要和其他业务共用。一是隔离流量二是监控方便三是保留消息时可以设置不同的保留时间。广播Topic的保留时间可以设置很短因为消息时效性高过期消息留着没有意义私发Topic可能要保留稍久以便离线补推时能查到最近消息。如果项目规模更大单靠WebSocket和MQ还不够可以引入专业的推送SDK或者长连接网关方案。自己做推送系统的核心收益是完全可控代价是运维和排障成本不低。我的建议是当前项目里已经有现成的WebSocket连接体系就自己扩展私发广播功能如果从零开始且连接规模预期会很大还是先评估成熟的方案再决定要不要自研。12. 私发和广播在生产环境运行的最后一公里功能上线只是开始真正影响线上体验的是心跳参数、断线重连、消息补推这三个机制。心跳参数上我建议客户端每30秒发送一次Ping服务端60秒内未收到任何消息就判定失活。太频繁浪费流量太慢会增加“僵尸连接”存在的时间。断线重连一定要带增量拉取能力客户端重连成功后拿着本地最后一条消息的msgId请求服务端补推缺失消息这样才能保证断线期间的消息不丢。消息补推是私发场景的兜底王牌。用户离线期间的消息服务端要存下来用户上线时主动推给他。默认存储最近24小时或者最近100条离线消息即可。实现方式可以是一张离线消息表用户上线时查询并推送给所有活跃连接。能接受这个成本私发的可靠性就能达到“只要用户最终上线消息就不会丢”的程度。上面这些机制配合上消息中间件本身的重试投递整条链路才算完整闭环。我在实际运行里还有一个心得私发和广播的监控指标要分开私发重点看路由命中率和P999延迟广播重点看单条广播的完成时间和节点消费延迟。两类指标混在一张监控大盘上很容易掩盖问题。分开之后每次发布或调优都能清晰地看到改动的效果。