鸿蒙 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();}
}