DemoWatermarkDesk页面StreamAuditPage任务WM-3109。文中的采样流、运行状态、终端界面与 HiLog 全部属于确定性的演示资料不是传感器实测、真机编译结果或网络压力测试。一、先把四条数据故意按错误顺序送进来这篇我想换一种写法不从界面开始而从一份“看上去乱了”的数据开始。巡检手持终端可能先收到时间戳为1000毫秒的包再收到3000毫秒的包随后迟到的2000毫秒才出现。直接按到达顺序拼曲线会产生一次假的回退简单按时间排序又有一个麻烦究竟要等多久才敢把当前批次当成稳定数据提交我给演示项目起名WatermarkDesk任务 ID 为WM-3109页面叫StreamAuditPage。正常进入的四条样本是S-011000、S-033000、S-022000、S-066000。另外补两条异常S-LATE2500在水位线推进后才到达S-033000作为重复包再来一次。这里只讨论毫秒时间戳的整数比较不声称模拟了真实蓝牙、工业总线或网络缓冲区。目标结果先写明白允许迟到窗口1500毫秒最新已见事件时间为6000水位线最终为4500已经提交的是S-01,S-02,S-03待处理的是S-06。累计接受4条唯一有效样本提交3条暂存1条拒绝迟到1条过滤重复1条页面状态WATERMARK_OK。这个结果是后续代码、HiLog 和纯屏幕演示图的共同验收点。二、不要把“最新收到”误当成“最新发生”水位线不是系统自动产生的时间刻度而是业务侧为乱序事件约定的一条提交边界。我们记录见过的最大事件时间maxSeen然后减去一个允许迟到量latenessMs得到候选边界。考虑到后续事件时间可能倒退水位线只能向前不能因某条较旧的包而倒退。本例设置latenessMs1500所以最大时间为6000时边界是4500。这个定义有前提事件时间来自同一种可信时间基准。若某个设备时钟被人为调快最大事件时间就会虚高旧样本可能被提前判为迟到。生产系统应先明确时钟同步策略、包序号或可信服务端时间再决定是否采用类似水位线算法。本文仅展示本地有界重排的计算方法不把1500毫秒写成任何 HarmonyOS 系统默认值。先定义输入和输出的数据契约解决“算法需要哪些字段才能解释自己的判断”的问题。id用来拦截重复消息ts用来判断时间位置value是演示读数输出把已提交、暂存、迟到和重复分开列出避免一个总数让人误判已经落盘。interfaceSample{id:string;ts:number;value:number;}interfaceWatermarkSnapshot{taskId:string;maxSeen:number;watermark:number;accepted:number;committedIds:string[];pendingIds:string[];late:number;duplicated:number;status:string;}constDEMO_STREAM:Sample[][{id:S-01,ts:1000,value:26},{id:S-03,ts:3000,value:29},{id:S-02,ts:2000,value:27},{id:S-06,ts:6000,value:30},{id:S-LATE,ts:2500,value:28},{id:S-03,ts:3000,value:29}];这些接口类型仅约束演示业务对象不依赖未核验的传感器 SDK。此处没有资源分配所以没有释放动作真正接入设备时应把订阅、注销和错误回调留在独立的数据源模块。代码里的数值都能直接演算若输入ts是NaN、负数或超出业务可接受范围进入缓冲区前还应增加校验。本案例为便于核对使用已验证的固定整数样本。三、排序应该发生在待处理区不该重排历史事实拿到事件后我先判断重复 ID再判断它是不是落到了已关闭的水位线之后。顺序不能反过来一个已经提交的样本再次传入应该记作重复而不是把相同事件误记成新的迟到错误。这里维护一个Set存储见过的 ID一个数组暂存还可等待排序的事件。Array.sort只整理待处理区已经提交的数组不再重新解释。下面的WatermarkBuffer是纯 ArkTS 业务类没有任何虚构的系统WatermarkKit。ingest在接受一条新样本之后更新maxSeen与边界将已经落在边界内的暂存样本按时间顺序送入 committed。我们在本地演示里使用Map记录 ID 的时间戳便于调试与去重如果追求极低内存也可以把长期去重交给持久化存储但那是另一项设计取舍。classWatermarkBuffer{privatereadonlylatenessMs:number1500;privateseen:Mapstring,numbernewMapstring,number();privatepending:Sample[][];privatecommitted:Sample[][];privatemaxSeen:number0;privatewatermark:number-1;privatelateCount:number0;privateduplicateCount:number0;ingest(sample:Sample):void{if(this.seen.has(sample.id)){this.duplicateCount;return;}if(sample.tsthis.watermark){this.lateCount;return;}this.seen.set(sample.id,sample.ts);this.pending.push(sample);this.maxSeenMath.max(this.maxSeen,sample.ts);this.watermarkMath.max(this.watermark,this.maxSeen-this.latenessMs);this.pending.sort((a:Sample,b:Sample)a.ts-b.ts||a.id.localeCompare(b.id));constready:Sample[]this.pending.filter((p:Sample)p.tsthis.watermark);this.committed.push(...ready);this.pendingthis.pending.filter((p:Sample)p.tsthis.watermark);}snapshot():WatermarkSnapshot{return{taskId:WM-3109,maxSeen:this.maxSeen,watermark:this.watermark,accepted:this.seen.size,committedIds:this.committed.map((p:Sample)p.id),pendingIds:this.pending.map((p:Sample)p.id),late:this.lateCount,duplicated:this.duplicateCount,status:WATERMARK_OK};}}细看事件推进过程会发现S-033000进入时水位线是1500只有S-011000可以提交。S-022000虽然后收到但时间仍大于1500因此合法进入暂存区。直到S-066000到来水位线跳到4500S-02和S-03才按时间先后提交。S-LATE2500此时不可能再改写已经公开的区间所以被拒绝重复的S-03则由 Map 优先识别。这里的committed表示算法上已确定的输出不等于写入数据库成功。真实持久化还要记录事务结果、重试键与失败补偿不能在磁盘写失败时把它仍显示为业务成功。长期运行时seen会持续增长必须结合设备序列号、时间分区或固定保留周期清理去重记录清理太早则会失去跨窗口去重能力。这是示例为了透明而保留的边界。四、页面只读快照不拥有事件重排权我不让 UI 直接去操作pending数组。组件只需要知道水位线、处理结果和异常计数真正的状态迁移统一发生在ingest中。这与把传感器值一个个塞进多个State列表相比调试路径更短也更容易把文字、图和示例日志对起来。页面的重放按钮每次都创建新的缓冲实例避免第二次演示还带着第一次的seen集合。真正接入数据源时这个实例应由页面生命周期或专门的服务对象管理页面关闭要解除订阅避免重复消费。本文只使用六条内存数据因此没有定时器、文件或通信句柄需要关闭。EntryComponentstruct StreamAuditPage{StateresultText:string等待演示;StatecommitText:string—;StatependingText:string—;privatereplay():void{constbuffer:WatermarkBuffernewWatermarkBuffer();DEMO_STREAM.forEach((item:Sample)buffer.ingest(item));constreport:WatermarkSnapshotbuffer.snapshot();this.resultText${report.status}· 水位线${report.watermark};this.commitTextreport.committedIds.join(, );this.pendingTextreport.pendingIds.join(, );}aboutToAppear():void{this.replay();}build(){Column({space:16}){Text(WM-3109 · WatermarkDesk)Text(this.resultText)Text(已提交${this.commitText})Text(暂存${this.pendingText})Button(重放六条样本).onClick(()this.replay())}}}这段代码没有在页面上展示全部计数只保留关键重放路径。完整示意界面另外展示“接收4 / 提交3 / 暂存1 / 迟到1 / 重复1”其来源是同一个WatermarkSnapshot。如果用于正式工程可将report作为整个只读视图模型更新而不是像简化页面这样拆成多个文本字段这样避免某个字段先刷新、另一个字段仍保留上一轮数据。这里的生命周期入口采用aboutToAppear不是后台采样启动机制。五、看日志时要核对三个时间不是只盯一行成功很多乱序问题之所以难查是日志只写“处理成功”没有记录事件时间、最大已见时间与当前水位线。遇到一条迟到包研发无法判断是设备突然发旧数据还是自己提前推进了边界。本例明确把maxSeen6000、latenessMs1500与watermark4500同时展示并把拒绝原因拆为late1和duplicated1。二者看似都是丢弃其实意味着完全不同的设备行为。演示日志固定为WM-3109 maxSeen6000 watermark4500 accepted4 committed3 pending1 late1 duplicated1 statusWATERMARK_OK。若改动代码中的迟到窗口就必须同步修改图片和验算预期不能只更新一处数字。下图是纯屏幕的设计演示不包含真实传感器采样曲线也不能据此推断设备吞吐、端到端延迟或耗电。六、做成长期运行模块还需要哪些护栏事件时间不能无条件信任。生产环境通常至少检查时间戳范围、设备身份、数据值的物理区间以及数据包是否属于当前采集会话。服务异常恢复时还要确定水位线从哪里加载、去重键是否继续有效、缓存积压是否允许丢弃。否则应用每次冷启动都把水位线清零老包可能重新进入结果破坏原先的单调性约束。本例采用的整数时间戳可以用常规number比较但超大精度计数、微秒级时间以及跨进程传输应重新评估精度和序列化规则。若大量包同时到达逐条sort的复杂度也会让性能不稳定可以考虑批量处理或优先队列不要在没有测量之前宣称某个替代容器更快。我们只验证了六条固定样本的输出不把这种可预测的算例描述成压力实测。最后还有一点产品层面的取舍迟到包被拒绝并不意味着现场价值为零。某些审计场景应将它单独留在异常记录里供人工复核而不能悄悄丢掉。水位线负责的是当前有序结果的不可逆边界并不负责替业务定义所有数据的保存期限。这次 Demo 的结论是把重复判断放在前面让待处理区有序让提交区单调再把迟到消息作为显式异常暴露出来才有可能解释一条乱序数据究竟走到了哪里。参考资料华为官方 ArkTS 简介https://developer.huawei.com/consumer/cn/doc/harmonyos-guides/arkts-overview此处Map、Array.sort和水位线计算是 ArkTS 语言与应用层算法组合“事件水位线”并非本文声称的 HarmonyOS 平台内置 API。
