一、问题从一条重复的派单开始

OpsStream 是配送运维看板，页面 LiveOpsPage 订阅 dispatch-east 频道，把调度事件实时更新到列表。测试同事说它“切后台再回来，会偶尔多一条相同任务”。服务端查了发送记录，每个 seq 只有一份；客户端 HiLog 却在 11:42 连续打印两次 seq=18417。当时页面顶部仍写着“连接正常”，所以这个问题一度被当成列表 diff 错误。

沿着订阅链倒查才发现：页面 aboutToAppear() 每次都会调用 connect()，WebSocket 的旧 message 监听还在；RxJS 上层又新建了一次订阅。更糟的是，断网期间服务端产生了 6 条事件，重连后客户端只从“当前最新”继续，重复和丢失竟然同时存在。任务 WS-1142 最后不是靠列表去重补丁解决，而是重新定义了连接、回放和 UI 订阅各自的边界。

二、把“连上了”拆成四个可观测状态

HarmonyOS 的 WebSocket 流程很明确：createWebSocket() 创建对象，on('open'|'message'|'close'|'error') 订阅事件，connect() 建连，结束时 close() 并配合 off() 移除监听。问题在于，系统 API 只告诉我们传输层发生了什么，产品还需要知道回放是否完成。

因此 OpsStream 不再使用一个 connected: boolean，而是四态：DISCONNECTED → CONNECTING → SYNCING → LIVE。open 只代表物理连接建立，此时发送 resumeFrom=lastAck+1；服务端回放结束帧到达后才进入 LIVE。如果第一条实时事件的 seq 大于期望值，就退回 SYNCING 请求缺口，而不是直接渲染一个看似正常、实际断档的列表。

最终运行值固定下来：会话 ws_72，最后确认序号 18420，重连从 18421 请求，恢复缺口 6 条，丢弃重复 4 条，活跃订阅者 1 个，重连 2 次，端到端 P95 为 88 ms。这些值同时出现在状态仓、HiLog 和手机页，不再各算各的。

三、目录按“传输、序列、视图”切开

工程名 OpsStream，页面没有持有原生 WebSocket 对象。项目结构如下：

entry/src/main/ets/
├── pages/LiveOpsPage.ets
├── stream/HarmonySocketSource.ets
├── stream/DispatchStreamStore.ets
├── stream/SequenceGate.ets
└── model/DispatchEvent.ets

HarmonySocketSource 只负责把回调 API 转成 Observable；SequenceGate 维护 expectedSeq、重复计数和缺口；DispatchStreamStore 负责重连、回放状态与共享；页面只拿 StreamSnapshot。把三层分开之后，测试可以向 SequenceGate 注入乱序数组，不必真的断一次网。

四、第一段代码：原生回调只对应一个 Observable 生命周期

之前的适配器在 connect() 里不断 on('message')，却只在应用退出时关闭。新的源 Observable 在订阅时创建 WebSocket，在 teardown 中成对执行 off() 和 close()。下面省略证书与鉴权头，只保留 Demo 真正参与的生命周期。

import { webSocket } from '@kit.NetworkKit';
import { BusinessError } from '@kit.BasicServicesKit';
import { Observable } from 'rxjs';

export function socketFrames(url: string): Observable<string> {
  return new Observable<string>((subscriber) => {
    const ws = webSocket.createWebSocket();
    const onOpen = (): void => subscriber.next('{"type":"OPEN"}');
    const onMessage = (_err: BusinessError, value: string | ArrayBuffer): void => {
      if (typeof value === 'string') subscriber.next(value);
    };
    const onClose = (): void => subscriber.complete();
    const onError = (err: BusinessError): void => subscriber.error(err);

    ws.on('open', onOpen);
    ws.on('message', onMessage);
    ws.on('close', onClose);
    ws.on('error', onError);
    ws.connect(url);

    return () => {
      ws.off('open', onOpen);
      ws.off('message', onMessage);
      ws.off('close', onClose);
      ws.off('error', onError);
      ws.close();
    };
  });
}

这里故意保留四个稳定函数引用，因为 off() 必须拿到与 on() 对应的回调。teardown 不是只在页面销毁时执行：switchMap 切换连接、重试替换源、最后一个订阅者离开时都会触发。close() 可能在未完成建连时调用，所以工程版会把同步异常记录为 closeSkipped，但不会让 teardown 再向已结束的 subscriber 抛错。

另一个边界是二进制帧。Demo 协议只接受 JSON 文本，遇到 ArrayBuffer 直接计入协议错误，而不是用默认编码猜测。若产品要传 protobuf，应在这一层明确解码，并把解码失败与网络断开分开统计。

五、第二段代码：序号闸门决定重复、缺口和确认点

列表组件做 id 去重太晚了，因为重复事件可能已经触发通知、振动或统计。SequenceGate 在任何副作用之前检查 seq。它只接受三种结果：ACCEPT、DUPLICATE、GAP；只有连续事件才能推进 lastAck。

export type GateResult = 'ACCEPT' | 'DUPLICATE' | 'GAP';

export class SequenceGate {
  lastAck: number = 18420;
  duplicates: number = 0;
  gapSize: number = 0;

  inspect(seq: number): GateResult {
    if (seq <= this.lastAck) {
      this.duplicates++;
      return 'DUPLICATE';
    }
    const expected = this.lastAck + 1;
    if (seq > expected) {
      this.gapSize = seq - expected;
      return 'GAP';
    }
    this.lastAck = seq;
    this.gapSize = 0;
    return 'ACCEPT';
  }
}

为什么不接受“后到的大序号，再等小序号补齐”？因为 OpsStream 的协议保证单频道单调发送，出现 gap 就说明客户端确实缺数据。保持一个小型乱序缓存只会掩盖服务端或链路问题。lastAck 在事件完成业务落库后更新，而不是一收到帧就更新；否则渲染过程崩溃，重连也不会再要这条数据。

本轮注入测试先收到 18423，再回放 18421、18422、18423。第一帧触发缺口 2，回放推进确认点，第二次出现的 18423 被判为重复。最终 6 条缺口全部恢复，重复计数累计 4，没有对列表产生二次副作用。

六、第三段代码：共享连接，但不把缓存变成僵尸

RxJS 的 shareReplay 默认 refCount 为 false，这意味着订阅者归零后，源可能仍保持订阅。对 HTTP 缓存或许有用，对页面级 WebSocket 却会留下后台连接。Store 使用 shareReplay({ bufferSize: 1, refCount: true })，让状态栏和列表共享同一条源，同时保证最后一个消费者离开后触发原生 teardown。

import { BehaviorSubject, Observable, timer } from 'rxjs';
import { retry, shareReplay, switchMap } from 'rxjs/operators';

export class DispatchStreamStore {
  private readonly channel$ = new BehaviorSubject<string>('dispatch-east');
  readonly events$: Observable<DispatchEvent> = this.channel$.pipe(
    switchMap((channel) => socketFrames(this.url(channel)).pipe(
      retry({
        count: 3,
        delay: (_error, retryCount) => timer(retryCount * 800)
      }),
      this.decodeAndResume(channel)
    )),
    shareReplay({ bufferSize: 1, refCount: true })
  );

  setChannel(channel: string): void {
    if (channel !== this.channel$.value) this.channel$.next(channel);
  }

  dispose(): void {
    this.channel$.complete();
  }
}

switchMap 解决频道切换：新频道进入时，旧 Observable 先 teardown，旧 socket 的晚帧没有机会混进新页面。retry 放在共享之前，所有消费者共同经历一次重连；如果放在页面各自的订阅之后，状态栏和列表会各自建立一条连接。800、1600、2400 ms 的有限退避只处理可恢复网络错误，鉴权失败不会重试。

bufferSize: 1 缓存的是最近一条已校验事件，不是完整历史。页面重新订阅能立即得到当前态，但仍要通过 resumeFrom 补齐离线区间。离开页面时，两个 UI 订阅都取消，引用计数归零，原生回调与 socket 随之释放；aboutToDisappear() 另外调用 store.dispose()，防止页面实例留在导航缓存时继续接受频道切换。

调试阶段我增加了 subscribers 计数，并让每次源订阅打印 sessionId。修复前回前台一次会看到 ws_70、ws_71 同时接收；修复后只有 ws_72。关键日志为：task=WS-1142 session=ws_72 state=LIVE lastAck=18420 subscribers=1，下一行是 resumeFrom=18421 gapRecovered=6 duplicates=4 reconnects=2 p95=88ms。

七、恢复边界比重连次数更重要

把 socket 自动连回来只是第一层。真正决定数据是否可信的是回放窗口。服务端为 dispatch-east 保留最近 10 分钟或 5000 条事件；客户端若携带的 resumeFrom 早于窗口下界，会收到 RESET_REQUIRED。此时不能假装补齐成功，而要清空增量状态，调用一次快照接口，再以快照的 headSeq 建立实时流。

相反，短暂断网不应该全量刷新。lastAck=18420 时只请求 18421 之后的事件，收到 6 条回放并见到 REPLAY_END 后进入 LIVE。回放期间 UI 仍显示旧列表，但顶部明确标为“同步中”，避免用户把未完成的列表当成实时结果。

前后台也是一个边界。进入后台后 OpsStream 主动退订，而不是依赖系统最终清理；回前台重新订阅，共享源创建一个新 socket，并从持久化的确认点恢复。确认点每 20 条或 2 秒批量写入 Preferences，减少频繁 I/O。若进程在批量写入前退出，最多重复回放 20 条，序号闸门能安全丢弃，而不会丢数据。

八、最终运行状态能互相对账

在飞行模式开关、前后台切换、频道切换三组测试中，当前会话稳定为 ws_72，状态从 CONNECTING 经过 SYNCING 回到 LIVE。最后确认序号 18420；离线缺口恢复 6 条；由历史确认点和重试产生的 4 条重复全部在副作用前丢弃；活跃订阅者始终为 1。两次重连的消息端到端 P95 是 88 ms。

手机页 11:42 展示同一组数据，并把 resumeFrom 18421 放在“回放边界”卡片里。按钮只有“切换频道”和“重新同步”，没有虚构一个“修复网络”的营销动作。用户看到的是当前数据可信度，而不只是绿色连接图标。

九、这次留下的不是一个重试操作符

最终方案的核心并非 RxJS 写法，而是三个所有权：原生监听属于源 Observable 的订阅周期；序号确认点属于业务副作用完成之后；共享连接属于仍在场的消费者。三个边界一旦混在页面生命周期里，双订阅、僵尸连接和错误确认迟早会同时出现。

这个结构也有适用范围。若频道事件不保证单调，SequenceGate 必须改成带窗口的重排器；若每条事件都要求强事务确认，Preferences 批量持久化不够，需要关系型存储；若后台必须持续接收，则不能简单依赖页面 refCount，应把 Store 上移到明确的长生命周期能力中。工程封装不是把所有情况藏起来，而是把当前协议的假设写得足够清楚。

参考资料：华为开发者文档《WebSocket Connection》（页面更新于 2025-05-20）：https://developer.huawei.com/consumer/en/doc/harmonyos-guides-V14/websocket-connection-V14 ；RxJS shareReplay API：https://rxjs.dev/api/index/function/shareReplay