资讯详情

深入了解 · 专业解读

多线程并发模型：Actor模型在ArkTS中的应用（172）

发布时间：2026/7/15 22:43:10
多线程并发模型：Actor模型在ArkTS中的应用（172） 在 ArkTS 中多线程并发编程彻底摒弃了传统基于“共享内存”的模型全面采用了基于消息传递的Actor 并发模型。这种设计从底层架构上避免了复杂的锁竞争与数据竞争问题。以下是 Actor 模型在 ArkTS 中的核心原理、通信机制及两大并发能力的应用一、 核心原理内存隔离与消息传递在 Actor 模型下每一个线程Actor都是一个独立的计算单元拥有自己独占的内存空间、虚拟机实例和事件轮询机制EventLoop。内存隔离线程之间绝对不直接访问对方的内存空间从根本上消除了内存抢占和脏读脏写问题开发者无需处理复杂的锁机制。消息通信Actor 之间通过发送和接收消息来触发彼此的行为。这种异步的消息传递机制保证了高并发下的系统稳定性。二、 线程间通信机制数据传输对象由于内存隔离跨线程的数据交换必须经过序列化或特定的共享机制。ArkTS 支持以下四种数据传输对象普通对象采用标准的结构化克隆算法Structured Clone进行深拷贝。支持基础类型、Date、Array、Map、Set 等但对象越大序列化耗时越长。可转移对象Transferable如ArrayBuffer。采用地址转移机制不进行内容拷贝。转移后原线程中的对象将不可用所有权完全移交。可共享对象SharedArrayBuffer支持多线程间真正的内存共享。由于多线程可能同时修改必须通过Atomics原子操作来保证数据的同步性与安全性。Native 绑定对象如Context和RemoteObject支持序列化传输用于跨线程访问底层系统功能或实现跨进程通信。三、 ArkTS 的两大并发能力基于 Actor 模型ArkTS 提供了TaskPool和Worker两种并发 API适用于不同的业务场景1. TaskPool任务池特点由系统统一管理和调度线程开发者无需手动创建和销毁线程有效避免了线程泛滥。适用场景适合执行相互独立、执行周期短、无上下文依赖的耗时任务。例如JSON 解析、图片/视频编解码、压缩/解压缩、数据库操作等。开发规范任务函数必须使用Concurrent装饰器修饰且只能依赖模块化导入或参数传入不能依赖外部上下文。import { taskpool } from kit.ArkTS; // 【开发规范】任务函数必须是模块顶层函数且使用 Concurrent 装饰器 Concurrent function heavyCalculation(max: number): number { let count 0; for (let i 2; i max; i) { let isPrime true; for (let j 2; j * j i; j) { if (i % j 0) { isPrime false; break; } } if (isPrime) count; } return count; } Entry Component struct TaskPoolDemo { State resultText: string 点击按钮开始计算; build() { Column({ space: 20 }) { Text(this.resultText).fontSize(18) Button(计算 1~200000 的质数) .onClick(async () { this.resultText 计算中...; // 创建任务并派发至 TaskPool const task new taskpool.Task(heavyCalculation, 200000); try { // 异步等待结果不阻塞主线程 const count await taskpool.execute(task) as number; this.resultText 计算完成共有 ${count} 个质数; } catch (e) { this.resultText 执行失败: ${JSON.stringify(e)}; } }) } .width(100%).height(100%).justifyContent(FlexAlign.Center) } }2. Worker工作线程特点开发者手动创建的独立常驻线程拥有完整的运行环境。线程不会主动销毁需要开发者自行管理生命周期。适用场景适合执行有关联性、需要长时间运行大于3分钟或依赖特定句柄/类对象调度的常驻任务。例如游戏业务的主逻辑线程、产线硬件阻塞式压测等。场景创建一个后台常驻线程主线程发送指令启动任务Worker 线程周期性地向主线程回传状态数据。第一步创建 Worker 线程文件 (workers/BackgroundWorker.ets)import { worker, ThreadWorkerGlobalScope, MessageEvents } from kit.ArkTS; const workerPort: ThreadWorkerGlobalScope worker.workerPort; let isRunning: boolean false; // 接收主线程的消息 workerPort.onmessage (e: MessageEvents) { const msg e.data as string; if (msg start) { isRunning true; performTask(); } else if (msg stop) { isRunning false; workerPort.postMessage(Worker 已停止任务); } }; // 模拟长驻任务并持续回传数据 function performTask() { if (isRunning) { workerPort.postMessage(当前时间戳: ${Date.now()}); setTimeout(performTask, 1000); // 每秒执行一次 } }第二步在主线程中管理 Worker 生命周期 (Index.ets)import { worker, MessageEvents } from kit.ArkTS; Entry Component struct WorkerDemo { State statusText: string Worker 未启动; private workerInstance: worker.ThreadWorker | null null; build() { Column({ space: 20 }) { Text(this.statusText).fontSize(18) Button(启动 Worker) .onClick(() { if (!this.workerInstance) { // 创建 Worker 实例 this.workerInstance new worker.ThreadWorker(entry/ets/workers/BackgroundWorker.ets); // 监听 Worker 回传的消息 this.workerInstance.onmessage (e: MessageEvents) { this.statusText e.data as string; }; } this.workerInstance.postMessage(start); }) Button(停止 Worker) .onClick(() { this.workerInstance?.postMessage(stop); }) } .width(100%).height(100%).justifyContent(FlexAlign.Center) } // 页面销毁时务必终止 Worker防止内存泄漏 aboutToDisappear() { this.workerInstance?.terminate(); this.workerInstance null; } }3. TaskPool 中的持续通信sendData / onReceiveData场景如果不想使用 Worker但 TaskPool 任务在执行过程中需要向主线程发送中间进度数据。import { taskpool } from kit.ArkTS; Concurrent function taskWithProgress(total: number): string { for (let i 0; i total; i) { // 通过 sendData 向主线程发送中间状态 taskpool.Task.sendData(i); } return 任务全部完成; } // 在主线程中监听进度 const task new taskpool.Task(taskWithProgress, 100); task.onReceiveData((progress: number) { console.info(当前进度: ${progress}%); }); const finalResult await taskpool.execute(task) as string; console.info(finalResult);4. 进阶通信机制ArrayBuffer 的零拷贝转移与共享在跨线程传输大量二进制数据如图片像素、音频流时默认的深拷贝Structured Clone会带来极大的性能损耗。ArkTS 提供了零拷贝的传输方案。实战代码使用 Transfer 机制转移 ArrayBuffer 所有权import { taskpool } from kit.ArkTS; Concurrent function processImageBuffer(buffer: ArrayBuffer): ArrayBuffer { // 在子线程中直接处理二进制数据无需深拷贝 const view new Uint8Array(buffer); view[0] 255; // 修改数据 return buffer; } async function transferDemo() { const buffer new ArrayBuffer(1024 * 1024); // 1MB 数据 // 将 buffer 作为 Transfer 对象传递 const task new taskpool.Task(processImageBuffer, buffer); // 关键设置转移列表传递后主线程中的 buffer 将被置为不可用Neutered task.setTransferList([buffer]); await taskpool.execute(task); // console.log(buffer.byteLength); // 此时访问会报错所有权已转移 }5. 进阶通信机制SharedArrayBuffer 与 Atomics 无锁并发当多个线程需要高频读写同一块内存如共享状态计数器、自旋锁时使用 SharedArrayBuffer 配合 Atomics 实现真正的内存共享。实战代码子线程安全的原子计数器import { taskpool } from kit.ArkTS; Concurrent function atomicCounter(sharedBuffer: SharedArrayBuffer): void { const int32 new Int32Array(sharedBuffer); for (let i 0; i 1000; i) { // 原子加法绝对安全不会发生数据竞争 Atomics.add(int32, 0, 1); } } async function sharedMemoryDemo() { const sab new SharedArrayBuffer(4); const int32 new Int32Array(sab); // 并发派发 10 个任务 const group new taskpool.TaskGroup(); for (let i 0; i 10; i) { group.addTask(new taskpool.Task(atomicCounter, sab)); } await taskpool.execute(group); console.info(最终计数: ${Atomics.load(int32, 0)}); // 预期结果严格为 10000 }6. 任务编排TaskGroup 与任务依赖管理在实际业务中任务往往不是孤立的。TaskPool 提供了强大的任务编排能力支持任务组批量执行和树状依赖。实战代码并行加载资源全部完成后触发渲染import { taskpool } from kit.ArkTS; Concurrent function loadTexture(id: number): string { return Texture_${id}_Loaded; } Concurrent function loadModel(id: number): string { return Model_${id}_Loaded; } async function orchestrationDemo() { const group new taskpool.TaskGroup(); // 添加多个并行任务 group.addTask(new taskpool.Task(loadTexture, 1)); group.addTask(new taskpool.Task(loadTexture, 2)); group.addTask(new taskpool.Task(loadModel, 1)); // 等待组内所有任务执行完毕统一返回结果数组 const results await taskpool.execute(group) as string[]; console.info(所有资源加载完成:, results); }7. 任务调度优先级控制与防饥饿机制为了防止高负载下低优先级任务被饿死TaskPool 提供了优先级调度机制底层对接了系统的 QoSQuality of Service。实战代码区分 UI 响应任务与后台备份任务import { taskpool } from kit.ArkTS; Concurrent function uiResponsiveTask(): string { return UI 刷新数据; } Concurrent function backgroundBackupTask(): string { return 后台数据备份; } async function priorityDemo() { // 高优先级立即调度保障 UI 流畅 const highTask new taskpool.Task(uiResponsiveTask); highTask.setPriority(taskpool.Priority.HIGH); // 空闲优先级仅在所有线程空闲时执行不抢占资源 const idleTask new taskpool.Task(backgroundBackupTask); idleTask.setPriority(taskpool.Priority.IDLE); taskpool.execute(highTask); taskpool.execute(idleTask); }8. 跨线程状态共享Sendable 装饰器与对象传递从 API version 11 开始若需要在跨并发实例传递带方法的实例对象该类必须使用Sendable装饰器标注以告知编译器该对象是线程安全的。实战代码跨线程传递自定义业务对象import { taskpool, Sendable } from kit.ArkTS; // 标记为 Sendable允许跨线程传递 Sendable class UserInfo { name: string; age: number; constructor(name: string, age: number) { this.name name; this.age age; } } Concurrent function processUser(user: UserInfo): string { return 处理用户: ${user.name}, 年龄: ${user.age}; } async function sendableDemo() { const user new UserInfo(Alice, 25); const task new taskpool.Task(processUser, user); const result await taskpool.execute(task) as string; console.info(result); }

相关新闻