当前位置：首页 > article >正文

鸿蒙 Next 实现线程之间的通信

article 2025/3/15 21:31:30

鸿蒙 Next 实现线程之间的通信

在鸿蒙 Next 开发中，线程间通信是一个常见需求，尤其是在多线程任务处理中。鸿蒙 Next 提供了多种机制来实现线程间通信，包括事件驱动的 Emitter、共享内存 SharedArrayBuffer 以及基于消息传递的 Worker 和 TaskPool。本文将详细介绍这些机制的使用方法和实现示例。

一、使用 Emitter 实现线程间通信

Emitter 是鸿蒙 Next 提供的一种事件驱动机制，用于实现线程间的通信。它支持事件的订阅、发送和取消订阅。

（一）订阅事件

在主线程或子线程中订阅事件，定义事件回调函数来处理事件。

import { emitter } from '@kit.BasicServicesKit';
import { promptAction } from '@kit.ArkUI';

const event = {
  eventId: 1,
};

const callback = (eventData: emitter.EventData) => {
  promptAction.showToast({
    message: JSON.stringify(eventData),
  });
};

emitter.on(event, callback);

（二）发送事件

在主线程或子线程中发送事件，将事件数据传递给订阅者。

import { emitter } from '@kit.BasicServicesKit';

const event = {
  eventId: 1,
  priority: emitter.EventPriority.LOW,
};

const eventData = {
  data: {
    content: 'Hello from thread',
    id: 1,
    isEmpty: false,
  },
};

emitter.emit(event, eventData);

二、使用 SharedArrayBuffer 实现跨线程共享内存

SharedArrayBuffer 是一种低级的跨线程共享内存机制，适用于需要高性能通信的场景。

（一）主线程初始化共享内存

在主线程中初始化 SharedArrayBuffer，并将其传递给子线程。

@Component
export struct LockUsage {
  sabInLock: SharedArrayBuffer = new SharedArrayBuffer(4);
  sabForLine: SharedArrayBuffer = new SharedArrayBuffer(4);

  build() {
    // 初始化子线程锁标志位和偏移位
    Atomics.store(new Int32Array(this.sabInLock), 0, 0);
    Atomics.store(new Int32Array(this.sabForLine), 0, 0);
  }
}

（二）子线程使用共享内存

在子线程中，通过 SharedArrayBuffer 进行线程间通信。

@Concurrent
async function createWriteTask(sabInLock: SharedArrayBuffer, sabForLine: SharedArrayBuffer) {
  const lock = new NonReentrantLock(sabInLock);
  const whichLineToWrite = new Int32Array(sabForLine);

  lock.lock();
  try {
    // 执行任务
    console.log('Writing to line:', whichLineToWrite[0]);
    whichLineToWrite[0] += 1; // 更新偏移量
  } finally {
    lock.unlock();
  }
}

三、使用 Worker 和 TaskPool 实现线程间通信

Worker 和 TaskPool 是基于消息传递的线程间通信机制，适用于处理复杂的异步任务。

（一）使用 Worker

Worker 是一个独立的线程，可以执行耗时任务，通过消息传递与主线程通信。

主线程代码

@Entry
@Component
struct MainWorkerComponent {
  build() {
    const worker = new Worker('worker.js');
    worker.onMessage = (message) => {
      console.info('Main thread received:', message.data);
    };
    worker.postMessage({ task: 'processData', data: [1, 2, 3] });
  }
}

Worker 代码（worker.js）

onmessage = function(event) {
  const { task, data } = event.data;
  if (task === 'processData') {
    const result = data.map(item => item * 2);
    postMessage({ result });
  }
};

（二）使用 TaskPool

TaskPool 是一个线程池，用于管理多个任务的执行。

class ThreadPool {
  constructor(public maxThreads: number) {
    this.pool = [];
  }

  runTask(task) {
    if (this.pool.length < this.maxThreads) {
      const worker = new Worker('worker.js');
      this.pool.push(worker);
      worker.onMessage = (message) => {
        console.info('Task completed:', message.data);
        this.releaseWorker(worker);
      };
      worker.postMessage({ task });
    } else {
      console.info('All threads are busy, retrying...');
      setTimeout(() => this.runTask(task), 1000);
    }
  }

  releaseWorker(worker) {
    this.pool = this.pool.filter(w => w !== worker);
    worker.terminate();
  }
}