一、前置思考

分布式数据管理是鸿蒙提供的最强大的分布式能力之一——数据在同一账号下的多设备间自动同步，开发者只需操作本地KVStore，框架自动完成跨设备的数据复制、冲突解决和一致性保障。但"自动"不等于"不需要理解"，不理解同步机制的开发者往往会写出有严重缺陷的分布式应用。

本文聚焦：

• distributedKVStore的核心架构与同步原理
• CRDT算法在鸿蒙中的实际应用
• 事务机制与最终一致性保障
• 大容量数据场景下的性能优化

真实痛点场景：

1. 数据不同步：手机上改了设置，平板过了30秒还没更新
2. 数据冲突：两端同时修改同一条数据，最后只有一边的数据保留了
3. 同步风暴：大量数据频繁变更，网络带宽被占满
4. 数据泄露：卸载重装后发现敏感数据留在了其他设备上

二、核心原理

2.1 distributedKVStore架构

┌─────────────────────────────────────────┐
│              应用层                       │
│   put(key, value)  /  get(key)          │
├─────────────────────────────────────────┤
│       distributedKVStore (JS API)       │
│  ┌───────────────────────────────────┐  │
│  │     KVStore客户端                  │  │
│  │  ├── 本地SQLite缓存               │  │
│  │  ├── CRDT冲突解决引擎             │  │
│  │  ├── 变更日志 (ChangeLog)         │  │
│  │  └── 同步策略控制                 │  │
│  └───────────────────────────────────┘  │
├─────────────────────────────────────────┤
│           同步服务 (Sync Service)        │
│  ├── 增量同步 (Delta Sync)              │
│  ├── 全量同步 (Full Sync)               │
│  ├── 优先级队列                         │
│  └── 版本向量管理                       │
├─────────────────────────────────────────┤
│           软总线 (DSoftBus)              │
└─────────────────────────────────────────┘

2.2 KVStore三种类型对比

类型	特点	适用场景	同步方式
SINGLE_VERSION	单版本，最新覆盖	设置、开关、配置	全量同步
MULTI_VERSION_CRDT	CRDT多版本，自动合并	协作编辑、购物车	增量合并
DEVICE_COLLABORATION	设备间协作，支持离线	表单协同、笔记同步	增量+合并

2.3 CRDT冲突解决原理

CRDT（Conflict-free Replicated Data Type）保证在无中心服务器的情况下，多端并发写入最终达到一致状态：

// CRDT Register（寄存器）：最后一次写入优先（LWW）
// 场景：用户头像URL这种"最后修改为准"的数据
interface CRDTRegister<T> {
  value: T;
  timestamp: number;  // 全局唯一时间戳
  nodeId: string;     // 修改者ID
}

// LWW合并规则
function mergeRegister<T>(a: CRDTRegister<T>, b: CRDTRegister<T>): CRDTRegister<T> {
  // 时间戳大的优先；时间戳相等时，nodeId大的优先（打破平局）
  if (a.timestamp > b.timestamp) return a;
  if (a.timestamp < b.timestamp) return b;
  return a.nodeId > b.nodeId ? a : b;
}

// CRDT Counter（计数器）：增量型，保证最终数值一致
// 场景：多端同时统计点赞数
interface CRDTCounter {
  increments: Map<string, number>;  // nodeId -> delta
}

function incrementCounter(counter: CRDTCounter, nodeId: string, delta: number): void {
  const current: number = counter.increments.get(nodeId) ?? 0;
  counter.increments.set(nodeId, current + delta);
}

function getCounterValue(counter: CRDTCounter): number {
  let total: number = 0;
  const values: number[] = [];
  const entries: MapIterator<[string, number]> = counter.increments.entries();
  // sum all deltas from all nodes
  for (let entry = entries.next(); !entry.done; entry = entries.next()) {
    values.push(entry.value[1]);
  }
  for (let i: number = 0; i < values.length; i++) {
    total += values[i];
  }
  return total;
}

2.4 同步策略选择

import { distributedKVStore } from '@kit.DistributedKVStore';

// 初始化KVStore
async function initKVStore(): Promise<distributedKVStore.SingleKVStore> {
  const kvManager: distributedKVStore.KVManager =
    distributedKVStore.createKVManager({
      bundleName: 'com.example.app'
    });

  const options: distributedKVStore.Options = {
    createIfMissing: true,
    encrypt: true,           // 加密存储
    backup: true,            // 自动备份
    kvStoreType: distributedKVStore.KVStoreType.MULTI_VERSION,
    securityLevel: distributedKVStore.SecurityLevel.S2  // 安全级别
  };

  const store: distributedKVStore.SingleKVStore =
    await kvManager.getKVStore('userSettings', options);

  // 设置同步策略
  store.setSyncRange(['user:*']);              // 只同步user:前缀的key
  store.setSyncMode(distributedKVStore.SyncMode.PULL_ONLY);  // 只拉不推
  // 可选: PUSH_ONLY / PUSH_PULL（双向） / PULL_ONLY

  return store;
}

// 增量同步（推荐）
async function syncIncremental(store: distributedKVStore.SingleKVStore): Promise<void> {
  await store.sync(
    [], // 空数组表示所有设备
    distributedKVStore.SyncMode.PUSH_PULL  // 双向同步
  );
}

// 带条件同步
async function syncConditional(store: distributedKVStore.SingleKVStore): Promise<void> {
  // 仅同步最近1小时内修改的数据
  const oneHourAgo: number = Date.now() - 3600000;
  const predicate: distributedKVStore.SyncPredicate =
    new distributedKVStore.SyncPredicate();
  predicate.setTimestampRange(oneHourAgo, Date.now());

  await store.sync([], distributedKVStore.SyncMode.PUSH_PULL, predicate);
}

三、事务与一致性

3.1 事务操作

async function transferBalance(
  store: distributedKVStore.SingleKVStore,
  fromAccount: string,
  toAccount: string,
  amount: number
): Promise<boolean> {
  try {
    // 开启事务
    await store.startTransaction();

    // 操作1: 扣款
    const fromBalance: number = Number(await store.get(fromAccount) ?? 0);
    if (fromBalance < amount) {
      await store.rollbackTransaction();
      return false;
    }
    await store.put(fromAccount, fromBalance - amount);

    // 操作2: 收款
    const toBalance: number = Number(await store.get(toAccount) ?? 0);
    await store.put(toAccount, toBalance + amount);

    // 提交事务
    await store.commitTransaction();
    return true;

  } catch (e) {
    // 异常回滚
    await store.rollbackTransaction();
    return false;
  }
}

3.2 变更监听

// 监听本地KVStore变化
// 注意：on后不要加括号！ArkTS中使用属性的方式来设置回调
store.on('dataChange', distributedKVStore.SubscribeType.SUBSCRIBE_TYPE_ALL,
  (data: distributedKVStore.ChangeNotification) => {
    const insertEntries: distributedKVStore.Entry[] =
      data.insertEntries as distributedKVStore.Entry[];
    const updateEntries: distributedKVStore.Entry[] =
      data.updateEntries as distributedKVStore.Entry[];

    for (let i: number = 0; i < insertEntries.length; i++) {
      const entry: distributedKVStore.Entry = insertEntries[i];
      console.info('[KVStore] 新增: ' + entry.key + ' = ' + entry.value);
    }
    for (let i: number = 0; i < updateEntries.length; i++) {
      const entry: distributedKVStore.Entry = updateEntries[i];
      console.info('[KVStore] 更新: ' + entry.key + ' = ' + entry.value);
    }
  }
);

// 监听远端同步完成
store.on('syncComplete', (data: distributedKVStore.SyncNotification) => {
  console.info('[KVStore] 同步完成, 状态: ' + data.status);
});

3.3 性能优化策略

// 批量写入优化
async function batchWrite(
  store: distributedKVStore.SingleKVStore,
  entries: Array<{ key: string; value: string }>
): Promise<void> {
  await store.startTransaction();
  for (let i: number = 0; i < entries.length; i++) {
    const item: { key: string; value: string } = entries[i];
    await store.put(item.key, item.value);
  }
  await store.commitTransaction();
  // 一次事务只产生一次同步通知，而非N次
}

// 大value存入单独文件
async function storeLargeValue(
  store: distributedKVStore.SingleKVStore,
  key: string,
  largeData: ArrayBuffer
): Promise<void> {
  // KVStore中只存文件路径的索引
  await store.put(key + '_filepath', '/data/storage/el2/base/files/' + key + '.dat');
  await store.put(key + '_size', String(largeData.byteLength));
  await store.put(key + '_md5', calculateMD5(largeData));
  // 实际文件通过Session/P2P传输
}

四、完整代码架构

Demo中的分布式数据管理架构：

Layer 1: 数据层
  ├── 本地KVStore模拟
  ├── 数据变更日志
  └── CRDT计数器实现

Layer 2: 同步层
  ├── 增量同步 (Delta Sync)
  ├── 全量同步 (Full Sync)
  ├── 同步策略 (PUSH/PULL/PUSH_PULL)
  └── 冲突解决方案（LWW + CRDT）

Layer 3: 业务层
  ├── 用户设置同步
  ├── 购物车多端合并
  └── 点赞计数CRDT去重

五、避坑速查

坑	现象	原因	解决
同步延迟大	修改后10秒+才同步到对端	默认同步间隔是手动触发的	关键数据修改后立即调用store.sync()
加密key丢失	升级后KVStore不可读	加密key随设备重启变化	使用固定的password参数而非自动生成
多端计数不正确	点赞数时多时少	用了SINGLE_VERSION导致覆盖	计数场景用MULTI_VERSION_CRDT
syncRange过宽	同步速度慢	全量同步所有key	按业务模块设置syncRange前缀
未关闭store造成泄漏	内存持续增长	getKVStore后未close	aboutToDisappear中关闭store
同步风暴	网络流量激增	批量写每条都触发同步	用事务包裹批量写入，出事务后一次同步
类型转换错误	get()的值类型和预期不符	KVStore存的是string	明确存取的序列化格式，读后用Number()转换
安全等级不足	敏感数据被其他应用访问	SecurityLevel设置过低	敏感数据使用S3或S4级别
离线数据丢失	离线时修改数据未同步	未配置backup选项	createIfMissing+backup开启，离线修改在本地SQLite暂存
删除后仍同步	delete后其他设备又出现	删除操作未同步	使用逻辑删除(softDelete flag)而非物理删除

六、总结

分布式数据管理的设计哲学是本地优先 + 异步同步 + 最终一致：

1. 本地优先：所有读写操作都在本地完成（毫秒级），不受网络影响
2. 异步同步：后台自动将变更增量推送到其他设备，不阻塞UI
3. 最终一致：通过CRDT算法保证多端并发写入最终一致
4. 选择正确的KVStore类型：单版本/MultiVersion/DeviceCollaboration，选错会引入严重bug

记住：KVStore存的都是字符串，读出来要显式转换类型；Batch写入用事务包裹。