登录
推荐 文章 Go 技术 课程 下载 专题 AI
首页 >  文章 >  java教程

Java HttpClient WebSocket 怎样处理重连后的消息顺序

来源:17golang原创

时间:2026-11-14 05:44:24 146浏览 收藏

Java HttpClient 的 WebSocket 重连后,要靠应用层的消息序列与恢复协议保证业务顺序:服务端为同一业务流提供稳定序号,客户端记录最后连续提交的游标,重连时从该游标继续回放;本地再用连接代次隔离旧回调。重新建立 WebSocket 只恢复通信通道,不会自动补发断线期间的消息。

官方地址:https://docs.oracle.com/en/java/javase/25/

使用场景:连接重新通了,业务状态却跳了一步

以库存变更订阅为例,客户端已经处理到 120,断线后先收到 122,随后又收到重放的 120。直接按到达顺序修改库存,可能漏掉 121,也可能重复扣减。另一个隐蔽来源是旧连接已经投递到业务线程池的任务:新连接开始工作后,旧任务才完成。

Java 的 Listener 对单个 WebSocket 的回调有顺序约束,但不同 WebSocket 的回调可能并发。回调把任务提交到并行线程池后,业务提交完成的顺序也需要自己管理。因此必须分别回答三个问题:这一条消息属于哪次连接、属于哪个业务流、是否接在最后成功提交的事件后面。

候选方案:直接继续、游标回放还是快照恢复

对于只展示最新温度或在线人数的页面,可以在重连后取最新值,不追求每条中间变化。对于库存增减、任务状态演进等增量事件,应优先选择带序号的回放协议。服务端若只保留有限历史,还需要快照加水位的恢复方式。

方案适用数据重连后的主要要求
直接接收最新状态允许覆盖的完整状态用版本过滤旧状态
游标回放不能随意丢弃的增量事件稳定序号、历史保留、重复处理规则
快照加水位历史已过期或重放成本很高快照与后续事件具有一致的切换水位

这些是应用协议的设计选择,不是 HttpClient 的开关。若服务端既没有历史查询,也没有可靠快照,客户端不能仅靠本地排序恢复已经丢失的内容。无限等待一个从未收到的序号,只会让故障变成缓冲区不断增长。

对比维度:四个字段不要混用

stream 标识业务序列的归属,例如一个仓库的一条变更流;sequence 表示这个流里的事件顺序;lastApplied 表示已经连续、成功提交的位置;generation 则是客户端本地的连接代次。连接断开并不意味着业务序列从头开始,连接代次也不能替代业务事件编号。

WebSocket连接代次与业务序列的职责划分
连接回调与业务序列的静态关系说明图。连接域负责识别新旧连接,业务域负责判断事件是否连续,两个条件共同限制提交。

本文选择一个明确契约:单一 stream,正整数序号从 1 开始且连续递增,0 表示尚未处理任何事件。多个 stream 应各自维护游标。若服务端按全局序号编号、订阅过滤后允许跳号,就不能照搬“必须等于上一条加一”的规则,应改用服务端提供的前驱游标、恢复令牌或其他连续性约定。

同样,服务端重新创建流并把序号归零时,需要新的流标识或服务端 epoch。单独保存一个数字不足以区别“新流的 10”和“旧流已经处理过的 10”。游标恢复必须携带完整订阅身份,而不只是最大序号。

推荐选择:一个串行闸门同时管代次、重复与缺口

下面是应用层核心类,采用 Java 17 及以上可用的语法,不包含具体服务端协议、网络连接器或数据库实现。它刻意把提交操作留成接口:实现方必须保证业务修改与持久化游标在同一事务里成功,否则跨进程恢复仍会出现重复或遗漏。

final class SequenceGate {
    interface Commit {
        // 实现必须把业务修改和持久化游标放在同一事务中。
        void applyAndSave(long sequence, String payload) throws Exception;
    }
    record Ticket(long generation, long after) {}
    enum Result { APPLIED, DUPLICATE, STALE, GAP, PAUSED }

    private long generation;
    private long lastApplied;
    private boolean paused = true;
    private final Commit commit;

    SequenceGate(long persistedCursor, Commit commit) {
        if (persistedCursor 

每次尝试建立连接前调用 beginConnection(),为该连接创建独立 Listener,固定持有返回的 Ticket。通过 HttpClient.newWebSocketBuilder().buildAsync(uri, listener) 建立连接后,用约定的应用消息请求 after=Ticket.after() 的事件;“after”表示严格晚于游标。它是本文协议约定,不是 WebSocket 标准参数。

握手失败或连接出错时,由连接管理器对该 Ticket 调用 retire;退避重试也归同一个管理器管理,避免 onError、onClose 和发送失败各自启动一条重连链。过期的连接建立完成回调应关闭自己的连接,不能把它登记成当前连接。

闸门把 seq 当作重放重复,只允许恰好下一条提交;发现缺口就暂停本代并通知管理器恢复。它不会缓存后续事件,因此不会偷偷跳过缺失项。示例在锁内调用提交接口,是为了展示最简单的串行提交边界,适合短小同步事务;高吞吐场景可以改为每条业务流一个有界串行执行器,但代次切换和实际提交仍要在同一顺序域内完成。

序列闸门、恢复请求和业务事务的关系
序列闸门与持久化提交的静态结构说明图。事务边界包含业务修改与持久化游标,本地暂停状态只阻止继续提交,不代替服务端回放。

假设已提交到 120:旧代事件直接忽略;当前代重放 120 判为重复;当前代先发 122 则进入缺口暂停,不能直接把游标改成 122。重新建立恢复会话后,从 120 回放 121、122 才能连续提交。这是按规则推导的例子,没有声称运行结果。

不适用情况:分片、异步副作用和关闭也有边界

Listener 的一次 onText 不一定是一整条业务消息。每个连接使用独立累积器,把分片复制并组装到 last=true 后再解码序号,同时设置总长度上限,断开时丢弃未完成片段。不能把旧连接的一半文本留给新连接继续拼接。

request(1) 增加的是回调次数额度,不是业务消息确认,也不是一次完整消息的保证。返回的 CompletionStage 与需求额度是不同机制;如果回调立即派发并行异步任务然后返回,不能由此推出任务已按顺序完成。需要在任务完成后继续发放额度时,应把它与有界队列、片段组装和失败处理共同设计。

本例的同步锁只覆盖当前进程。多进程同时消费同一条流时,还需要在存储层限制游标竞争;业务副作用若发生在外部服务,也不能靠一个数据库事务自动实现跨系统“恰好一次”。可按业务选择幂等键或可靠投递机制。提交结果不确定时,应读取持久化状态核对,不能简单把内存游标向前推。

主动停止时,先将管理器置为停止状态,取消已安排的重连,并 retire 当前代,再调用 sendClose(WebSocket.NORMAL_CLOSURE, "") 发起正常关闭。sendClose 的完成不等于输入已经关闭;需要等待对端关闭并设置应用自己的等待期限,到期可用 abort() 终止。停止后的迟到 onError/onClose 不应再次安排重连。

决策表:按业务语义选择恢复方式

条件选择不能承诺的事
完整状态可覆盖重连拉最新状态并比较版本保留全部中间事件
增量事件可回放连续提交游标、代次隔离、去重仅靠连接成功实现恢复
游标早于历史保留范围一致快照加切换水位从不存在的历史补齐事件
多个业务分区独立有序每分区维护序列与提交域由一个最大数字推导全局顺序

常见问题

换成单线程业务池就足够了吗

它只能帮助串行执行,不能补齐断线消息,也不能区分旧连接和重放重复。仍需要稳定序列、恢复协议以及在提交点执行的代次检查。

为什么不直接按时间戳排序

时间戳可能重复,也不一定表达业务提交顺序。除非服务端定义了可靠的排序与缺口语义,否则它只能作为展示字段,不能替代恢复游标。

重连后先接实时消息,再异步补历史可以吗

必须先有服务端定义的水位与缓冲合并规则。本文的简化方案要求恢复会话按序返回历史并衔接实时流;不具备这个契约时,应冻结提交并通过独立同步流程恢复。

参考资料

声明:本文转载于:17golang原创 如有侵犯,请联系study_golang@163.com删除
相关阅读
更多>
最新阅读
更多>
课程推荐
更多>