从单机应用到分布式调度：基于 HarmonyOS 构建车-空协同任务引擎

@VON

发布于 2025-12-21 13:11:23

1050

个人首页： VON
鸿蒙系列专栏：鸿蒙开发小型案例总结
综合案例：鸿蒙综合案例开发
鸿蒙6.0：从0开始的开源鸿蒙6.0.0
鸿蒙5.0：鸿蒙5.0零基础入门到项目实战
Electron适配开源鸿蒙专栏：Electron for OpenHarmony
Flutter 适配开源鸿蒙专栏：Flutter for OpenHarmony

基于 HarmonyOS 构建车-空协同任务引擎

作者：VON 技术栈：Vue 3 + TypeScript + Vite + DevUI 参考文档

MateChat：https://gitcode.com/DevCloudFE/MateChat
MateChat官网：https://matechat.gitcode.com
DevUI官网：https://devui.design/home

起点：一个“只能跑在车机上”的巡检 App

2024 年初，我开发了一个道路巡检 App，功能简单：车载终端启动后，调用本地摄像头识别路面病害，并上传结果到云端。一切逻辑都运行在单一设备上——典型的 单体式应用架构。

问题很快暴露：

无法覆盖盲区（如弯道、遮挡路段）
识别精度受限于车载算力
人工上报效率低

团队提出新方案：引入无人机作为移动感知节点，由车机发起任务，无人机执行巡检，结果回传融合分析。这要求系统具备 跨设备任务分发、状态同步、容错恢复 能力——远超传统 App 范畴。

于是，我开始构建一个 轻量级分布式任务调度引擎，完全基于 HarmonyOS 原生能力实现。

架构设计：三层协同模型

我们采用 “调度层 - 执行层 - 数据层” 三层架构：

[ 车载终端 ] ——(DSoftBus)——> [ 无人机 ]
     |                               |
     v                               v
[ 任务调度器 ] <——(DDM)——> [ 分布式事件存储 ]

调度层：运行在车机，负责任务创建、分配、监控
执行层：运行在无人机，接收指令、执行 AI 推理、返回结果
数据层：基于 DDM 的分布式 KV 存储，保证状态一致

核心模块一：任务定义与序列化

首先，定义统一的任务结构，确保跨设备可解析：

// TaskModel.ets
export interface RoadInspectionTask {
  taskId: string;
  type: 'aerial_scan' | 'ground_check';
  targetArea: {
    centerLat: number;
    centerLng: number;
    radius: number; // 米
  };
  priority: 'high' | 'normal';
  createdAt: number;
  status: 'pending' | 'running' | 'completed' | 'failed';
  executorDeviceId?: string; // 执行设备 networkId
}

为支持跨进程/跨设备传输，我们使用 JSON 序列化 + 校验：

// TaskSerializer.ets
export class TaskSerializer {
  static serialize(task: RoadInspectionTask): string {
    return JSON.stringify(task);
  }

  static deserialize(jsonStr: string): RoadInspectionTask | null {
    try {
      const obj = JSON.parse(jsonStr);
      if (!obj.taskId || !obj.type) return null;
      return obj as RoadInspectionTask;
    } catch {
      return null;
    }
  }
}

核心模块二：分布式任务分发（基于 DSoftBus）

车机通过 deviceManager 发现可用无人机，并分发任务：

// TaskDispatcher.ets
import deviceManager from '@ohos.distributedHardware.deviceManager';
import rpc from '@ohos.rpc';

class TaskDispatcher {
  private droneNetworkId: string | null = null;

  async discoverAndBindDrone(): Promise<boolean> {
    const devices = await deviceManager.getAvailableDeviceList();
    const drone = devices.find(d => d.deviceName?.includes('Drone'));
    if (drone) {
      this.droneNetworkId = drone.networkId;
      return true;
    }
    return false;
  }

  async dispatchTask(task: RoadInspectionTask): Promise<boolean> {
    if (!this.droneNetworkId) {
      console.error('No drone bound');
      return false;
    }

    try {
      // 连接远程服务
      const remoteObj = await rpc.connectRemoteObject(
        this.droneNetworkId,
        'com.example.drone.InspectionService'
      );

      // 调用远程方法
      const result = await remoteObj.sendRequest(
        1001, // 自定义 method ID
        TaskSerializer.serialize(task)
      );

      return result.resultCode === 0;
    } catch (err) {
      console.error('Dispatch failed:', err);
      return false;
    }
  }
}

核心模块三：无人机端任务执行器

在无人机端，我们实现一个 InspectionService，继承自 rpc.RemoteObject：

// InspectionService.ets (运行在无人机)
import rpc from '@ohos.rpc';
import taskRunner from '@ohos.taskRunner';

class InspectionService extends rpc.RemoteObject {
  constructor() {
    super('InspectionService');
  }

  onRemoteRequest(code: number, data: rpc.MessageSequence): boolean {
    if (code === 1001) {
      const taskJson = data.readString();
      const task = TaskSerializer.deserialize(taskJson);
      if (task) {
        // 异步执行任务（避免阻塞主线程）
        taskRunner.execute(() => {
          this.executeInspection(task);
        });
        // 回写成功响应
        const reply = new rpc.MessageSequence();
        reply.writeInt(0); // resultCode
        data.reply(reply);
        return true;
      }
    }
    return false;
  }

  private async executeInspection(task: RoadInspectionTask) {
    // 1. 控制云台对准目标区域
    await GimbalController.aimAt(task.targetArea);

    // 2. 启动摄像头 + AI 推理
    const frames = Camera.captureFrames(5);
    const risks = await AIModel.detectRisks(frames);

    // 3. 更新任务状态并持久化
    task.status = 'completed';
    task.result = risks;

    // 4. 写入分布式存储（自动同步到车机）
    await DistributedStore.put(task.taskId, task);
  }
}

核心模块四：分布式状态同步（基于 DDM）

我们在两端初始化同一个分布式 KV Store：

// DistributedStore.ets（车机 & 无人机共用）
import distributedData from '@ohos.data.distributedData';

const kvManager = distributedData.createKVManager({
  bundleName: 'com.example.roadguardian',
  storeName: 'inspection_tasks'
});

export class DistributedStore {
  static async put(key: string, value: any): Promise<void> {
    await kvManager.put(key, JSON.stringify(value));
  }

  static async get<T>(key: string): Promise<T | null> {
    const val = await kvManager.get(key);
    return val ? JSON.parse(val) : null;
  }

  static on(event: 'dataChange', callback: (key: string) => void) {
    kvManager.on(event, callback);
  }
}

车机监听状态变更，实时更新 UI：

// 车载端监听任务完成
DistributedStore.on('dataChange', async (taskId) => {
  const task = await DistributedStore.get<RoadInspectionTask>(taskId);
  if (task?.status === 'completed') {
    this.showRiskAlert(task.result);
    this.updateTaskListUI();
  }
});

容错与降级：应对现实世界的不确定性

实际部署中，我们遇到诸多挑战：

无人机突然断连
车机网络中断
多任务并发冲突

为此，我们加入：

任务心跳机制：执行端每 5 秒上报进度
超时重试队列：pending 超过 60 秒自动重发
本地缓存兜底：即使离线，任务指令也暂存本地，待网络恢复后同步

// 任务超时检测
setInterval(async () => {
  const pendingTasks = await TaskLocalCache.getPendingTasks();
  for (const task of pendingTasks) {
    if (Date.now() - task.createdAt > 60_000) {
      task.retryCount++;
      if (task.retryCount < 3) {
        dispatcher.dispatchTask(task); // 重试
      } else {
        task.status = 'failed';
        DistributedStore.put(task.taskId, task);
      }
    }
  }
}, 10_000);