一、前置思考

分布式应用的真正考验在于"离线"——用户在地铁上没有网络，手机上的修改如何在到家后自动同步到平板？用户在山里拍照，如何在下山后自动将照片流转到PC？鸿蒙的离线分布式能力让应用在无网络环境下也能正常使用，恢复网络后自动同步。

本文聚焦：

• 离线操作队列（Operation Queue）的设计
• 本地优先原则（Local-First）与Sync-on-Reconnect
• 操作日志（WAL：Write-Ahead Log）的实现
• 上线后批量冲突解决策略

真实痛点场景：

1. 离线修改丢失：没网时改了数据，联网后发现全部回到旧状态
2. 同步风暴：上线后N个设备的离线修改同时涌来
3. 时间戳混乱：离线时设备时间不准，导致同步以错误的时间戳排序
4. 巨大操作日志：离线3天的操作日志膨胀到几百MB

二、核心原理

2.1 离线优先架构

┌────────────────────────────────────────┐
│            Local-First 架构              │
│                                         │
│  用户操作 ──→ 本地SQLite (WAL模式)      │
│               │                         │
│               ▼                         │
│          操作日志队列 (OpLog)            │
│            ┌──┬──┬──┬──┬──┐            │
│            │op│op│op│op│op│ 待同步      │
│            └──┴──┴──┴──┴──┘            │
│               │                         │
│     ┌─────────▼──── (网络恢复) ────┐    │
│     │  增量同步 PUSH               │    │
│     │  冲突解决 (CRDT/LWW)          │    │
│     │  合并到目标端                 │    │
│     └──────────────────────────────┘    │
└────────────────────────────────────────┘

2.2 操作队列设计

interface Operation {
  opId: string;            // 全局唯一操作ID
  opType: string;          // 'PUT' | 'DELETE' | 'UPDATE'
  kvKey: string;           // 操作的Key
  kvValue: string | null;  // 操作的Value(null=DELETE)
  timestamp: number;       // 操作时间戳 (使用Clock向量)
  deviceId: string;        // 操作设备ID
  status: OpStatus;        // PENDING / SYNCED / CONFLICT
}

enum OpStatus {
  PENDING = 0,   // 等待同步
  SYNCED = 1,    // 已同步
  CONFLICT = 2,  // 冲突待解决
  DROPPED = 3    // 已丢弃
}

class OfflineOperationQueue {
  private queue: Operation[] = [];
  private readonly MAX_QUEUE_SIZE: number = 5000;

  // 记录操作
  recordOperation(
    type: string, key: string, value: string | null
  ): void {
    const op: 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.length > this.MAX_QUEUE_SIZE) {
      this.queue.shift();
    }
  }

  // 压缩：同一key只保留最新操作
  private compressOps(key: string): void {
    let latestIdx: number = -1;
    for (let i: number = 0; i < this.queue.length; i++) {
      if (this.queue[i].kvKey === key) {
        if (latestIdx >= 0) {
          // 删除前一个旧操作
          this.queue.splice(latestIdx, 1);
          i--;
        }
        latestIdx = i;
      }
    }
  }

  // 获取待同步操作
  getPendingOps(): Operation[] {
    const pending: Operation[] = [];
    for (let i: number = 0; i < this.queue.length; i++) {
      if (this.queue[i].status === OpStatus.PENDING) {
        pending.push(this.queue[i]);
      }
    }
    return pending;
  }

  // 批量标记已同步
  markSynced(opIds: string[]): void {
    for (let i: number = 0; i < this.queue.length; i++) {
      const op: Operation = this.queue[i];
      for (let j: number = 0; j < opIds.length; j++) {
        if (op.opId === opIds[j]) {
          op.status = OpStatus.SYNCED;
          break;
        }
      }
    }
  }

  private generateOpId(): string {
    return this.deviceId + '-' + Date.now() + '-' +
      Math.random().toString(36).slice(2, 6);
  }

  private deviceId: string = 'd1';
}

2.3 Sync-on-Reconnect

class SyncOnReconnectManager {
  private opQueue: OfflineOperationQueue;
  private kvStore: distributedKVStore.SingleKVStore | null = null;

  constructor(opQueue: OfflineOperationQueue) {
    this.opQueue = opQueue;
  }

  // 网络恢复时的同步
  async onNetworkRestored(): Promise<SyncResult> {
    const pendingOps: Operation[] = this.opQueue.getPendingOps();
    if (pendingOps.length === 0) {
      return { syncedCount: 0, conflictedCount: 0 };
    }

    console.info('[Sync] 开始同步' + String(pendingOps.length) + '个离线操作');

    let syncedCount: number = 0;
    let conflictedCount: number = 0;

    if (this.kvStore !== null) {
      await this.kvStore.startTransaction();

      for (let i: number = 0; i < pendingOps.length; i++) {
        const op: Operation = pendingOps[i];
        try {
          await this.applyOperation(op);
          syncedCount++;
        } catch (e) {
          op.status = OpStatus.CONFLICT;
          conflictedCount++;
        }
      }

      await this.kvStore.commitTransaction();
    }

    // 标记已同步
    const syncedIds: string[] = [];
    for (let i: number = 0; i < pendingOps.length; i++) {
      if (pendingOps[i].status === OpStatus.PENDING) {
        syncedIds.push(pendingOps[i].opId);
      }
    }
    this.opQueue.markSynced(syncedIds);

    return { syncedCount: syncedCount, conflictedCount: conflictedCount };
  }

  private async applyOperation(op: Operation): Promise<void> {
    if (this.kvStore === null) return;

    switch (op.opType) {
      case 'PUT':
        await this.kvStore.put(op.kvKey, op.kvValue ?? '');
        break;
      case 'DELETE':
        await this.kvStore.delete(op.kvKey);
        break;
      case 'UPDATE':
        await this.kvStore.put(op.kvKey, op.kvValue ?? '');
        break;
      default:
        break;
    }
  }
}

interface SyncResult {
  syncedCount: number;
  conflictedCount: number;
}

三、避坑速查

四、总结

离线分布式能力的核心：

1. 本地WAL操作日志 → 先写本地，再同步
2. 操作压缩 → 同key只保留最新
3. Sync-on-Reconnect → 网络恢复自动同步
4. 分批同步 → 避免同步风暴