HarmonyOS 鸿蒙Next开发者技术支持 - 线程间通信技术解决方案

一、问题说明

在HarmonyOS应用开发中，线程间通信是多线程并发场景下的核心需求。主要问题包括：

1. ArkTS线程间数据传输限制：Actor内存隔离模型下，线程间无法直接共享内存，必须通过序列化/反序列化传递数据，单次传输限制16MB。
2. Native子线程无法回调ArkTS函数：Native侧通过std::thread或pthread创建的子线程无法直接调用只能在UI主线程执行的ArkTS函数。
3. 通信效率低下：JS对象序列化/反序列化带来性能开销，大量数据传递时影响应用响应性能。
4. 线程数量限制：Worker线程数量限制为64个，过多的线程创建会导致内存和调度开销增大。

二、原因分析

根据HarmonyOS线程模型和Actor并发模型，问题根源如下：

2.1 内存隔离机制

• ArkTS线程模型：每个线程拥有独立的ArkTS引擎实例和内存空间，基于Actor模型实现内存隔离。
• 线程间通信限制：必须通过消息传递机制进行通信，不能直接访问对方内存空间。

2.2 ArkTS线程类型限制

• UI主线程：负责UI绘制、事件分发、ArkTS引擎管理。
• TaskPool线程（推荐）：自动管理生命周期，支持优先级调度和负载均衡。
• Worker线程：开发者自行管理生命周期，支持耗时任务和线程间通信。

2.3 序列化约束

• 对象类型限制：不支持@State、@Prop、@Link、@Observed等装饰器修饰的复杂类型。
• 序列化方式：采用Structured Clone算法，通过深拷贝实现对象传输。

2.4 Native线程约束

• 线程安全：主线程中的napi_env、napi_value、napi_ref不能在子线程直接使用。
• 回调限制：ArkTS函数只能在UI主线程调用。

三、解决思路

基于不同通信场景，提供分层解决方案：

3.1 ArkTS线程间通信

1. 基本通信：使用TaskPool和Worker的标准通信接口。
2. 高效通信：使用@Sendable装饰器实现引用传递，减少序列化开销。
3. 内存共享：使用SharedArrayBuffer配合Atomics原子操作。
4. 事件通信：使用Emitter实现进程内线程间异步事件通信。

3.2 Native与ArkTS线程通信

1. 线程安全函数：使用Node-API的napi_create_threadsafe_function机制。
2. libuv异步通信：使用uv_async_send方法（备选方案）。
3. Native回调：Native侧执行耗时任务后，通过安全机制回调ArkTS函数。

3.3 数据传递优化

1. 分类传输：根据数据类型选择合适的通信对象。
2. 减少数据量：控制单次传输数据大小。
3. 异步锁管理：多线程访问共享数据时使用异步锁保证安全。

四、解决方案

方案一：ArkTS线程间高效通信

4.1.1 TaskPool任务与宿主线程通信

/**
 * SharedData.ets
 * 并发任务处理工具
 * 功能：
 * 1. 定义并发处理函数 processData
 * 2. 提供任务管理和进度监控
 * 3. 支持并发处理大量数据
 */
import { taskpool } from '@kit.ArkTS';
import { hilog } from '@kit.PerformanceAnalysisKit';

/**
 * Params 接口
 * 定义处理数据的参数
 */
interface Params{
  data: number[];    // 要处理的数据数组
  threshold: number; // 阈值，用于筛选数据
}

/**
 * ProgressData 接口
 * 定义进度数据的结构
 */
interface ProgressData{
  type: string;     // 数据类型，用于标识进度信息
  current: number;  // 当前处理的索引
  total: number;    // 总数据量
  processed: number; // 已处理并符合条件的数据量
}
/**
 * processData 函数
 * 并发处理数据，筛选出大于阈值的数据
 * @param params 处理参数，包含数据数组和阈值
 * @returns 筛选后的结果数组
 */
@Concurrent
export function processData(params: Params): number[] {
  const results: number[] = [];

  // 处理数据并实时发送进度
  for (let i = 0; i < params.data.length; i++) {
    // 筛选出大于阈值的数据
    if (params.data[i] > params.threshold) {
      results.push(params.data[i]);
    }

    // 每处理100个数据发送一次进度
    if (i % 100 === 0) {
      taskpool.Task.sendData({
        type: 'progress',
        current: i + 1,
        total: params.data.length,
        processed: results.length
      });
    }
  }

  return results;
}



/**
 * MainPage 类
 * 示例类，展示如何使用并发任务处理
 */
class MainPage {
  private currentTask: taskpool.Task | null = null; // 当前正在执行的任务

  /**
   * 启动任务
   * 1. 生成测试数据
   * 2. 创建任务参数
   * 3. 创建并配置任务
   * 4. 执行任务并处理结果
   */
  async startProcessing(): Promise<void> {
    // 生成测试数据
    const data = this.generateTestData(10000);
    
    // 创建任务参数
    const params: Params = {
      data: data,
      threshold: 50
    };
    
    // 创建任务
    const task = new taskpool.Task(processData, params);

    // 设置数据接收回调，用于处理进度信息
    task.onReceiveData((progressData: ProgressData) => {
      hilog.info(0x0000, 'MainPage', `Progress: ${progressData.current}/${progressData.total}, Found: ${progressData.processed}`);
      this.updateProgress(progressData.current, progressData.total);
    });

    // 保存当前任务
    this.currentTask = task;

    try {
      // 执行任务并等待结果
      const result: number[] = await taskpool.execute(task) as number[];
      hilog.info(0x0000, 'MainPage', `Processing completed, found ${result.length} items`);
      this.displayResults(result);
    } catch (error) {
      hilog.error(0x0000, 'MainPage', `Task failed: ${error.message}`);
    }
  }

  /**
   * 生成测试数据
   * @param count 数据量
   * @returns 随机数据数组
   */
  private generateTestData(count: number): number[] {
    const data: number[] = [];
    for (let i = 0; i < count; i++) {
      data.push(Math.floor(Math.random() * 100));
    }
    return data;
  }

  /**
   * 更新进度显示（示例）
   * @param current 当前处理的索引
   * @param total 总数据量
   */
  private updateProgress(current: number, total: number): void {
    // 更新UI进度条
  }

  /**
   * 显示结果（示例）
   * @param results 处理后的结果数组
   */
  private displayResults(results: number[]): void {
    // 处理结果显示
  }
}

4.1.2 Worker线程即时通信

/**
 * WorkerData.ets
 * Worker线程处理工具
 * 功能：
 * 1. 定义Worker线程消息处理逻辑
 * 2. 提供数据处理功能
 * 3. 支持取消处理操作
 */
import { worker, ThreadWorkerGlobalScope, ErrorEvent } from '@kit.ArkTS';
import { hilog } from '@kit.PerformanceAnalysisKit';

const workerPort: ThreadWorkerGlobalScope = worker.workerPort;
let shouldCancel = false;

// 数据负载类型定义
interface DataPayload {
  items: number[];
}

// 消息类型定义
interface WorkerMessage {
  command: string;
  payload?: DataPayload;
}

// 批量结果消息类型定义
interface BatchResultMessage {
  type: string;
  batchId: number;
  processedCount: number;
  completed: boolean;
}

// 取消消息类型定义
interface CancelMessage {
  type: string;
  processedCount: number;
}

// 结果消息类型定义
interface ResultMessage {
  type: string;
  processedCount: number;
}

// 错误消息类型定义
interface ErrorMessage {
  type: string;
  error: string;
}

/**
 * 消息事件类型定义
 */
interface WorkerMessageEvent {
  data: WorkerMessage;
}

/**
 * 结果消息事件类型定义
 */
interface ResultMessageEvent {
  data: BatchResultMessage | CancelMessage | ErrorMessage;
}



/**
 * 处理Worker线程消息
 */
workerPort.onmessage = (event: WorkerMessageEvent): void => {
  const message: WorkerMessage = event.data;

  switch (message.command) {
    case 'process':
      if (message.payload) {
        processData(message.payload);
      } else {
        hilog.error(0x0000, 'DataProcessor', 'Missing payload for process command');
      }
      break;
    case 'cancel':
      shouldCancel = true;
      break;
    default:
      hilog.warn(0x0000, 'DataProcessor', `Unknown command: ${message.command}`);
  }
};

/**
 * 数据处理函数
 * @param data 要处理的数据
 */
function processData(data: DataPayload): void {
  let count = 0;
  const batchSize = 100;

  for (let i = 0; i < data.items.length; i++) {
    // 模拟数据处理
    const processed = data.items[i] * 2;
    count++;

    // 分批发送结果
    if (count % batchSize === 0 || i === data.items.length - 1) {
      const result: BatchResultMessage = {
        type: 'batch_result',
        batchId: Math.floor(i / batchSize),
        processedCount: count,
        completed: i === data.items.length - 1
      };
      workerPort.postMessage(result);
    }

    // 检查取消标志
    if (shouldCancel) {
      const cancelledMessage: CancelMessage = {
        type: 'cancelled',
        processedCount: count
      };
      workerPort.postMessage(cancelledMessage);
      shouldCancel = false; // 重置取消标志
      return;
    }
  }
}



/**
 * MainPage 类
 * 示例类，展示如何使用Worker线程处理数据
 */
class MainPage {
  private workerInstance: worker.ThreadWorker | null = null;

  /**
   * 启动Worker处理
   * 1. 创建Worker实例
   * 2. 设置消息接收回调
   * 3. 设置错误处理回调
   * 4. 发送处理命令
   */
  async startWorkerProcessing(): Promise<void> {
    this.workerInstance = new worker.ThreadWorker('entry/ets/workers/DataProcessor.ets');

    // 设置消息接收
    this.workerInstance.onmessage = (event: ResultMessageEvent): void => {
      const data: BatchResultMessage | CancelMessage | ErrorMessage = event.data;

      switch (data.type) {
        case 'batch_result':
          const batchResult = data as BatchResultMessage;
          this.handleBatchResult(batchResult.batchId, batchResult.processedCount, batchResult.completed);
          break;
        case 'cancelled':
          const cancelMessage = data as CancelMessage;
          this.handleCancellation(cancelMessage.processedCount);
          break;
        case 'error':
          const errorMessage = data as ErrorMessage;
          this.handleError(errorMessage.error);
          break;
      }
    };

    // 设置错误处理
    this.workerInstance.onerror = (err: ErrorEvent): void => {
      hilog.error(0x0000, 'MainPage', `Worker error: ${err.message}`);
    };

    // 发送处理命令
    const data = this.generateLargeDataSet();
    const message: WorkerMessage = {
      command: 'process',
      payload: { items: data }
    };
    this.workerInstance.postMessage(message);
  }

  /**
   * 生成测试数据
   * @returns 测试数据数组
   */
  private generateLargeDataSet(): number[] {
    const data: number[] = [];
    for (let i = 0; i < 10000; i++) {
      data.push(Math.floor(Math.random() * 1000));
    }
    return data;
  }

  /**
   * 处理批量结果
   * @param batchId 批次ID
   * @param processedCount 已处理数量
   * @param completed 是否完成
   */
  private handleBatchResult(batchId: number, processedCount: number, completed: boolean): void {
    hilog.info(0x0000, 'MainPage', `Batch ${batchId} completed, processed ${processedCount} items${completed ? ', all done!' : ''}`);
  }

  /**
   * 处理取消操作
   * @param processedCount 已处理数量
   */
  private handleCancellation(processedCount: number): void {
    hilog.info(0x0000, 'MainPage', `Processing cancelled, processed ${processedCount} items`);
  }

  /**
   * 处理错误
   * @param error 错误信息
   */
  private handleError(error: string): void {
    hilog.error(0x0000, 'MainPage', `Processing error: ${error}`);
  }

  /**
   * 取消处理
   */
  cancelProcessing(): void {
    if (this.workerInstance) {
      const message: WorkerMessage = {
        command: 'cancel'
      };
      this.workerInstance.postMessage(message);
    }
  }

  /**
   * 清理Worker
   */
  cleanupWorker(): void {
    if (this.workerInstance) {
      this.workerInstance.terminate();
      this.workerInstance = null;
    }
  }
}

方案二：Native侧子线程与UI主线程通信

4.2.1 基于线程安全函数机制（推荐方案）

// Native侧代码 - ThreadSafeCommunicator.cpp
#include <napi/native_api.h>
#include <hilog/log.h>
#include <thread>
#include <vector>

  // 回调上下文结构
  struct ThreadCallbackContext {
  napi_env env;
  napi_ref jsCallbackRef;
  std::vector<int> processedData;
  int requestId;
};

class ThreadSafeCommunicator {
  private:
    napi_threadsafe_function tsFunction_;
  bool initialized_;

  public:
    ThreadSafeCommunicator() : initialized_(false) {}

  ~ThreadSafeCommunicator() {
  cleanup();
}

  // 初始化线程安全函数
  napi_status initialize(napi_env env, napi_value jsCallback) {
  if (initialized_) {
    return napi_ok;
  }

  // 创建线程安全函数
  napi_status status = napi_create_threadsafe_function(
  env,
  jsCallback,
  nullptr,
  napi_value("ThreadSafeCallback"),
  0,  // 无限队列
  1,  // 初始线程数
  nullptr,
  nullptr,
  nullptr,
  [](napi_env env, napi_value js_callback, void* context, void* data) {
  // 此回调在主线程执行
  ThreadCallbackContext* ctx = static_cast<ThreadCallbackContext*>(data);
  if (env && js_callback && ctx) {
  // 准备回调参数
  napi_value resultArray;
  napi_create_array_with_length(env, ctx->processedData.size(), &resultArray);

  for (size_t i = 0; i < ctx->processedData.size(); i++) {
  napi_value element;
  napi_create_int32(env, ctx->processedData[i], &element);
  napi_set_element(env, resultArray, i, element);
}

napi_value requestId;
napi_create_int32(env, ctx->requestId, &requestId);

// 调用JavaScript回调
napi_value argv[2];
argv[0] = resultArray;
argv[1] = requestId;

napi_value global;
napi_get_global(env, &global);
napi_call_function(env, global, js_callback, 2, argv, nullptr);
}

// 清理资源
if (ctx) {
  napi_delete_reference(ctx->env, ctx->jsCallbackRef);
  delete ctx;
}
},
&tsFunction_
);

if (status == napi_ok) {
  initialized_ = true;
  OH_LOG_INFO(LOG_APP, "Thread safe function initialized.");
}

return status;
}

// 从子线程调用ArkTS函数
void callFromWorkerThread(const std::vector<int>& data, int requestId, napi_env env, napi_ref callbackRef) {
  if (!initialized_) {
    OH_LOG_ERROR(LOG_APP, "Thread safe function not initialized!");
    return;
  }

  // 准备上下文数据
  auto context = new ThreadCallbackContext();
  context->env = env;
  context->jsCallbackRef = callbackRef;
  context->processedData = data;
  context->requestId = requestId;

  // 获取线程安全函数
  napi_acquire_threadsafe_function(tsFunction_, napi_tsfn_blocking);

  // 调用线程安全函数
  napi_call_threadsafe_function(tsFunction_, context, napi_tsfn_nonblocking);

  // 释放线程安全函数
  napi_release_threadsafe_function(tsFunction_, napi_tsfn_release);
}

// 清理资源
void cleanup() {
  if (initialized_ && tsFunction_) {
    napi_release_threadsafe_function(tsFunction_, napi_tsfn_release);
    initialized_ = false;
  }
}
};

// 导出的Native方法
static ThreadSafeCommunicator gCommunicator;

// 异步处理函数（在子线程执行）
static void processInWorkerThread(int requestId) {
  OH_LOG_INFO(LOG_APP, "Worker thread started for request %{public}d", requestId);

  // 模拟耗时处理
  std::vector<int> results;
  for (int i = 0; i < 100; i++) {
    results.push_back(i * requestId);
    std::this_thread::sleep_for(std::chrono::milliseconds(10));
  }

  OH_LOG_INFO(LOG_APP, "Worker thread completed for request %{public}d", requestId);
}