HarmonyOS开发分布式任务调度：Ability跨设备迁移

背景

还记得那个场景吗？你在手机上看着视频，突然想在大屏电视上继续播放；或者在平板上编辑文档，走到客厅想用电视的大屏幕展示。这就是分布式任务调度要解决的问题——让应用在不同设备间无缝迁移。

传统开发模式下，每个设备都是孤岛。你在手机上打开的应用，想在电视上继续用，只能重新打开、重新登录、重新找到刚才的位置。这种体验，说实话，挺糟糕的。

HarmonyOS的分布式任务调度能力，本质上是在构建一个"超级终端"的概念。你的手机、平板、电视、手表不再是独立的设备，而是一个整体的各个部分。Ability跨设备迁移，就是让应用的界面和数据能够在这些设备间自由流动。

这事儿说起来简单，实现起来可不轻松。需要考虑设备发现、连接建立、状态同步、数据传输、界面恢复等一系列复杂流程。好在HarmonyOS已经把这些能力封装好了，我们只需要按照规范调用就行。

核心原理

分布式任务调度的架构

分布式任务调度基于以下几个核心组件：

1. 分布式数据管理：负责应用状态的序列化和传输
2. 设备管理服务：负责设备发现、认证和连接
3. Ability迁移框架：负责Ability的冻结、迁移和恢复
4. 分布式安全机制：确保跨设备通信的安全性
   

迁移流程详解

Ability迁移分为三个阶段：冻结阶段、传输阶段、恢复阶段。

冻结阶段：源设备的Ability需要将当前状态保存到Want对象中。这个Want对象就像一个"快照"，记录了Ability的所有关键信息——页面路由、用户输入、滚动位置等等。

传输阶段：系统通过分布式软总线将Want对象传输到目标设备。这个过程对开发者是透明的，但底层涉及设备发现、安全认证、数据加密等复杂操作。

恢复阶段：目标设备根据Want对象创建Ability实例，并恢复到之前的状态。用户看到的就是应用"瞬移"到了新设备上，而且状态完全一致。

关键API解析

// 迁移相关的核心接口
interface IAbilityContinuation {
    // 保存状态数据
    onSaveData(want: Want): boolean;
    
    // 恢复状态数据
    onRestoreData(want: Want): boolean;
    
    // 迁移完成回调
    onContinuationDone(result: number): void;
}

这三个回调方法是实现迁移的核心。onSaveData负责序列化状态，onRestoreData负责反序列化，onContinuationDone则是迁移完成后的清理工作。

代码实战

示例一：基础迁移实现

先来看一个最简单的迁移示例。这是一个视频播放应用，支持从手机迁移到电视继续播放。

// VideoPlayAbility.ets
import UIAbility from '@ohos.app.ability.UIAbility';
import Want from '@ohos.app.ability.Want';
import AbilityConstant from '@ohos.app.ability.AbilityConstant';
import window from '@ohos.window';

export default class VideoPlayAbility extends UIAbility {
    // 播放状态数据
    private playState: VideoPlayState = {
        videoId: '',
        currentPosition: 0,
        isPlaying: false,
        volume: 50,
        brightness: 100
    };

    // 保存状态数据 - 迁移前调用
    onSaveData(want: Want): boolean {
        console.info('[VideoPlayAbility] onSaveData called');
        
        // 将播放状态序列化到Want的parameters中
        want.parameters = {
            ...want.parameters,
            'videoId': this.playState.videoId,
            'currentPosition': this.playState.currentPosition,
            'isPlaying': this.playState.isPlaying,
            'volume': this.playState.volume,
            'brightness': this.playState.brightness,
            'migrateTimestamp': Date.now() // 添加时间戳用于校验
        };
        
        console.info('[VideoPlayAbility] State saved:', JSON.stringify(this.playState));
        return true; // 返回true表示保存成功
    }

    // 恢复状态数据 - 迁移后调用
    onRestoreData(want: Want): boolean {
        console.info('[VideoPlayAbility] onRestoreData called');
        
        if (!want.parameters) {
            console.error('[VideoPlayAbility] No parameters in want');
            return false;
        }
        
        // 从Want中恢复播放状态
        this.playState = {
            videoId: want.parameters['videoId'] as string,
            currentPosition: want.parameters['currentPosition'] as number,
            isPlaying: want.parameters['isPlaying'] as boolean,
            volume: want.parameters['volume'] as number,
            brightness: want.parameters['brightness'] as number
        };
        
        // 验证数据有效性
        const migrateTimestamp = want.parameters['migrateTimestamp'] as number;
        const elapsed = Date.now() - migrateTimestamp;
        if (elapsed > 300000) { // 超过5分钟的数据可能失效
            console.warn('[VideoPlayAbility] Migrated data is stale, elapsed:', elapsed);
        }
        
        console.info('[VideoPlayAbility] State restored:', JSON.stringify(this.playState));
        return true;
    }

    // 迁移完成回调
    onContinuationDone(result: number): void {
        console.info('[VideoPlayAbility] onContinuationDone, result:', result);
        
        if (result === AbilityConstant.OnContinuationResult.CONTINUATION_SUCCESS) {
            // 迁移成功，可以做一些清理工作
            console.info('[VideoPlayAbility] Migration succeeded');
        } else {
            // 迁移失败，需要恢复本地状态
            console.error('[VideoPlayAbility] Migration failed, result:', result);
        }
    }

    onCreate(want: Want, launchParam: AbilityConstant.LaunchParam): void {
        console.info('[VideoPlayAbility] onCreate');
        
        // 如果是迁移启动，恢复状态
        if (launchParam.launchReason === AbilityConstant.LaunchReason.CONTINUATION) {
            console.info('[VideoPlayAbility] Launched by continuation');
            this.onRestoreData(want);
        }
    }
}

// 播放状态接口定义
interface VideoPlayState {
    videoId: string;          // 视频ID
    currentPosition: number;  // 播放位置（毫秒）
    isPlaying: boolean;       // 是否正在播放
    volume: number;           // 音量
    brightness: number;       // 亮度
}

这个示例展示了迁移的基本流程。注意onCreate方法中对launchReason的判断——这是区分正常启动和迁移启动的关键。

示例二：主动发起迁移

接下来看看如何主动发起迁移。用户点击"投屏"按钮时，需要调用系统API将Ability迁移到目标设备。

// MigrateController.ets
import deviceManager from '@ohos.distributedDeviceManager';
import Want from '@ohos.app.ability.Want';
import common from '@ohos.app.ability.common';

export class MigrateController {
    private context: common.UIAbilityContext;
    private deviceManager: deviceManager.DeviceManager | null = null;

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

    // 初始化设备管理器
    private async initDeviceManager(): Promise<void> {
        try {
            // 创建设备管理器实例
            this.deviceManager = deviceManager.createDeviceManager('com.example.videoapp');
            console.info('[MigrateController] DeviceManager created');
            
            // 监听设备状态变化
            this.deviceManager.on('deviceStateChange', (data: deviceManager.DeviceInfo) => {
                console.info('[MigrateController] Device state changed:', data.deviceId, data.deviceState);
            });
        } catch (error) {
            console.error('[MigrateController] Failed to create DeviceManager:', error);
        }
    }

    // 获取可用的目标设备列表
    async getAvailableDevices(): Promise<deviceManager.DeviceInfo[]> {
        if (!this.deviceManager) {
            console.error('[MigrateController] DeviceManager not initialized');
            return [];
        }

        try {
            // 获取已认证的在线设备
            const devices = this.deviceManager.getTrustedDeviceListSync();
            console.info('[MigrateController] Found', devices.length, 'trusted devices');
            
            // 过滤出支持迁移的设备（电视、平板等）
            const availableDevices = devices.filter(device => {
                // 设备类型：0-本设备，1-手机，2-平板，3-电视，4-智能手表
                return device.deviceType === 2 || device.deviceType === 3;
            });
            
            return availableDevices;
        } catch (error) {
            console.error('[MigrateController] Failed to get devices:', error);
            return [];
        }
    }

    // 发起迁移到指定设备
    async migrateToDevice(targetDeviceId: string): Promise<boolean> {
        console.info('[MigrateController] Migrating to device:', targetDeviceId);
        
        try {
            // 构建迁移参数
            const migrateParam: AbilityConstant.ContinuationState = {
                deviceId: targetDeviceId,
                fixScreen: true // 固定屏幕方向
            };
            
            // 调用系统的迁移接口
            await this.context.continueAbility(migrateParam);
            
            console.info('[MigrateController] Migration initiated successfully');
            return true;
        } catch (error) {
            const err = error as BusinessError;
            console.error('[MigrateController] Migration failed:', err.code, err.message);
            
            // 处理常见错误
            if (err.code === 401) {
                console.error('[MigrateController] Invalid parameter');
            } else if (err.code === 16000050) {
                console.error('[MigrateController] Cross-device migration not supported');
            }
            
            return false;
        }
    }

    // 带超时的迁移
    async migrateWithTimeout(targetDeviceId: string, timeoutMs: number = 10000): Promise<boolean> {
        return new Promise<boolean>((resolve) => {
            // 设置超时定时器
            const timeoutId = setTimeout(() => {
                console.warn('[MigrateController] Migration timeout');
                resolve(false);
            }, timeoutMs);

            // 发起迁移
            this.migrateToDevice(targetDeviceId).then((success) => {
                clearTimeout(timeoutId);
                resolve(success);
            });
        });
    }

    // 释放资源
    destroy(): void {
        if (this.deviceManager) {
            deviceManager.releaseDeviceManager(this.deviceManager);
            this.deviceManager = null;
        }
    }
}

// 业务错误接口
interface BusinessError {
    code: number;
    message: string;
}

这个控制器封装了设备发现和迁移发起的逻辑。实际使用时，通常会先展示设备列表让用户选择，然后调用migrateToDevice方法。

示例三：双向迁移与状态同步

有些场景需要支持双向迁移——比如在手机和电视之间来回切换。这就需要更复杂的状态管理。

// BidirectionalMigrateAbility.ets
import UIAbility from '@ohos.app.ability.UIAbility';
import Want from '@ohos.app.ability.Want';
import AbilityConstant from '@ohos.app.ability.AbilityConstant';
import distributedData from '@ohos.data.distributedData';

export default class BidirectionalMigrateAbility extends UIAbility {
    // 分布式键值存储，用于跨设备状态同步
    private kvStore: distributedData.SingleKVStore | null = null;
    private sessionId: string = '';

    // 应用状态
    private appState: AppState = {
        currentPage: 'home',
        scrollOffset: 0,
        selectedItems: [],
        lastSyncTime: 0
    };

    async onCreate(want: Want, launchParam: AbilityConstant.LaunchParam): Promise<void> {
        console.info('[BidirectionalMigrateAbility] onCreate');
        
        // 初始化分布式数据存储
        await this.initDistributedStore();
        
        // 处理迁移启动
        if (launchParam.launchReason === AbilityConstant.LaunchReason.CONTINUATION) {
            this.onRestoreData(want);
            
            // 从分布式存储同步最新状态
            await this.syncFromDistributedStore();
        }
    }

    // 初始化分布式键值存储
    private async initDistributedStore(): Promise<void> {
        try {
            // 创建分布式数据管理器
            const kvManager = distributedData.createKVManager({
                bundleName: 'com.example.app',
                userInfo: {
                    userId: 'default',
                    userType: distributedData.UserType.SAME_USER_ID
                }
            });
            
            // 创建键值存储
            this.kvStore = await kvManager.getKVStore({
                storeId: 'migrate_state_store',
                securityLevel: distributedData.SecurityLevel.S1
            });
            
            // 监听数据变化
            this.kvStore.on('dataChange', distributedData.SubscribeType.SUBSCRIBE_TYPE_REMOTE, 
                (data: distributedData.ChangeNotification) => {
                    console.info('[BidirectionalMigrateAbility] Remote data changed');
                    this.handleRemoteDataChange(data);
                });
            
            console.info('[BidirectionalMigrateAbility] Distributed store initialized');
        } catch (error) {
            console.error('[BidirectionalMigrateAbility] Failed to init distributed store:', error);
        }
    }

    // 保存状态到Want（用于迁移）
    onSaveData(want: Want): boolean {
        want.parameters = {
            ...want.parameters,
            'currentPage': this.appState.currentPage,
            'scrollOffset': this.appState.scrollOffset,
            'selectedItems': JSON.stringify(this.appState.selectedItems),
            'sessionId': this.sessionId || this.generateSessionId()
        };
        
        return true;
    }

    // 从Want恢复状态
    onRestoreData(want: Want): boolean {
        if (!want.parameters) return false;
        
        this.appState = {
            currentPage: want.parameters['currentPage'] as string,
            scrollOffset: want.parameters['scrollOffset'] as number,
            selectedItems: JSON.parse(want.parameters['selectedItems'] as string),
            lastSyncTime: Date.now()
        };
        
        this.sessionId = want.parameters['sessionId'] as string;
        
        return true;
    }

    // 从分布式存储同步状态
    private async syncFromDistributedStore(): Promise<void> {
        if (!this.kvStore || !this.sessionId) return;
        
        try {
            const entries = await this.kvStore.getEntries(this.sessionId);
            if (entries && entries.length > 0) {
                // 合并远程状态
                const remoteState = JSON.parse(entries[0].value.value as string);
                this.mergeState(remoteState);
                console.info('[BidirectionalMigrateAbility] State synced from distributed store');
            }
        } catch (error) {
            console.error('[BidirectionalMigrateAbility] Failed to sync from distributed store:', error);
        }
    }

    // 保存状态到分布式存储
    async saveToDistributedStore(): Promise<void> {
        if (!this.kvStore || !this.sessionId) return;
        
        try {
            const stateJson = JSON.stringify({
                ...this.appState,
                lastSyncTime: Date.now()
            });
            
            await this.kvStore.put(this.sessionId, {
                key: this.sessionId,
                value: stateJson
            });
            
            console.info('[BidirectionalMigrateAbility] State saved to distributed store');
        } catch (error) {
            console.error('[BidirectionalMigrateAbility] Failed to save to distributed store:', error);
        }
    }

    // 处理远程数据变化
    private handleRemoteDataChange(data: distributedData.ChangeNotification): void {
        if (!data.updateRecords || data.updateRecords.length === 0) return;
        
        const remoteState = JSON.parse(data.updateRecords[0].value.value as string);
        
        // 冲突解决：使用时间戳判断最新状态
        if (remoteState.lastSyncTime > this.appState.lastSyncTime) {
            this.mergeState(remoteState);
            console.info('[BidirectionalMigrateAbility] Merged remote state');
        }
    }

    // 合并状态
    private mergeState(remoteState: AppState): void {
        // 简单的合并策略：远程优先
        this.appState = {
            ...this.appState,
            ...remoteState
        };
        
        // 通知UI更新
        this.notifyStateUpdate();
    }

    // 生成会话ID
    private generateSessionId(): string {
        return `session_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
    }

    // 通知UI状态更新（需要配合ArkUI的状态管理）
    private notifyStateUpdate(): void {
        // 发送事件通知UI刷新
        console.info('[BidirectionalMigrateAbility] Notifying UI update');
    }
}

// 应用状态接口
interface AppState {
    currentPage: string;
    scrollOffset: number;
    selectedItems: string[];
    lastSyncTime: number;
}

这个示例引入了分布式键值存储，用于在迁移后持续同步状态。这样即使在目标设备上修改了状态，源设备也能感知到，实现真正的双向同步。

踩坑与注意事项

坑一：序列化数据过大导致迁移失败

问题描述：当Want.parameters中的数据超过一定大小（通常是几MB），迁移会失败，但错误信息不够明确。

解决方案：

// 错误示例：直接序列化大对象
onSaveData(want: Want): boolean {
    want.parameters = {
        'largeData': this.largeDataObject  // 可能很大
    };
    return true;
}

// 正确示例：分片存储或使用分布式数据
onSaveData(want: Want): boolean {
    // 大数据存到分布式存储
    await this.saveLargeDataToDistributedStore(this.largeDataObject);
    
    // Want中只存引用ID
    want.parameters = {
        'dataRefId': this.dataRefId
    };
    return true;
}

坑二：迁移后生命周期回调顺序问题

问题描述：迁移启动时，onCreate和onRestoreData的调用顺序可能与预期不同，导致状态恢复时机错误。

关键点：

• onCreate先于onRestoreData调用
• 在onCreate中判断launchReason，但不要在这里恢复UI状态
• UI状态的恢复应该在onWindowStageCreate之后

onCreate(want: Want, launchParam: AbilityConstant.LaunchParam): void {
    if (launchParam.launchReason === AbilityConstant.LaunchReason.CONTINUATION) {
        // 标记为迁移启动，但不立即恢复UI
        this.isMigrationLaunch = true;
        this.cachedWant = want;
    }
}

onWindowStageCreate(windowStage: window.WindowStage): void {
    // 在这里恢复UI状态
    if (this.isMigrationLaunch && this.cachedWant) {
        this.restoreUIState(this.cachedWant);
    }
}

坑三：设备断线后的处理

问题描述：迁移过程中设备突然断线，应用可能卡在中间状态。

解决方案：添加超时和重试机制

async migrateWithRetry(targetDeviceId: string, maxRetries: number = 3): Promise<boolean> {
    for (let i = 0; i < maxRetries; i++) {
        try {
            const success = await this.migrateWithTimeout(targetDeviceId, 10000);
            if (success) return true;
            
            console.warn(`[Migrate] Attempt ${i + 1} failed, retrying...`);
            await this.sleep(2000); // 等待2秒后重试
        } catch (error) {
            console.error(`[Migrate] Attempt ${i + 1} error:`, error);
        }
    }
    
    return false;
}

private sleep(ms: number): Promise<void> {
    return new Promise(resolve => setTimeout(resolve, ms));
}

坑四：跨设备权限不一致

问题描述：源设备有某些权限（如相机权限），但目标设备没有，迁移后功能异常。

解决方案：迁移前检查目标设备能力

async checkTargetDeviceCapabilities(deviceId: string): Promise<boolean> {
    // 检查目标设备是否支持所需能力
    const requiredCapabilities = ['camera', 'microphone'];
    
    for (const capability of requiredCapabilities) {
        const hasCapability = await this.queryDeviceCapability(deviceId, capability);
        if (!hasCapability) {
            console.warn(`[Migrate] Target device lacks capability: ${capability}`);
            return false;
        }
    }
    
    return true;
}

HarmonyOS 6适配

API变更

HarmonyOS 6对分布式任务调度API进行了若干调整：

1. continueAbility方法签名变更

// HarmonyOS 5.0
continueAbility(deviceId: string): Promise<void>;

// HarmonyOS 6.0 - 增加了更多配置选项
continueAbility(options: AbilityConstant.ContinuationOptions): Promise<void>;

// 适配代码
function migrateAbility(context: common.UIAbilityContext, deviceId: string): Promise<void> {
    // 兼容新旧API
    if (isHarmonyOS6()) {
        return context.continueAbility({
            deviceId: deviceId,
            migrationType: AbilityConstant.MigrationType.CROSS_DEVICE,
            reversible: true  // 支持回迁
        });
    } else {
        return (context as any).continueAbility(deviceId);
    }
}

2. 新增迁移类型枚举

// HarmonyOS 6.0新增
export enum MigrationType {
    CROSS_DEVICE = 0,      // 跨设备迁移
    CROSS_APP = 1,         // 跨应用迁移（新特性）
    CROSS_USER = 2         // 跨用户迁移（新特性）
}

3. 迁移结果回调增强

// HarmonyOS 6.0新增详细错误码
export enum ContinuationErrorCode {
    SUCCESS = 0,
    DEVICE_OFFLINE = 1001,        // 目标设备离线
    NETWORK_ERROR = 1002,         // 网络错误
    DATA_TOO_LARGE = 1003,        // 数据过大
    PERMISSION_DENIED = 1004,     // 权限拒绝
    TARGET_REFUSED = 1005,        // 目标设备拒绝
    VERSION_MISMATCH = 1006       // 版本不匹配
}

// 适配处理
onContinuationDone(result: number): void {
    const errorMap = {
        [ContinuationErrorCode.DEVICE_OFFLINE]: '目标设备已离线',
        [ContinuationErrorCode.NETWORK_ERROR]: '网络连接失败',
        [ContinuationErrorCode.DATA_TOO_LARGE]: '迁移数据过大，请精简后重试',
        [ContinuationErrorCode.PERMISSION_DENIED]: '缺少必要权限',
        [ContinuationErrorCode.TARGET_REFUSED]: '目标设备拒绝迁移',
        [ContinuationErrorCode.VERSION_MISMATCH]: '设备版本不兼容'
    };
    
    if (result !== ContinuationErrorCode.SUCCESS) {
        const errorMsg = errorMap[result] || '未知错误';
        this.showErrorMessage(errorMsg);
    }
}

行为变更

1. 迁移超时时间调整

HarmonyOS 6将默认迁移超时从5秒调整为10秒，但建议开发者仍显式设置超时：

// 显式设置超时（推荐）
const migrateOptions: AbilityConstant.ContinuationOptions = {
    deviceId: targetDeviceId,
    timeout: 15000  // 自定义15秒超时
};

2. 自动重连机制

HarmonyOS 6新增了自动重连机制，迁移失败时会自动尝试重连设备：

// 配置自动重连
const migrateOptions: AbilityConstant.ContinuationOptions = {
    deviceId: targetDeviceId,
    autoReconnect: true,        // 启用自动重连
    reconnectAttempts: 3,       // 最多重试3次
    reconnectInterval: 2000     // 重试间隔2秒
};

3. 增强的安全校验

HarmonyOS 6对跨设备迁移增加了更严格的安全校验：

• 设备认证级别检查
• 应用签名验证
• 数据加密传输

// 检查设备安全等级
async function checkDeviceSecurityLevel(deviceId: string): Promise<boolean> {
    const deviceInfo = await getDeviceInfo(deviceId);
    
    // HarmonyOS 6要求目标设备安全等级不低于源设备
    const currentLevel = getSecurityLevel();
    const targetLevel = deviceInfo.securityLevel;
    
    if (targetLevel < currentLevel) {
        console.warn('Target device security level too low');
        return false;
    }
    
    return true;
}

总结

分布式任务调度是HarmonyOS分布式能力的核心基础，Ability跨设备迁移让应用真正实现了"一次开发，多端部署，无缝流转"。本文从原理到实战，详细介绍了迁移的实现方式、常见问题和解决方案。

核心要点回顾：

1. 理解迁移三阶段：冻结→传输→恢复，每个阶段都有对应的回调方法
2. 合理设计状态数据：既要完整保存必要信息，又要控制数据大小
3. 处理边界情况：设备离线、权限差异、数据冲突等都需要妥善处理
4. 适配新版本：HarmonyOS 6提供了更丰富的API和更强的安全机制

最佳实践建议：

• 迁移前检查目标设备能力和权限
• 使用分布式数据存储管理大体积数据
• 实现合理的超时和重试机制
• 做好错误处理和用户提示
• 充分测试各种异常场景

分布式任务调度打开了多设备协同的大门，但这只是开始。接下来的文章中，我们将深入探讨Ability迁移的更多实战技巧、Continuation开发、分布式窗口等更高级的能力。掌握了这些，你就能真正构建出"超级终端"级别的应用体验。