一、前置思考分布式应用的真正考验在于离线——用户在地铁上没有网络手机上的修改如何在到家后自动同步到平板用户在山里拍照如何在下山后自动将照片流转到PC鸿蒙的离线分布式能力让应用在无网络环境下也能正常使用恢复网络后自动同步。本文聚焦离线操作队列Operation Queue的设计本地优先原则Local-First与Sync-on-Reconnect操作日志WALWrite-Ahead Log的实现上线后批量冲突解决策略真实痛点场景离线修改丢失没网时改了数据联网后发现全部回到旧状态同步风暴上线后N个设备的离线修改同时涌来时间戳混乱离线时设备时间不准导致同步以错误的时间戳排序巨大操作日志离线3天的操作日志膨胀到几百MB二、核心原理2.1 离线优先架构┌────────────────────────────────────────┐ │ Local-First 架构 │ │ │ │ 用户操作 ──→ 本地SQLite (WAL模式) │ │ │ │ │ ▼ │ │ 操作日志队列 (OpLog) │ │ ┌──┬──┬──┬──┬──┐ │ │ │op│op│op│op│op│ 待同步 │ │ └──┴──┴──┴──┴──┘ │ │ │ │ │ ┌─────────▼──── (网络恢复) ────┐ │ │ │ 增量同步 PUSH │ │ │ │ 冲突解决 (CRDT/LWW) │ │ │ │ 合并到目标端 │ │ │ └──────────────────────────────┘ │ └────────────────────────────────────────┘2.2 操作队列设计interfaceOperation{opId:string;// 全局唯一操作IDopType:string;// PUT | DELETE | UPDATEkvKey:string;// 操作的KeykvValue:string|null;// 操作的Value(nullDELETE)timestamp:number;// 操作时间戳 (使用Clock向量)deviceId:string;// 操作设备IDstatus:OpStatus;// PENDING / SYNCED / CONFLICT}enumOpStatus{PENDING0,// 等待同步SYNCED1,// 已同步CONFLICT2,// 冲突待解决DROPPED3// 已丢弃}classOfflineOperationQueue{privatequeue:Operation[][];privatereadonlyMAX_QUEUE_SIZE:number5000;// 记录操作recordOperation(type:string,key:string,value:string|null):void{constop:Operation{opId:this.generateOpId(),opType:type,kvKey:key,kvValue:value,timestamp:Date.now(),deviceId:this.deviceId,status:OpStatus.PENDING};this.queue.push(op);// 操作日志压缩相同key只保留最新操作this.compressOps(key);// 限制队列大小if(this.queue.lengththis.MAX_QUEUE_SIZE){this.queue.shift();}}// 压缩同一key只保留最新操作privatecompressOps(key:string):void{letlatestIdx:number-1;for(leti:number0;ithis.queue.length;i){if(this.queue[i].kvKeykey){if(latestIdx0){// 删除前一个旧操作this.queue.splice(latestIdx,1);i--;}latestIdxi;}}}// 获取待同步操作getPendingOps():Operation[]{constpending:Operation[][];for(leti:number0;ithis.queue.length;i){if(this.queue[i].statusOpStatus.PENDING){pending.push(this.queue[i]);}}returnpending;}// 批量标记已同步markSynced(opIds:string[]):void{for(leti:number0;ithis.queue.length;i){constop:Operationthis.queue[i];for(letj:number0;jopIds.length;j){if(op.opIdopIds[j]){op.statusOpStatus.SYNCED;break;}}}}privategenerateOpId():string{returnthis.deviceId-Date.now()-Math.random().toString(36).slice(2,6);}privatedeviceId:stringd1;}2.3 Sync-on-ReconnectclassSyncOnReconnectManager{privateopQueue:OfflineOperationQueue;privatekvStore:distributedKVStore.SingleKVStore|nullnull;constructor(opQueue:OfflineOperationQueue){this.opQueueopQueue;}// 网络恢复时的同步asynconNetworkRestored():PromiseSyncResult{constpendingOps:Operation[]this.opQueue.getPendingOps();if(pendingOps.length0){return{syncedCount:0,conflictedCount:0};}console.info([Sync] 开始同步String(pendingOps.length)个离线操作);letsyncedCount:number0;letconflictedCount:number0;if(this.kvStore!null){awaitthis.kvStore.startTransaction();for(leti:number0;ipendingOps.length;i){constop:OperationpendingOps[i];try{awaitthis.applyOperation(op);syncedCount;}catch(e){op.statusOpStatus.CONFLICT;conflictedCount;}}awaitthis.kvStore.commitTransaction();}// 标记已同步constsyncedIds:string[][];for(leti:number0;ipendingOps.length;i){if(pendingOps[i].statusOpStatus.PENDING){syncedIds.push(pendingOps[i].opId);}}this.opQueue.markSynced(syncedIds);return{syncedCount:syncedCount,conflictedCount:conflictedCount};}privateasyncapplyOperation(op:Operation):Promisevoid{if(this.kvStorenull)return;switch(op.opType){casePUT:awaitthis.kvStore.put(op.kvKey,op.kvValue??);break;caseDELETE:awaitthis.kvStore.delete(op.kvKey);break;caseUPDATE:awaitthis.kvStore.put(op.kvKey,op.kvValue??);break;default:break;}}}interfaceSyncResult{syncedCount:number;conflictedCount:number;}三、避坑速查坑现象原因解决离线日志膨胀操作队列占用几百MB未压缩或限制大小同key只保留最新操作MAX_QUEUE_SIZE限制同步时序错乱旧数据覆盖新数据离线时间戳不准使用逻辑时钟(Lamport Clock)替代物理时间重复同步离线操作重复执行幂等性不足每条op分配全局唯一opId目标端去重大量冲突上线后冲突解决卡顿一次性处理所有离线op分批同步每批100条每批间隔200ms四、总结离线分布式能力的核心本地WAL操作日志 → 先写本地再同步操作压缩 → 同key只保留最新Sync-on-Reconnect → 网络恢复自动同步分批同步 → 避免同步风暴
