HarmonyOS开发：分布式测试——跨设备测试

📌 核心要点：分布式是鸿蒙的杀手特性——设备间数据同步、能力流转、跨设备调用。但分布式测试也是最难的：设备上下线、网络断连、数据冲突、一致性验证……这些场景在单设备上根本测不出来，必须系统性地覆盖。

一、背景与动机

你的App支持"手机上编辑、平板上继续"——听起来很美好。但实际跑起来——手机上改了数据，平板上还是旧的；两台设备同时改了同一条数据，谁说了算？设备A下线了，设备B还在傻等同步结果。

分布式场景的Bug有个特点——在单设备上完全正常，多设备一组合就出问题。你测了100遍手机端，一切OK；加上平板端，数据同步就炸了。更要命的是，分布式Bug的复现条件极其苛刻——特定的设备组合、特定的网络状况、特定的操作时序——你在办公室根本复现不了用户遇到的问题。

手动测试分布式？你拿两台设备来回操作？改一下手机上的数据，去看平板上有没有同步——这种测试方式，5个场景你就累趴了。分布式测试就是用代码系统性地验证跨设备场景，确保数据同步可靠、设备上下线处理正确、一致性有保障。

二、核心原理

2.1 HarmonyOS分布式能力体系

flowchart TB
    A[HarmonyOS分布式能力] --> B[分布式数据同步]
    A --> C[分布式能力调用]
    A --> D[分布式任务迁移]
    A --> E[设备发现与管理]

    B --> B1[DistributedData<br/>分布式键值数据库]
    B --> B2[DistributedRdb<br/>分布式关系型数据库]
    B --> B3[DataShare<br/>数据共享]

    C --> C1[远程Ability启动]
    C --> C2[远程ServiceExtension]
    C --> C3[分布式RPC]

    D --> D1[Mission续播]
    D --> D2[页面流转]
    D --> D3[状态迁移]

    E --> E1[DeviceManager<br/>设备发现]
    E --> E2[设备认证]
    E --> E3[设备状态监听]

    classDef mainStyle fill:#4CAF50,stroke:#388E3C,color:#fff,font-weight:bold
    classDef dataStyle fill:#3498DB,stroke:#2980B9,color:#fff
    classDef abilityStyle fill:#2ECC71,stroke:#27AE60,color:#fff
    classDef migrateStyle fill:#F39C12,stroke:#E67E22,color:#fff
    classDef deviceStyle fill:#9B59B6,stroke:#8E44AD,color:#fff

    class A mainStyle
    class B,B1,B2,B3 dataStyle
    class C,C1,C2,C3 abilityStyle
    class D,D1,D2,D3 migrateStyle
    class E,E1,E2,E3 deviceStyle

2.2 分布式数据同步机制

HarmonyOS的分布式数据同步基于@ohos.data.distributedData和@ohos.data.relationalStore：

import distributedData from '@ohos.data.distributedData'

// 创建分布式KV数据库
const kvManager = distributedData.createKVManager({
  bundleName: 'com.example.app',
  userInfo: { userId: 'default', appId: 'com.example.app' }
})

const kvStore = await kvManager.getKVStore('sync-store', {
  createIfMissing: true,
  encrypt: false,
  autoSync: true,        // 自动同步
  kvStoreType: distributedData.KVStoreType.DEVICE_COLLABORATION
})

// 写入数据（自动同步到其他设备）
await kvStore.put('config-theme', 'dark')

// 监听数据变更
kvStore.on('dataChange', distributedData.SubscribeType.SUBSCRIBE_TYPE_REMOTE, (data) => {
  // 远程设备数据变更通知
})

2.3 分布式一致性模型

HarmonyOS默认使用最终一致性，冲突解决策略为"最后写入胜出"（Last Write Wins）。

三、代码实战

3.1 基础用法：跨设备数据同步验证

// DistributedDataHelper.ets - 分布式数据辅助工具
import distributedData from '@ohos.data.distributedData'

export class DistributedDataHelper {
  private kvManager: distributedData.KVManager
  private kvStore: distributedData.KVStore | null = null
  private changeCallbacks: Array<(key: string, value: string, deviceId: string) => void> = []

  constructor(bundleName: string) {
    this.kvManager = distributedData.createKVManager({
      bundleName,
      userInfo: { userId: 'default', appId: bundleName }
    })
  }

  // 初始化KV Store
  async init(storeId: string): Promise<void> {
    this.kvStore = await this.kvManager.getKVStore(storeId, {
      createIfMissing: true,
      encrypt: false,
      autoSync: true,
      kvStoreType: distributedData.KVStoreType.DEVICE_COLLABORATION
    })

    // 监听远程数据变更
    this.kvStore.on('dataChange', distributedData.SubscribeType.SUBSCRIBE_TYPE_REMOTE, (data) => {
      if (data.updateRecords) {
        for (const entry of data.updateRecords) {
          for (const callback of this.changeCallbacks) {
            callback(entry.key, entry.value.value as string, entry.deviceId)
          }
        }
      }
    })
  }

  // 写入数据
  async put(key: string, value: string): Promise<void> {
    if (!this.kvStore) throw new Error('KV Store未初始化')
    await this.kvStore.put(key, value)
  }

  // 读取数据
  async get(key: string): Promise<string | undefined> {
    if (!this.kvStore) throw new Error('KV Store未初始化')
    const entries = await this.kvStore.getEntries(key)
    if (entries.length > 0) {
      return entries[0].value.value as string
    }
    return undefined
  }

  // 删除数据
  async delete(key: string): Promise<void> {
    if (!this.kvStore) throw new Error('KV Store未初始化')
    await this.kvStore.delete(key)
  }

  // 注册变更回调
  onDataChange(callback: (key: string, value: string, deviceId: string) => void): void {
    this.changeCallbacks.push(callback)
  }

  // 手动触发同步
  async sync(deviceIds: Array<string>, mode: distributedData.SyncMode = distributedData.SyncMode.PUSH_PULL): Promise<void> {
    if (!this.kvStore) throw new Error('KV Store未初始化')
    await this.kvStore.sync(deviceIds, mode)
  }

  // 销毁
  async destroy(): Promise<void> {
    if (this.kvStore) {
      this.kvStore.off('dataChange')
    }
  }
}

// ==================== 数据同步测试 ====================
import { describe, it, beforeAll, afterAll, assertEqual, assertTrue, assertNotNull } from '@ohos/hypium'
import { DistributedDataHelper } from '../DistributedDataHelper'

export default function dataSyncTest() {
  describe('跨设备数据同步测试', () => {
    let deviceA: DistributedDataHelper
    let deviceB: DistributedDataHelper

    beforeAll(async () => {
      deviceA = new DistributedDataHelper('com.example.app')
      deviceB = new DistributedDataHelper('com.example.app')
      await deviceA.init('test-sync-store')
      await deviceB.init('test-sync-store')
    })

    afterAll(async () => {
      await deviceA.destroy()
      await deviceB.destroy()
    })

    it('A写入数据_B同步后可读取', 0, async () => {
      await deviceA.put('theme', 'dark')

      // 等待同步（实际场景中需要等待dataChange回调）
      // 在测试中，由于是同一个KVManager，数据立即可见
      const value = await deviceB.get('theme')
      assertEqual(value, 'dark', 'B应能读到A写入的数据')
    })

    it('A更新数据_B读到最新值', 0, async () => {
      await deviceA.put('language', 'zh')
      await deviceA.put('language', 'en')  // 更新

      const value = await deviceB.get('language')
      assertEqual(value, 'en', 'B应读到最新值')
    })

    it('A删除数据_B也读不到', 0, async () => {
      await deviceA.put('temp-key', 'temp-value')
      await deviceA.delete('temp-key')

      const value = await deviceB.get('temp-key')
      assertEqual(value, undefined, '删除后B应读不到')
    })

    it('变更回调_远程数据变更时触发', 0, async () => {
      let callbackFired = false
      let callbackKey = ''
      let callbackValue = ''

      deviceB.onDataChange((key, value, deviceId) => {
        callbackFired = true
        callbackKey = key
        callbackValue = value
      })

      await deviceA.put('callback-test', 'hello')

      // 等待回调触发（可能需要短暂延迟）
      await new Promise<void>(resolve => setTimeout(resolve, 500))

      assertTrue(callbackFired, '变更回调应被触发')
      assertEqual(callbackKey, 'callback-test')
      assertEqual(callbackValue, 'hello')
    })
  })
}

3.2 进阶用法：设备上下线场景测试

// DeviceManager.ets - 设备管理器
import deviceManager from '@ohos.distributedDeviceManager'

export interface DeviceInfo {
  deviceId: string
  deviceName: string
  deviceType: string
  isOnline: boolean
}

export class DeviceManager {
  private dmInstance: deviceManager.DeviceManager | null = null
  private knownDevices: Map<string, DeviceInfo> = new Map()
  private onDeviceOnline: ((device: DeviceInfo) => void) | null = null
  private onDeviceOffline: ((device: DeviceInfo) => void) | null = null

  // 初始化设备管理器
  async init(): Promise<void> {
    this.dmInstance = deviceManager.createDeviceManager('com.example.app')

    this.dmInstance.on('deviceStateChange', (data) => {
      const device: DeviceInfo = {
        deviceId: data.device.id,
        deviceName: data.device.name,
        deviceType: data.device.deviceType,
        isOnline: data.action === deviceManager.DeviceStateChangeAction.ONLINE
      }
      this.knownDevices.set(device.deviceId, device)

      if (device.isOnline) {
        this.onDeviceOnline?.(device)
      } else {
        this.onDeviceOffline?.(device)
      }
    })
  }

  // 发现设备
  async discoverDevices(): Promise<Array<DeviceInfo>> {
    if (!this.dmInstance) return []
    this.dmInstance.startDiscovering()
    // 等待发现结果
    await new Promise<void>(resolve => setTimeout(resolve, 3000))
    this.dmInstance.stopDiscovering()
    return Array.from(this.knownDevices.values())
  }

  // 获取在线设备
  getOnlineDevices(): Array<DeviceInfo> {
    return Array.from(this.knownDevices.values()).filter(d => d.isOnline)
  }

  // 获取所有已知设备
  getAllDevices(): Array<DeviceInfo> {
    return Array.from(this.knownDevices.values())
  }

  // 设置回调
  setOnDeviceOnline(callback: (device: DeviceInfo) => void): void {
    this.onDeviceOnline = callback
  }

  setOnDeviceOffline(callback: (device: DeviceInfo) => void): void {
    this.onDeviceOffline = callback
  }

  // 模拟设备上线（测试用）
  simulateDeviceOnline(deviceId: string, deviceName: string): void {
    const device: DeviceInfo = {
      deviceId,
      deviceName,
      deviceType: 'tablet',
      isOnline: true
    }
    this.knownDevices.set(deviceId, device)
    this.onDeviceOnline?.(device)
  }

  // 模拟设备下线（测试用）
  simulateDeviceOffline(deviceId: string): void {
    const device = this.knownDevices.get(deviceId)
    if (device) {
      device.isOnline = false
      this.onDeviceOffline?.(device)
    }
  }

  // 销毁
  destroy(): void {
    this.dmInstance?.off('deviceStateChange')
    this.dmInstance?.destroyInstance()
  }
}

// ==================== 设备上下线测试 ====================
import { describe, it, beforeAll, afterAll, beforeEach, assertEqual, assertTrue } from '@ohos/hypium'
import { DeviceManager, DeviceInfo } from '../DeviceManager'
import { DistributedDataHelper } from '../DistributedDataHelper'

export default function deviceOnlineOfflineTest() {
  describe('设备上下线场景测试', () => {
    let deviceManager: DeviceManager
    let localData: DistributedDataHelper
    let pendingSyncData: Map<string, string> = new Map()

    beforeAll(async () => {
      deviceManager = new DeviceManager()
      await deviceManager.init()
      localData = new DistributedDataHelper('com.example.app')
      await localData.init('device-sync-store')
    })

    afterAll(async () => {
      deviceManager.destroy()
      await localData.destroy()
    })

    beforeEach(() => {
      pendingSyncData.clear()
    })

    it('设备上线_自动同步离线期间的数据', 0, async () => {
      // 模拟设备上线
      let onlineDevice: DeviceInfo | null = null
      deviceManager.setOnDeviceOnline((device) => {
        onlineDevice = device
      })

      deviceManager.simulateDeviceOnline('device-tablet-001', '我的平板')

      assertNotNull(onlineDevice)
      assertEqual(onlineDevice?.deviceId, 'device-tablet-001')
      assertEqual(onlineDevice?.isOnline, true)
    })

    it('设备下线_本地操作进入待同步队列', 0, async () => {
      // 模拟设备下线
      deviceManager.simulateDeviceOnline('device-tablet-001', '我的平板')
      deviceManager.simulateDeviceOffline('device-tablet-001')

      // 离线期间本地操作
      const onlineDevices = deviceManager.getOnlineDevices()
      assertEqual(onlineDevices.length, 0, '应无在线设备')

      // 本地写入数据（应进入待同步队列）
      pendingSyncData.set('offline-key-1', 'offline-value-1')
      pendingSyncData.set('offline-key-2', 'offline-value-2')
      assertEqual(pendingSyncData.size, 2, '应有2条待同步数据')
    })

    it('设备重新上线_待同步数据自动推送', 0, async () => {
      // 先下线
      deviceManager.simulateDeviceOnline('device-tablet-001', '我的平板')
      deviceManager.simulateDeviceOffline('device-tablet-001')

      // 离线期间写入
      pendingSyncData.set('sync-key', 'sync-value')

      // 设备重新上线
      let syncTriggered = false
      deviceManager.setOnDeviceOnline(async (device) => {
        if (pendingSyncData.size > 0) {
          // 推送待同步数据
          for (const [key, value] of pendingSyncData) {
            await localData.put(key, value)
          }
          syncTriggered = true
        }
      })

      deviceManager.simulateDeviceOnline('device-tablet-001', '我的平板')
      assertTrue(syncTriggered, '设备上线应触发待同步数据推送')
    })

    it('多设备同时在线_数据广播到所有设备', 0, async () => {
      deviceManager.simulateDeviceOnline('device-phone-001', '我的手机')
      deviceManager.simulateDeviceOnline('device-tablet-001', '我的平板')

      const onlineDevices = deviceManager.getOnlineDevices()
      assertEqual(onlineDevices.length, 2, '应有2台在线设备')

      // 写入数据应同步到所有在线设备
      await localData.put('broadcast-key', 'broadcast-value')
    })

    it('设备反复上下线_状态正确切换', 0, async () => {
      const stateChanges: Array<string> = []

      deviceManager.setOnDeviceOnline(() => { stateChanges.push('online') })
      deviceManager.setOnDeviceOffline(() => { stateChanges.push('offline') })

      deviceManager.simulateDeviceOnline('device-001', '设备A')
      deviceManager.simulateDeviceOffline('device-001')
      deviceManager.simulateDeviceOnline('device-001', '设备A')
      deviceManager.simulateDeviceOffline('device-001')
      deviceManager.simulateDeviceOnline('device-001', '设备A')

      assertEqual(stateChanges.join('-'), 'online-offline-online-offline-online',
        '状态应正确切换')
    })
  })
}

3.3 完整示例：分布式一致性测试

// DistributedConsistencyTest.ets - 分布式一致性测试
import { describe, it, beforeAll, afterAll, assertEqual, assertTrue } from '@ohos/hypium'
import { DistributedDataHelper } from '../DistributedDataHelper'

export default function distributedConsistencyTest() {
  describe('分布式一致性测试', () => {
    let deviceA: DistributedDataHelper
    let deviceB: DistributedDataHelper
    let deviceC: DistributedDataHelper

    beforeAll(async () => {
      deviceA = new DistributedDataHelper('com.example.app')
      deviceB = new DistributedDataHelper('com.example.app')
      deviceC = new DistributedDataHelper('com.example.app')
      await deviceA.init('consistency-store')
      await deviceB.init('consistency-store')
      await deviceC.init('consistency-store')
    })

    afterAll(async () => {
      await deviceA.destroy()
      await deviceB.destroy()
      await deviceC.destroy()
    })

    it('最终一致性_所有设备最终读到相同值', 0, async () => {
      // A写入数据
      await deviceA.put('consistency-key', 'final-value')

      // 等待同步
      await new Promise<void>(resolve => setTimeout(resolve, 1000))

      // 验证所有设备读到的值一致
      const valueA = await deviceA.get('consistency-key')
      const valueB = await deviceB.get('consistency-key')
      const valueC = await deviceC.get('consistency-key')

      assertEqual(valueA, valueB, 'A和B应一致')
      assertEqual(valueB, valueC, 'B和C应一致')
      assertEqual(valueA, 'final-value', '最终值应正确')
    })

    it('写冲突_最后写入胜出', 0, async () => {
      // A和B同时写入同一个key
      await deviceA.put('conflict-key', 'from-A')
      await deviceB.put('conflict-key', 'from-B')

      // 等待同步和冲突解决
      await new Promise<void>(resolve => setTimeout(resolve, 1000))

      // 所有设备最终应读到同一个值（最后写入的胜出）
      const valueA = await deviceA.get('conflict-key')
      const valueB = await deviceB.get('conflict-key')
      const valueC = await deviceC.get('conflict-key')

      // 三者应一致
      assertEqual(valueA, valueB, '冲突解决后A和B应一致')
      assertEqual(valueB, valueC, '冲突解决后B和C应一致')
    })

    it('批量写入_顺序一致', 0, async () => {
      // A连续写入多个key
      for (let i = 0; i < 10; i++) {
        await deviceA.put(`batch-key-${i}`, `batch-value-${i}`)
      }

      // 等待同步
      await new Promise<void>(resolve => setTimeout(resolve, 2000))

      // 验证B能读到所有数据
      let allPresent = true
      for (let i = 0; i < 10; i++) {
        const value = await deviceB.get(`batch-key-${i}`)
        if (value !== `batch-value-${i}`) {
          allPresent = false
          break
        }
      }
      assertTrue(allPresent, '批量写入后所有数据应可读取')
    })

    it('删除操作_同步到所有设备', 0, async () => {
      await deviceA.put('delete-key', 'will-be-deleted')
      await new Promise<void>(resolve => setTimeout(resolve, 500))

      // A删除
      await deviceA.delete('delete-key')
      await new Promise<void>(resolve => setTimeout(resolve, 500))

      // 验证B和C也读不到
      const valueB = await deviceB.get('delete-key')
      const valueC = await deviceC.get('delete-key')
      assertEqual(valueB, undefined, 'B应读不到已删除的数据')
      assertEqual(valueC, undefined, 'C应读不到已删除的数据')
    })

    it('网络分区恢复后_数据最终一致', 0, async () => {
      // 模拟网络分区：A写入后B暂时读不到
      await deviceA.put('partition-key', 'from-A')

      // 在实际测试中，网络分区很难模拟
      // 这里验证的是：分区恢复后，数据最终一致
      await new Promise<void>(resolve => setTimeout(resolve, 1000))

      const valueB = await deviceB.get('partition-key')
      assertEqual(valueB, 'from-A', '分区恢复后B应能读到A的数据')
    })
  })
}

四、踩坑与注意事项

坑点1：同步延迟导致测试断言失败

分布式数据同步不是实时的——A写入后，B可能要几百毫秒甚至几秒才能读到。如果你的断言在同步完成前就执行了，测试就会失败——但不是Bug，只是时序问题。

建议：在断言前加等待（setTimeout），或者用回调机制（监听dataChange事件）等待同步完成后再断言。不要用固定延迟，用轮询+超时的方式更可靠。

坑点2：设备ID在不同场景下不同

同一个物理设备，在不同API中返回的deviceId可能不同——deviceManager返回的是网络ID，distributedData用的是设备ID，abilityManager又是一种ID。如果你用错误的ID做同步，数据根本同步不过去。

建议：统一使用deviceManager获取的设备ID，其他API需要的ID通过转换函数获取。不要硬编码设备ID。

坑点3：分布式权限配置遗漏

分布式功能需要ohos.permission.DISTRIBUTED_DATASYNC权限，而且需要在module.json5中声明。如果你忘了声明，编译能通过但运行时直接报错。

建议：在项目初期就检查权限配置，把分布式权限作为Checklist的一部分。

坑点4：大量数据同步导致性能问题

一次性同步几千条数据——同步时间可能要几十秒，期间设备CPU和网络都被占满，UI卡顿。

建议：分批同步，每批100-200条。优先同步关键数据，非关键数据延迟同步。使用增量同步而不是全量同步。

坑点5：设备认证失败

两台设备要同步数据，必须先完成设备认证（同一个华为账号或扫码认证）。如果认证失败，同步请求直接被拒绝，但错误信息可能不明确。

建议：在同步前检查设备认证状态，认证失败时给用户明确提示。测试环境中确保所有设备已完成认证。

坑点6：KV Store类型选错

KVStoreType.SINGLE_VERSION是单版本数据库，不支持分布式同步；KVStoreType.DEVICE_COLLABORATION才是分布式协作数据库。如果你用了SINGLE_VERSION，数据根本不会同步。

建议：需要分布式同步的场景，必须使用DEVICE_COLLABORATION类型。在初始化KV Store时明确指定类型。

坑点7：测试环境设备不足

分布式测试至少需要2台设备，但开发者可能只有1台。模拟器不支持分布式功能，真机又不够。

建议：在测试代码中用Mock模拟多设备场景——创建多个DistributedDataHelper实例模拟不同设备。核心逻辑可以在单设备上验证，但关键路径必须在多设备上验证。

五、HarmonyOS 6适配说明

API差异表

行为变更

1. 冲突解决策略可配置：HarmonyOS 6支持自定义冲突解决策略，不再局限于"最后写入胜出"。可以注册冲突回调，根据业务逻辑决定保留哪个版本。

2. 选择性同步：支持按条件过滤同步数据——只同步特定前缀的key，或者只同步最近N天的数据，减少不必要的同步开销。

3. @DistributedTest装饰器：测试框架新增分布式测试装饰器，自动管理多设备环境、数据同步等待、一致性验证。

适配代码

// HarmonyOS 6分布式测试
import { describe, it, assertEqual, assertTrue } from '@ohos/hypium'
import { DistributedTest, DeviceRole } from '@ohos/hypium'
import { ConsistencyChecker, ConsistencyLevel } from '@ohos.data.consistency'

export default function harmonyOS6DistributedTest() {
  describe('HarmonyOS 6分布式测试', () => {
    // 自动配置多设备环境
    @DistributedTest({
      devices: [
        { role: DeviceRole.PRIMARY, type: 'phone' },
        { role: DeviceRole.SECONDARY, type: 'tablet' }
      ],
      syncTimeout: 5000
    })
    it('跨设备数据同步_自动等待', 0, async () => {
      // 在主设备写入
      await kvStore.put('test-key', 'test-value')

      // 框架自动等待同步完成
      const value = await secondaryKvStore.get('test-key')
      assertEqual(value, 'test-value', '副设备应读到主设备的数据')
    })

    it('一致性校验_自动验证', 0, async () => {
      await kvStore.put('consistency-key', 'value')

      // 使用一致性校验框架
      const checker = new ConsistencyChecker()
      const result = await checker.check({
        key: 'consistency-key',
        level: ConsistencyLevel.EVENTUAL,
        timeout: 3000
      })
      assertTrue(result.consistent, '所有设备数据应最终一致')
    })

    it('自定义冲突解决策略', 0, async () => {
      const store = await kvManager.getKVStore('conflict-store', {
        conflictResolution: (local, remote) => {
          // 业务自定义：优先保留数值更大的版本
          if (local.version > remote.version) return local
          return remote
        }
      })

      await store.put('versioned-key', JSON.stringify({ value: 'v2', version: 2 }))
    })
  })
}

六、总结

分布式测试是鸿蒙测试中最难的部分——多设备、异步同步、冲突解决、一致性验证，每个环节都可能出问题。同步延迟是测试中最常见的坑，断言太早就失败，太晚又浪费时间——用回调+超时代替固定延迟。设备上下线是另一个重点——离线时的操作要排队，上线后要自动同步，这个流程必须测。一致性是底线——所有设备最终必须读到相同的值，冲突解决策略必须明确。分布式测试的核心原则：永远不要假设同步是实时的，永远不要假设设备永远在线。