OpenHarmony 统一互联开发指导——分布式数据对象跨设备同步

适用版本：OpenHarmony 4.1 Release 及以上（API 11+），接口以 API 12 为准
示例环境：两台 RK3568 开发板、OpenHarmony 5.0 Release、DevEco Studio 5.0 Release
说明：文中的延迟、对象大小等数值为设计建议值，实际以设备与网络实测为准。

1 概述

1.1 场景

统一互联（All Devices, One Connect）解决的是跨设备发现、连接、组网与协同。设备「看得到」之后，下一步通常是「状态要一致」。分布式数据对象（Distributed Data Object）就是应用侧最轻量的一种状态同步方案：

• 手机控制大屏播放器：进度、音量、播放状态双端一致；
• 平板与车机协同导航：目的地、路线偏好实时同步；
• 多端协同草稿：标题、光标位置、编辑状态同步；
• 智能家居面板：多个控制端显示同一设备状态；
• 教学场景：教师端翻页，学生端同步翻页。

这些场景的共同点：数据量小、变更频繁、要求低延迟、允许最终一致。如果数据是照片、视频、安装包，应该使用文件传输方案，而不是分布式数据对象。

1.2 能力边界

先把边界说清楚，避免选型错误：

维度	分布式数据对象	说明
数据形态	基本类型、数组、基本类型嵌套	不支持函数、类方法、undefined
数据规模	KB 级	不建议承载大对象或二进制数据
同步语义	内存级、最终一致	不是事务，不保证强一致
设备发现	不负责	需先通过设备管理完成发现与互信
鉴权	不提供	sessionId 只是会话标识，不是安全凭证
持久化	有限支持	save() 保存的数据有生命周期限制
适用网络	可信组网内	依赖软总线与分布式数据服务

一句话总结：它适合「同步一小块状态」，不适合「传文件」和「当数据库」。

2 技术架构

┌──────────────────────────────────────────────┐
│ 业务层：播放器 / 草稿 / 控制面板              │
├──────────────────────────────────────────────┤
│ 同步层：DataObject 封装、节流、冲突合并       │
├──────────────────────────────────────────────┤
│ 分布式数据对象：create / setSessionId / on    │
├──────────────────────────────────────────────┤
│ 分布式数据服务：内存数据库 + 变更分发         │
├──────────────────────────────────────────────┤
│ 统一互联底座：设备管理 + 软总线 + 可信组网     │
└──────────────────────────────────────────────┘

工作方式：

1. 每个设备用同一份 source 结构创建 DataObject；
2. 各设备设置相同 sessionId，加入同一个协同组；
3. 任一设备修改对象属性，框架把变更同步给组内其他设备；
4. 其他设备触发 on('change')，回调中给出变更的字段名；
5. 设备上下线通过 on('status') 通知，业务据此提示「对端已离线」。

注意：框架同步的是对象属性，不会同步类方法、闭包和 UI 状态。业务层需要把「要同步的状态」抽成纯数据对象。

3 前提条件

3.1 权限

module.json5 声明：

{
  "module": {
    "name": "entry",
    "requestPermissions": [
      {
        "name": "ohos.permission.DISTRIBUTED_DATASYNC",
        "reason": "$string:reason_distributed_datasync",
        "usedScene": {
          "abilities": ["EntryAbility"],
          "when": "inuse"
        }
      }
    ]
  }
}

运行时申请：

import { abilityAccessCtrl, common, Permissions } from '@kit.AbilityKit';
import { BusinessError } from '@kit.BasicServicesKit';
import { hilog } from '@kit.PerformanceAnalysisKit';

const PERMISSION: Permissions = 'ohos.permission.DISTRIBUTED_DATASYNC';

export async function ensureSyncPermission(context: common.UIAbilityContext): Promise<boolean> {
  const atManager = abilityAccessCtrl.createAtManager();
  try {
    const result = await atManager.requestPermissionsFromUser(context, [PERMISSION]);
    return result.authResults.every((code: number) => code === 0);
  } catch (err) {
    const e = err as BusinessError;
    hilog.error(0x0000, 'DDO', 'request permission failed: %{public}d %{public}s', e.code, e.message);
    return false;
  }
}

setSessionId 需要该权限，未授权时会返回 201。权限被拒时应展示引导，而不是静默失败。

3.2 设备与账号

条件	说明
可信组网	两台设备已完成互信（同账号或系统设置确认）
同一应用	bundleName 与签名证书一致
网络可达	同一局域网，或系统支持的其他组网方式
版本一致	大版本尽量一致，避免能力差异

分布式数据对象不负责发现设备。设备发现、可信判断请参考《OpenHarmony 统一互联设备发现不到对端设备问题分析报告》，确认对端可信后再建立会话。

4 核心接口

4.1 接口速查

接口	作用	关键点
distributedDataObject.create(context, source)	创建数据对象	source 为纯数据对象，属性即同步字段
distributedDataObject.genSessionId()	生成随机 sessionId	只保证随机，不保证全局唯一语义
dataObject.setSessionId(sessionId)	加入/退出协同组	传 ''、null 或不传参表示退出
dataObject.on('change', cb)	监听属性变更	回调给出 sessionId 与变更字段名
dataObject.on('status', cb)	监听设备上下线	状态为 online / offline
dataObject.off('change' | 'status')	移除监听	页面销毁时必须调用
dataObject.save(deviceId)	持久化对象	deviceId 可为 'local'
dataObject.revokeSave()	撤回持久化	删除已保存的数据

4.2 数据模型约束

export class PlaybackState {
  title: string = '';
  position: number = 0;
  playing: boolean = false;
  volume: number = 50;
  updatedAt: number = 0;
  writer: string = '';
  version: number = 0;
}

要求与建议：

1. 只用纯数据：string、number、boolean、数组、基本类型嵌套；不要放函数、Date、类实例方法；
2. 字段扁平化：嵌套层级越深，变更通知与排查越困难，建议控制在 2 层以内；
3. 时间用时间戳：不要存 Date 对象，统一存毫秒数；
4. 字段名规范：非空、长度不超过 128 字节，建议小驼峰；
5. 控制对象大小：建议 KB 级，避免频繁全量同步造成卡顿；
6. 不存敏感数据：该能力面向状态同步，不应承载密码、令牌、个人隐私数据。

4.3 sessionId 约定

• 长度不超过 128 字节，只能包含字母、数字、下划线；
• 建议由发起端调用 genSessionId() 生成，再通过可信通道传给对端；
• 常见传递方式：二维码、短码输入、同一账号下的业务服务器下发、近场通信；
• sessionId 不是安全凭证：任何拿到它的同应用可信设备都能加入，加入权限要靠业务侧校验；
• 会话结束要显式退出，避免设备长期停留在协同组。

5 端到端实现

5.1 封装同步服务

import { distributedDataObject } from '@kit.ArkData';
import { common } from '@kit.AbilityKit';
import { BusinessError } from '@kit.BasicServicesKit';
import { hilog } from '@kit.PerformanceAnalysisKit';

const TAG = 'DDO';

export type RemoteChangeHandler = (state: PlaybackState) => void;
export type PeerStatusHandler = (networkId: string, online: boolean) => void;

export class PlaybackSync {
  private context: common.UIAbilityContext;
  private dataObject: distributedDataObject.DataObject | null = null;
  private sessionId: string = '';
  private onRemoteChange: RemoteChangeHandler = () => {};
  private onPeerStatus: PeerStatusHandler = () => {};
  private lastWriteAt: number = 0;

  constructor(context: common.UIAbilityContext) {
    this.context = context;
  }

  async createSession(local: PlaybackState, onChange: RemoteChangeHandler,
    onStatus: PeerStatusHandler): Promise<string> {
    this.onRemoteChange = onChange;
    this.onPeerStatus = onStatus;
    this.dataObject = distributedDataObject.create(this.context, local);

    const sessionId = distributedDataObject.genSessionId();
    await this.dataObject.setSessionId(sessionId);
    this.sessionId = sessionId;
    this.bindEvents();
    hilog.info(0x0000, TAG, 'session created: %{public}s', sessionId);
    return sessionId;
  }

  async joinSession(sessionId: string, local: PlaybackState,
    onChange: RemoteChangeHandler, onStatus: PeerStatusHandler): Promise<void> {
    this.onRemoteChange = onChange;
    this.onPeerStatus = onStatus;
    this.dataObject = distributedDataObject.create(this.context, local);
    await this.dataObject.setSessionId(sessionId);
    this.sessionId = sessionId;
    this.bindEvents();
    hilog.info(0x0000, TAG, 'session joined: %{public}s', sessionId);
  }

  private bindEvents(): void {
    if (!this.dataObject) {
      return;
    }

    this.dataObject.on('change', (sessionId: string, fields: Array<string>) => {
      if (sessionId !== this.sessionId || !this.dataObject) {
        return;
      }
      hilog.info(0x0000, TAG, 'remote change: %{public}s', fields.join(','));
      this.onRemoteChange(this.snapshot());
    });

    this.dataObject.on('status', (sessionId: string, networkId: string, status: 'online' | 'offline') => {
      if (sessionId !== this.sessionId) {
        return;
      }
      hilog.info(0x0000, TAG, 'peer %{public}s %{public}s', networkId, status);
      this.onPeerStatus(networkId, status === 'online');
    });
  }

  snapshot(): PlaybackState {
    const state = new PlaybackState();
    if (!this.dataObject) {
      return state;
    }
    state.title = this.dataObject['title'] as string;
    state.position = this.dataObject['position'] as number;
    state.playing = this.dataObject['playing'] as boolean;
    state.volume = this.dataObject['volume'] as number;
    state.updatedAt = this.dataObject['updatedAt'] as number;
    state.writer = this.dataObject['writer'] as string;
    state.version = this.dataObject['version'] as number;
    return state;
  }

  write(next: PlaybackState, throttleMs: number = 300): void {
    if (!this.dataObject) {
      return;
    }
    const now = Date.now();
    if (now - this.lastWriteAt < throttleMs) {
      return;
    }
    this.lastWriteAt = now;

    this.dataObject['title'] = next.title;
    this.dataObject['position'] = next.position;
    this.dataObject['playing'] = next.playing;
    this.dataObject['volume'] = next.volume;
    this.dataObject['updatedAt'] = now;
    this.dataObject['writer'] = next.writer;
    this.dataObject['version'] = next.version + 1;
  }

  async persist(): Promise<void> {
    if (!this.dataObject) {
      return;
    }
    try {
      const result = await this.dataObject.save('local');
      hilog.info(0x0000, TAG, 'saved version=%{public}d', result.version);
    } catch (err) {
      const e = err as BusinessError;
      hilog.error(0x0000, TAG, 'save failed: %{public}d %{public}s', e.code, e.message);
    }
  }

  async leave(): Promise<void> {
    if (!this.dataObject) {
      return;
    }
    try {
      await this.dataObject.setSessionId('');
    } catch (err) {
      const e = err as BusinessError;
      hilog.error(0x0000, TAG, 'leave failed: %{public}d %{public}s', e.code, e.message);
    }
    this.dataObject.off('change');
    this.dataObject.off('status');
    this.dataObject = null;
    this.sessionId = '';
    hilog.info(0x0000, TAG, 'session left');
  }
}

说明：

• create 传入的 local 对象是数据源，后续通过 dataObject['字段'] 读写；
• snapshot() 逐字段读取，避免把整个代理对象直接交给 UI 层；
• write() 做 300ms 节流，防止进度条每帧写入把同步通道打满；
• leave() 先退出会话再解绑监听，顺序不要颠倒；
• off('change') 不传回调会移除该对象的全部 change 监听，适合页面销毁场景。

5.2 页面接入

@Entry
@Component
struct PlayerPage {
  @State position: number = 0;
  @State playing: boolean = false;
  @State peerOnline: boolean = false;
  @State sessionId: string = '';
  private sync: PlaybackSync | null = null;
  private state: PlaybackState = new PlaybackState();
  private sliderChanging: boolean = false;

  async aboutToAppear(): Promise<void> {
    const context = getContext(this) as common.UIAbilityContext;
    const granted = await ensureSyncPermission(context);
    if (!granted) {
      return;
    }

    this.sync = new PlaybackSync(context);
    this.state.title = '示例视频';
    this.state.writer = 'local';

    // 发起端：创建会话；加入端改为 joinSession(传入 sessionId)
    this.sessionId = await this.sync.createSession(
      this.state,
      (remote: PlaybackState) => {
        // 用户正在拖动进度条时，不让远端值打断操作
        if (this.sliderChanging) {
          return;
        }
        this.position = remote.position;
        this.playing = remote.playing;
        this.state = remote;
      },
      (networkId: string, online: boolean) => {
        this.peerOnline = online;
      }
    );
  }

  aboutToDisappear(): void {
    if (this.sync) {
      this.sync.leave();
      this.sync = null;
    }
  }

  build() {
    Column({ space: 16 }) {
      Text(this.peerOnline ? '对端在线' : '等待对端加入')
        .fontSize(14)
        .fontColor(this.peerOnline ? '#1F8A4C' : '#999999')

      Slider({ value: this.position, max: 100 })
        .onChange((value: number, mode: SliderChangeMode) => {
          this.sliderChanging = mode !== SliderChangeMode.End;
          this.position = value;
          if (mode === SliderChangeMode.End && this.sync) {
            this.state.position = value;
            this.sync.write(this.state, 0);
          }
        })

      Button(this.playing ? '暂停' : '播放')
        .onClick(() => {
          this.playing = !this.playing;
          this.state.playing = this.playing;
          if (this.sync) {
            this.sync.write(this.state, 0);
          }
        })
    }
    .padding(24)
    .width('100%')
    .height('100%')
  }
}

交互细节：

1. 拖动时不接受远端值：否则两端互相拉扯，进度条抖动；
2. 拖动结束立即写：throttleMs 传 0，保证最终位置能同步；
3. 播放/暂停单独写一次：避免依赖节流窗口；
4. 状态提示明确：对端离线时给出可见提示，而不是让用户以为同步坏了。

6 典型场景：跨设备播放进度同步

6.1 角色划分

角色	职责	写入权限
控制端（手机）	下发播放/暂停/进度	可写
展示端（大屏）	播放并回传状态	只写播放器实际状态

推荐「单写多读」：控制端是唯一写入方，展示端只读。这样可以从根本上避免并发写冲突。若确实需要双向控制，再用 version + writer 做冲突判定。

6.2 同步时序

控制端                       展示端
  │  创建 sessionId            │
  │─── 二维码/短码传递 ───────→│
  │                            │ joinSession(sessionId)
  │←──── on('status') online ──│
  │  拖动进度条 → write()      │
  │──── change(position) ─────→│
  │                            │ 更新播放器
  │←── change(playing) ────────│ 回传实际状态
  │  离开页面 → leave()        │
  │──── status offline ───────→│

6.3 关键细节

• 进度同步建议 300ms~1s 一次，不要每帧写；
• 播放状态变化立即写，延迟会直接影响体验；
• 展示端回传的 position 应取播放器真实值，而不是本地估算；
• 网络恢复后，以 updatedAt 较新的状态为准，必要时由控制端全量重写一次。

7 可靠性设计

7.1 断网与重连

on('status') 的 offline 只表示对端暂时不可达，不代表会话失效：

1. 收到 offline 时，界面提示「对端离线，正在重连」；
2. 不要立即 setSessionId('')，否则会主动退出会话；
3. 保持本地状态可写，但标记为「待同步」；
4. 收到 online 后，主动全量写一次当前状态，避免增量丢失；
5. 若 30 秒仍无 online，提示用户检查网络或重新加入。

7.2 并发写冲突

分布式数据对象不提供事务与锁，多端同时写同一字段时，结果取决于到达顺序。工程上按优先级处理：

方案	适用场景	做法
单写多读	手机控制大屏	只允许控制端写，展示端只读
版本号	双向控制	写入前比较 version，落后者放弃或重试
时间戳	弱一致场景	以 updatedAt 较大者为准，需容忍时钟误差
字段拆分	双方各自维护不同字段	控制端写进度，展示端写音量，互不覆盖

无论采用哪种方案，都要在 UI 上避免「来回跳」：本地正在操作时，短暂忽略远端更新。

7.3 持久化与恢复

save(deviceId) 用于「应用退出后重新进入时恢复状态」，不是实时同步的替代品：

await this.dataObject.save('local');
// 恢复时：重新 create 并 setSessionId 后，框架会尝试恢复已保存的数据

需要知道的限制（官方行为）：

• 保存数据在 24 小时后释放；
• 应用卸载会释放；
• 数据成功恢复后释放；
• save 失败可能返回 801（能力不支持）或 401（参数错误），要做降级处理。

因此重要业务数据仍应落库（关系型数据库 / preferences），分布式数据对象只做「协同态」。

7.4 内存与性能

1. 对象字段控制在 20 个以内，避免每次变更携带大量字段；
2. 单个对象建议 KB 级，不要塞入长文本、图片 base64；
3. change 回调里不做重计算，只更新必要的 @State；
4. 高频写入必须节流，并在页面不可见时降低频率；
5. 页面销毁时 off 监听、清空引用，防止回调持有页面导致泄漏。

8 调试验证

8.1 日志

# 分布式数据对象相关日志
hdc shell "hilog | grep -iE 'DDO|distributed_data|dataobject'"

# 设备互信与软总线状态
hdc shell "hilog | grep -iE 'softbus|device_manager'"

建议应用侧日志统一带上 sessionId、字段名、version，双端日志按时间对齐即可还原同步过程。

8.2 测试用例

用例	步骤	期望
基本同步	A 创建会话，B 加入，A 改进度	B 在 1 秒内更新，字段名正确
双向同步	B 改播放状态	A 收到 change 并刷新
断网恢复	传输中关闭 Wi-Fi 10 秒	状态变 offline，恢复后 online 并追平
主动退出	A 调用 leave	B 收到 offline，A 本地不再收变更
持久化	save 后杀进程重进	24 小时内可恢复，版本号连续
并发写	两端同时改进度	无崩溃，最终值符合约定策略
高频写	1 秒写 20 次	有节流，UI 帧率无明显下降
页面销毁	退出页面再进入	无重复回调、无内存增长

8.3 验收标准

1. 局域网内同步延迟稳定在 1 秒以内；
2. 断网恢复后 3 秒内状态追平；
3. 连续操作 30 分钟无重复回调、无监听泄漏；
4. 页面销毁后不再收到任何 change/status 回调；
5. 异常场景（无权限、未互信、sessionId 非法）均有明确提示。

9 常见问题

现象	可能原因	处理方式
on('change') 不触发	两端 sessionId 不一致	打印并核对 sessionId
加入会话失败	未授权 DISTRIBUTED_DATASYNC	动态申请权限，检查返回码 201
设备看不到彼此	未互信或网络隔离	先走设备发现排查流程
修改属性无反应	字段名与创建时不一致	字段名必须完全一致
嵌套对象不同步	使用了不支持的复杂类型	改为基本类型或数组，扁平化
status 一直 offline	对端未加入同一会话	确认对端已 setSessionId
退出后仍收到回调	未 off 监听	页面销毁时解绑
两端进度来回跳	双向写 + 无冲突策略	单写多读或加版本号
save 后重启未恢复	超过 24 小时或未重新加入会话	重要数据落库，save 只做辅助
大对象同步卡顿	对象过大或写入过频	拆分字段、节流、移出大文本

10 最佳实践

1. 先发现，再同步：设备发现与互信是前置条件，不要试图用 sessionId 绕过；
2. 单写多读优先：能从设计上消除并发冲突，就不要靠代码打补丁；
3. 状态与 UI 分离：同步对象只放纯数据，UI 状态由页面自行维护；
4. sessionId 走可信通道：二维码、短码、业务服务器下发均可，但不要当作鉴权凭证；
5. 写入必节流，读取必判空：高频场景下这两条能避免大部分抖动与崩溃；
6. 监听与生命周期绑定：aboutToAppear 建立，aboutToDisappear 释放；
7. 持久化只做辅助：重要数据落库，save() 只用于短时恢复；
8. 异常路径要可见：无权限、未互信、对端离线，都要给用户明确提示，而不是静默失败。