27、鸿蒙Harmony Next开发：ArkTS并发（Promise和async/await和多线程并发TaskPool和Worker的使用）

异步并发 (Promise和async/await)

Promise

async/await

多线程并发

多线程并发模型

内存共享模型

Actor模型

TaskPool

TaskPool运作机制

TaskPool注意事项

@Concurrent装饰器

装饰器说明

装饰器使用示例

TaskPool扩缩容机制

扩容机制

缩容机制

Worker

Worker运作机制

Worker注意事项

创建Worker的注意事项

文件路径注意事项

生命周期注意事项

onAllErrors接口与onerror接口之间的行为差异

Worker基本用法示例

跨har包加载Worker

多级Worker生命周期管理

推荐使用示例

不推荐使用示例

TaskPool和Worker的对比

实现特点对比

适用场景对比

为了提升应用的响应速度与帧率，避免耗时任务影响UI主线程，ArkTS提供了异步并发和多线程并发两种处理策略。

异步并发是指异步代码在执行到一定程度后会被暂停，以便在未来某个时间点继续执行，这种情况下，同一时间只有一段代码在执行。ArkTS通过Promise和async/await提供异步并发能力，适用于单次I/O任务的开发场景。详细请参见使用异步并发能力。
多线程并发允许在同一时间段内同时执行多段代码。在UI主线程继续响应用户操作和更新UI的同时，后台线程也能执行耗时操作，从而避免应用出现卡顿。ArkTS通过TaskPool和Worker提供多线程并发能力，适用于耗时任务等并发场景。详细请参见多线程并发概述。

在并发多线程场景下，不同线程间需要进行数据通信。不同类别的对象采用不同的传输方式，如拷贝或内存共享。

并发能力广泛应用于多种场景，包括异步并发任务、耗时任务（如CPU密集型任务、I/O密集型任务和同步任务等）、长时任务、常驻任务等。开发者可以根据不同的任务诉求和场景，选择相应的并发策略进行优化和开发，具体案例可以参见应用多线程开发实践案例。

异步并发 (Promise和async/await)

Promise和async/await提供异步并发能力，是标准的JS异步语法。异步代码会被挂起并在之后继续执行，同一时间只有一段代码执行。以下是典型的异步并发使用场景：

I/O 非阻塞操作：网络请求、文件读写、定时器等。
任务轻量且无 CPU 阻塞：单次任务执行时间短。
逻辑依赖清晰：任务有明确的顺序或并行关系。

异步并发是一种编程语言的特性，允许程序在执行某些操作时不必等待其完成，可以继续执行其他操作。

Promise

Promise是一种用于处理异步操作的对象，可以将异步操作转换为类似于同步操作的风格，以方便代码编写和维护。Promise提供了一种状态机制来管理异步操作的不同阶段，Promise有三种状态：pending（进行中）、fulfilled（已完成，也叫resolved）和rejected（已拒绝）。其创建后处于pending状态，异步操作完成后转换为fulfilled或rejected状态。

Promise提供了then/catch方法来注册回调函数以处理异步操作的成功或失败的结果，Promise状态改变会触发回调函数加入微任务队列等待执行，依赖事件循环机制在宏任务执行完成后优先执行微任务，保证回调函数的异步调度。

最基本的用法是通过构造函数实例化一个Promise对象，同时传入一个带有两个参数的函数，通常称为executor函数。executor函数接收两个参数：resolve和reject，分别表示异步操作成功和失败时的回调函数。例如，以下代码创建了一个Promise对象并模拟了一个异步操作：

const promise: Promise<number> = new Promise((resolve: Function, reject: Function) => {setTimeout(() => {const randomNumber: number = Math.random();if (randomNumber > 0.5) {resolve(randomNumber);} else {reject(new Error('Random number is too small'));}}, 1000);
})

在上述代码中，setTimeout函数模拟了一个异步操作，1秒后生成一个随机数。如果随机数大于0.5，调用resolve回调函数并传递该随机数；否则调用reject回调函数并传递一个错误对象。

Promise对象创建后，可以使用then方法和catch方法指定fulfilled状态和rejected状态的回调函数。then方法可接受两个参数，一个处理fulfilled状态的函数，另一个处理rejected状态的函数。只传一个参数则表示当Promise对象状态变为fulfilled时，then方法会自动调用这个回调函数，并将Promise对象的结果作为参数传递给它。使用catch方法注册一个回调函数，用于处理“失败”的结果，即捕获Promise的状态改变为rejected状态或操作失败抛出的异常。例如：

import { BusinessError } from '@kit.BasicServicesKit';// 使用 then 方法定义成功和失败的回调
promise.then((result: number) => {console.info(`The number for success is ${result}`); // 成功时执行
}, (error: BusinessError) => {console.error(error.message); // 失败时执行
}
);// 使用 then 方法定义成功的回调，catch 方法定义失败的回调
promise.then((result: number) => {console.info(`Random number is ${result}`); // 成功时执行
}).catch((error: BusinessError) => {console.error(error.message); // 失败时执行
});

在上述代码中，then方法的回调函数接收Promise对象的成功结果作为参数，并输出到控制台。如果Promise对象进入rejected状态，catch方法的回调函数接收错误对象作为参数，并输出到控制台。

说明

当Promise被reject且未通过catch方法处理时，会触发unhandledrejection事件。可使用errorManager.on('unhandledrejection')接口监听该事件，以全局捕获未处理的Promise reject。

async/await

async/await是一种用于处理异步操作的Promise语法糖，使得编写异步代码变得更加简单和易读。通过使用async关键字声明一个函数为异步函数，并使用await关键字等待Promise的解析（完成或拒绝），以同步的方式编写异步操作的代码。

async函数通过返回Promise对象实现异步操作，其内部可以包含零个或者多个await关键字，通过await暂停执行直至关联的Promise完成状态转换（fulfilled/rejected）。若函数执行过程中抛出异常，该异常将直接触发返回的Promise进入rejected状态，错误对象可通过.catch()方法或then的第二个回调参数捕获。

下面是一个使用async/await的例子，其中模拟了一个以同步方式执行异步操作的场景，该操作会在3秒钟后返回一个字符串。

async function myAsyncFunction(): Promise<string> {const result: string = await new Promise((resolve: Function) => {setTimeout(() => {resolve('Hello, world!');}, 3000);});console.info(result); // 输出： Hello, world!return result;
}@Entry
@Component
struct Index {@State message: string = 'Hello World';build() {Row() {Column() {Text(this.message).fontSize(50).fontWeight(FontWeight.Bold).onClick(async () => {let res = await myAsyncFunction();console.info('res is: ' + res);})}.width('100%')}.height('100%')}
}

在上述示例代码中，使用了await关键字来等待Promise对象的解析，并将其解析值存储在result变量中。

需要注意的是，等待异步操作时，需将整个操作包在async函数中，并搭配await使用，同时也可使用try/catch块来捕获异步操作中的异常。

async function myAsyncFunction(): Promise<void> {try {const result: string = await new Promise((resolve: Function) => {resolve('Hello, world!');});} catch (e) {console.error(`Get exception: ${e}`);}
}myAsyncFunction();

多线程并发

多线程并发是指在单个程序中同时运行多个线程，通过并行或交替执行任务来提升性能和资源利用率的编程模型。在ArkTS应用开发过程中，需要用到多线程并发的业务场景有很多，针对常见的业务场景，主要可以分为以下三类。

业务逻辑包含较大计算量或多次I/O读写等需要长时间执行的任务，例如图片/视频编解码，压缩/解压缩，数据库操作等场景。
业务逻辑包含监听或定期采集数据等需要长时间保持运行的任务，例如定期采集传感器数据场景。
业务逻辑跟随主线程生命周期或与主线程绑定的任务，例如游戏中台场景。

并发模型用于实现不同应用场景中的并发任务。常见的并发模型有基于内存共享的模型和基于消息通信的模型。

Actor并发模型是基于消息通信的并发模型的典型代表。它使开发者无需处理锁带来的复杂问题，并且具有较高的并发度，因此得到了广泛的应用。

当前ArkTS提供了TaskPool和Worker两种并发能力，两者均基于Actor并发模型实现。

Actor并发模型和内存共享并发模型的具体对比请见多线程并发模型。

多线程并发模型

内存共享并发模型指多线程同时执行任务，这些线程依赖同一内存资源并且都有权限访问，线程访问内存前需要抢占并锁定内存的使用权，没有抢占到内存的线程需要等待其他线程释放使用权再执行。

Actor并发模型每一个线程都是一个独立Actor，每个Actor有自己独立的内存，Actor之间通过消息传递机制触发对方Actor的行为，不同Actor之间不能直接访问对方的内存空间。

Actor并发模型相较于内存共享并发模型，不同线程间的内存是隔离的，因此不会出现线程竞争同一内存资源的情况。开发者无需处理内存上锁相关的问题，从而提高开发效率。

Actor并发模型中，线程不共享内存，需通过线程间通信机制传递任务和结果。

本文以经典的生产者消费者问题为例，对比呈现这两种模型在解决具体问题时的差异。

内存共享模型

以下示例伪代码和示意图展示了如何使用内存共享模型解决生产者消费者问题。

为了避免不同生产者或消费者同时访问一块共享内存的容器时产生的脏读、脏写现象，同一时间只能有一个生产者或消费者访问该容器，也就是不同生产者和消费者争夺使用容器的锁。当一个角色获取锁之后其他角色需要等待该角色释放锁之后才能重新尝试获取锁以访问该容器。

// 此段示例为伪代码仅作为逻辑示意，便于开发者理解使用内存共享模型和Actor模型的区别
class Queue {// ...push(value: number) {// ...}empty(): boolean {// ...return true;}pop(value: number): number {// ...return value;}// ...
}class Mutex {// ...lock(): boolean {// ...return true;}unlock() {// ...}// ...
}class BufferQueue {queue: Queue = new Queue();mutex: Mutex = new Mutex();add(value: number) {// 尝试获取锁if (this.mutex.lock()) {this.queue.push(value);this.mutex.unlock();}}take(value: number): number {let res: number = 0;// 尝试获取锁if (this.mutex.lock()) {if (this.queue.empty()) {res = 1;return res;}let num: number = this.queue.pop(value);this.mutex.unlock();res = num;}return res;}
}// 构造一段全局共享的内存
let g_bufferQueue = new BufferQueue();class Producer {constructor() {}run() {let value = Math.random();// 跨线程访问bufferQueue对象g_bufferQueue.add(value);}
}class ConsumerTest {constructor() {}run() {// 跨线程访问bufferQueue对象let num = 123;let res = g_bufferQueue.take(num);if (res != null) {// 添加消费逻辑}}
}function Main(): void {let consumer: ConsumerTest = new ConsumerTest();let producer: Producer = new Producer();let threadNum: number = 10;for (let i = 0; i < threadNum; i++) {// 如下伪代码模拟启动多线程执行生产任务// let thread = new Thread();// thread.run(producer.run());// consumer.run();}
}

Actor模型

以下示例简单展示了如何使用基于Actor模型的TaskPool并发能力来解决生产者消费者问题。

Actor模型中，不同角色之间并不共享内存，生产者线程和UI线程都有自己的虚拟机实例，两个虚拟机实例之间拥有独占的内存，相互隔离。生产者生产出结果后，通过序列化通信将结果发送给UI线程，UI线程消费结果后再发送新的生产任务给生产者线程。

import { taskpool } from '@kit.ArkTS';// 跨线程并发任务
@Concurrent
async function produce(): Promise<number> {// 添加生产相关逻辑console.info('producing...');return Math.random();
}class Consumer {public consume(value: Object) {// 添加消费相关逻辑console.info('consuming value: ' + value);}
}@Entry
@Component
struct Index {@State message: string = 'Hello World';build() {Row() {Column() {Text(this.message).fontSize(50).fontWeight(FontWeight.Bold)Button() {Text('start')}.onClick(() => {let produceTask: taskpool.Task = new taskpool.Task(produce);let consumer: Consumer = new Consumer();for (let index: number = 0; index < 10; index++) {// 执行生产异步并发任务taskpool.execute(produceTask).then((res: Object) => {consumer.consume(res);}).catch((e: Error) => {console.error(e.message);})}}).width('20%').height('20%')}.width('100%')}.height('100%')}
}

也可以等待生产者完成所有生产任务，通过序列化通信将结果发送给UI线程。UI线程接收完毕后，由消费者统一消费结果。

import { taskpool } from '@kit.ArkTS';// 跨线程并发任务
@Concurrent
async function produce(): Promise<number> {// 添加生产相关逻辑console.info('producing...');return Math.random();
}class Consumer {public consume(value: Object) {// 添加消费相关逻辑console.info('consuming value: ' + value);}
}@Entry
@Component
struct Index {@State message: string = 'Hello World'build() {Row() {Column() {Text(this.message).fontSize(50).fontWeight(FontWeight.Bold)Button() {Text('start')}.onClick(async () => {let dataArray = new Array<number>();let produceTask: taskpool.Task = new taskpool.Task(produce);let consumer: Consumer = new Consumer();for (let index: number = 0; index < 10; index++) {// 执行生产异步并发任务let result = await taskpool.execute(produceTask) as number;dataArray.push(result);}for (let index: number = 0; index < dataArray.length; index++) {consumer.consume(dataArray[index]);}}).width('20%').height('20%')}.width('100%')}.height('100%')}
}

TaskPool

TaskPool为应用程序提供多线程环境，降低资源消耗、提高系统性能，无需管理线程生命周期。

TaskPool运作机制

TaskPool运作机制示意图

TaskPool支持开发者在宿主线程提交任务到任务队列，系统选择合适的工作线程执行任务，再将结果返回给宿主线程。接口易用，支持任务执行、取消和指定优先级，同时通过系统统一线程管理，结合动态调度及负载均衡算法，可以节约系统资源。系统默认启动一个任务工作线程，任务多时会扩容。工作线程数量上限取决于设备的物理核数，内部管理具体数量，确保调度和执行效率最优。长时间无任务分发时会缩容，减少工作线程数量。具体扩缩容机制详情请见TaskPool扩缩容机制。

TaskPool注意事项

实现任务的函数需要使用@Concurrent装饰器标注，且仅支持在.ets文件中使用。
从API version 11开始，跨并发实例传递带方法的实例对象时，该类必须使用装饰器@Sendable装饰器标注，且仅支持在.ets文件中使用。
任务函数（LongTask除外）在TaskPool工作线程中的执行时长不能超过3分钟。否则，任务将被强制终止。需要注意的是，这里的3分钟限制仅统计TaskPool线程的同步运行时长，不包含异步操作（如Promise或async/await）的等待时长。例如，数据库的插入、删除、更新等操作，如果是异步操作，仅计入CPU实际处理时长（如SQL解析），网络传输或磁盘I/O等待时长不计入；如果是同步操作，整个操作时长（含I/O阻塞时间）均计入限制。开发者可通过Task的属性ioDuration、cpuDuration获取执行当前任务的异步IO耗时和CPU耗时。
实现任务的函数入参需满足序列化支持的类型，详情请参见线程间通信对象。目前不支持使用@State装饰器、@Prop装饰器、@Link装饰器等装饰器修饰的复杂类型。

ArrayBuffer参数在TaskPool中默认转移，需要设置转移列表的话可通过接口setTransferList()设置。如果需要多次调用使用ArrayBuffer作为参数的task，则需要通过接口setCloneList()把ArrayBuffer在线程中的传输行为改成拷贝传递，避免对原有对象产生影响。

import { taskpool } from '@kit.ArkTS';
import { BusinessError } from '@kit.BasicServicesKit';@Concurrent
function printArrayBuffer(buffer: ArrayBuffer) {return buffer;
}function testArrayBuffer() {const buffer = new ArrayBuffer(1);const group = new taskpool.TaskGroup();const task = new taskpool.Task(printArrayBuffer, buffer);group.addTask(task);task.setCloneList([buffer]);for (let i = 0; i < 5; i++) {taskpool.execute(group).then(() => {console.info('execute group success');}).catch((e: BusinessError) => {console.error(`execute group error: ${e.message}`);})}
}

由于不同线程中上下文对象是不同的，因此TaskPool工作线程只能使用线程安全的库，例如UI相关的非线程安全库不能使用。
序列化传输的数据量限制为16MB。
Priority的IDLE优先级是用来标记需要在后台运行的耗时任务（例如数据同步、备份），它的优先级别是最低的。这种优先级的任务只在所有线程都空闲时触发执行，并且同一时间只会有一个IDLE优先级的任务执行。
Promise不支持跨线程传递。TaskPool返回pending或rejected状态的Promise时会失败，返回fulfilled状态的Promise时TaskPool会解析返回的结果，如果结果可以跨线程传递，则返回成功。
不支持在TaskPool工作线程中使用AppStorage。
TaskPool支持开发者在宿主线程封装任务并提交给任务队列，理论上支持的任务数量没有上限。然而，任务的执行效率受限于任务的优先级和系统资源。当工作线程达到最大数量时，任务的执行效率可能会下降。
TaskPool不支持指定任务所运行的线程，任务会被分配到空闲的线程中执行。如果需要指定任务所运行的线程，建议使用Worker。

@Concurrent装饰器

在使用TaskPool时，执行的并发函数需要使用该装饰器修饰，否则无法通过相关校验。

装饰器说明

@Concurrent并发装饰器	说明
装饰器参数	无。
使用场景	仅支持在Stage模型的工程中使用。仅支持在.ets文件中使用。
装饰的函数类型	允许标注async函数或普通函数。禁止标注generator、箭头函数、类方法。不支持类成员函数或者匿名函数。
装饰的函数内的变量类型	允许使用local变量、入参和通过import引入的变量。禁止使用闭包变量。
装饰的函数内的返回值类型	支持的类型请查线程间通信对象。

说明

由于@Concurrent标记的函数不能访问闭包，因此函数内部不能调用当前文件的其他函数，例如：
function bar() {
}@Concurrent
function foo() {bar(); // 违反闭包原则，报错
}

装饰器使用示例

并发函数一般使用

并发函数为一个计算两数之和的普通函数，taskpool执行该函数并返回结果。

import { taskpool } from '@kit.ArkTS';@Concurrent
function add(num1: number, num2: number): number {return num1 + num2;
}async function concurrentFunc(): Promise<void> {try {const task: taskpool.Task = new taskpool.Task(add, 1, 2);console.info(`taskpool res is: ${await taskpool.execute(task)}`); // 输出结果：taskpool res is: 3} catch (e) {console.error(`taskpool execute error is: ${e}}`);}
}@Entry
@Component
struct Index {@State message: string = 'Hello World';build() {Row() {Column() {Text(this.message).fontSize(50).fontWeight(FontWeight.Bold).onClick(() => {concurrentFunc();})}.width('100%')}.height('100%')}
}

并发函数返回Promise

并发函数中返回Promise时需要特别关注。如示例所示，testPromise和testPromise1等需处理Promise并返回结果。

import { taskpool } from '@kit.ArkTS';@Concurrent
function testPromise(args1: number, args2: number): Promise<number> {return new Promise<number>((resolve, reject) => {resolve(args1 + args2);});
}@Concurrent
async function testPromise1(args1: number, args2: number): Promise<number> {return new Promise<number>((resolve, reject) => {resolve(args1 + args2);});
}@Concurrent
async function testPromise2(args1: number, args2: number): Promise<number> {return await new Promise<number>((resolve, reject) => {resolve(args1 + args2);});
}@Concurrent
function testPromise3() {return Promise.resolve(1);
}@Concurrent
async function testPromise4(): Promise<number> {return 1;
}@Concurrent
async function testPromise5(): Promise<string> {return await new Promise((resolve) => {setTimeout(() => {resolve('Promise setTimeout after resolve');}, 1000)});
}async function testConcurrentFunc() {const task1: taskpool.Task = new taskpool.Task(testPromise, 1, 2);const task2: taskpool.Task = new taskpool.Task(testPromise1, 1, 2);const task3: taskpool.Task = new taskpool.Task(testPromise2, 1, 2);const task4: taskpool.Task = new taskpool.Task(testPromise3);const task5: taskpool.Task = new taskpool.Task(testPromise4);const task6: taskpool.Task = new taskpool.Task(testPromise5);taskpool.execute(task1).then((d: object) => {console.info(`task1 res is: ${d}`); // 输出结果：task1 res is: 3}).catch((e: object) => {console.error(`task1 catch e: ${e}`);})taskpool.execute(task2).then((d: object) => {console.info(`task2 res is: ${d}`);}).catch((e: object) => {console.error(`task2 catch e: ${e}`); // 输出结果：task2 catch e: Error: Can't return Promise in pending state})taskpool.execute(task3).then((d: object) => {console.info(`task3 res is: ${d}`); // 输出结果：task3 res is: 3}).catch((e: object) => {console.error(`task3 catch e: ${e}`);})taskpool.execute(task4).then((d: object) => {console.info(`task4 res is: ${d}`); // 输出结果：task4 res is: 1}).catch((e: object) => {console.error(`task4 catch e: ${e}`);})taskpool.execute(task5).then((d: object) => {console.info(`task5 res is: ${d}`); // 输出结果：task5 res is: 1}).catch((e: object) => {console.error(`task5 catch e: ${e}`);})taskpool.execute(task6).then((d: object) => {console.info(`task6 res is: ${d}`); // 输出结果：task6 res is: Promise setTimeout after resolve}).catch((e: object) => {console.error(`task6 catch e: ${e}`);})
}@Entry
@Component
struct Index {@State message: string = 'Hello World';build() {Row() {Column() {Button(this.message).fontSize(50).fontWeight(FontWeight.Bold).onClick(() => {testConcurrentFunc();})}.width('100%')}.height('100%')}
}

并发函数中使用自定义类或函数

在并发函数中使用自定义类或函数时，需将其定义在不同的文件中，否则会被认为是闭包。如以下示例所示。

// Index.ets
import { taskpool } from '@kit.ArkTS';
import { BusinessError } from '@kit.BasicServicesKit';
import { testAdd, MyTestA, MyTestB } from './Test';function add(arg: number) {return ++arg;
}class TestA {constructor(name: string) {this.name = name;}name: string = 'ClassA';
}class TestB {static nameStr: string = 'ClassB';
}@Concurrent
function TestFunc() {// case1：在并发函数中直接调用同文件内定义的类或函数// 直接调用同文件定义的函数add()，add飘红报错：Only imported variables and local variables can be used in @Concurrent decorated functions. <ArkTSCheck>// add(1);// 直接使用同文件定义的TestA构造，TestA飘红报错：Only imported variables and local variables can be used in @Concurrent decorated functions. <ArkTSCheck>// const a = new TestA('aaa');// 直接访问同文件定义的TestB的成员nameStr，TestB飘红报错：Only imported variables and local variables can be used in @Concurrent decorated functions. <ArkTSCheck>// console.info(`TestB name is: ${TestB.nameStr}`);// case2：在并发函数中调用定义在Test.ets文件并导入当前文件的类或函数// 输出结果：res1 is: 2console.info(`res1 is: ${testAdd(1)}`);const tmpStr = new MyTestA('TEST A');// 输出结果：res2 is: TEST Aconsole.info(`res2 is: ${tmpStr.name}`);// 输出结果：res3 is: MyTestBconsole.info(`res3 is: ${MyTestB.nameStr}`);
}@Entry
@Component
struct Index {@State message: string = 'Hello World';build() {RelativeContainer() {Text(this.message).id('HelloWorld').fontSize(50).fontWeight(FontWeight.Bold).alignRules({center: { anchor: '__container__', align: VerticalAlign.Center },middle: { anchor: '__container__', align: HorizontalAlign.Center }}).onClick(() => {const task = new taskpool.Task(TestFunc);taskpool.execute(task).then(() => {console.info('taskpool: execute task success!');}).catch((e: BusinessError) => {console.error(`taskpool: execute: Code: ${e.code}, message: ${e.message}`);})})}.height('100%').width('100%')}
}

// Test.ets
export function testAdd(arg: number) {return ++arg;
}@Sendable
export class MyTestA {constructor(name: string) {this.name = name;}name: string = 'MyTestA';
}export class MyTestB {static nameStr:string = 'MyTestB';
}

并发异步函数中使用Promise

在并发异步函数中使用Promise时，建议搭配await使用。这样TaskPool可以捕获Promise中的异常。推荐使用示例如下。

import { taskpool } from '@kit.ArkTS';@Concurrent
async function testPromiseError() {await new Promise<number>((resolve, reject) => {resolve(1);}).then(() => {throw new Error('testPromise error');})
}@Concurrent
async function testPromiseError1() {await new Promise<string>((resolve, reject) => {reject('testPromiseError1 error msg');})
}@Concurrent
function testPromiseError2() {return new Promise<string>((resolve, reject) => {reject('testPromiseError2 error msg');})
}async function testConcurrentFunc() {const task1: taskpool.Task = new taskpool.Task(testPromiseError);const task2: taskpool.Task = new taskpool.Task(testPromiseError1);const task3: taskpool.Task = new taskpool.Task(testPromiseError2);taskpool.execute(task1).then((d: object) => {console.info(`task1 res is: ${d}`);}).catch((e: object) => {console.error(`task1 catch e: ${e}`); // task1 catch e: Error: testPromise error})taskpool.execute(task2).then((d: object) => {console.info(`task2 res is: ${d}`);}).catch((e: object) => {console.error(`task2 catch e: ${e}`); // task2 catch e: testPromiseError1 error msg})taskpool.execute(task3).then((d: object) => {console.info(`task3 res is: ${d}`);}).catch((e: object) => {console.error(`task3 catch e: ${e}`); // task3 catch e: testPromiseError2 error msg})
}@Entry
@Component
struct Index {@State message: string = 'Hello World';build() {Row() {Column() {Button(this.message).fontSize(50).fontWeight(FontWeight.Bold).onClick(() => {testConcurrentFunc();})}.width('100%')}.height('100%')}
}

TaskPool扩缩容机制

扩容机制

一般情况下，开发者向任务队列提交任务时会触发扩容检测。扩容检测首先判断当前空闲的工作线程数是否大于任务数，如果大于，说明线程池中存在空闲工作线程，无需扩容。否则，通过负载计算确定所需工作线程数并创建。

缩容机制

扩容后，TaskPool新建多个工作线程，但当任务数减少后，这些线程就会处于空闲状态，造成资源浪费，因此TaskPool提供缩容机制。TaskPool使用了定时器，定时检测当前负载。定时器30s触发一次，每次尝试释放空闲的工作线程。释放的线程需要满足如下条件：

该线程空闲时长达到30s。
该线程上未执行长时任务（LongTask）。
该线程上没有业务申请且未释放的句柄，例如Timer(定时器)。
该线程处于非调试调优阶段。
该线程中不存在已创建未销毁的子Worker。

Worker

Worker的主要作用是为应用程序提供一个多线程的运行环境，满足应用程序在执行过程中与宿主线程分离，在后台线程中运行脚本进行耗时操作，避免计算密集型或高延迟的任务阻塞宿主线程。

Worker运作机制

图1 Worker运作机制示意图

创建Worker的线程称为宿主线程（不局限于主线程，Worker线程也支持创建Worker子线程）。Worker子线程（或Actor线程、工作线程）是Worker自身运行的线程。每个Worker子线程和宿主线程拥有独立的实例，包含独立执行环境、对象、代码段等。因此，启动每个Worker存在一定的内存开销，需要限制Worker子线程的数量。Worker子线程和宿主线程通过消息传递机制通信，利用序列化机制完成命令和数据的交互。

Worker注意事项

创建Worker时，提供手动和自动两种创建方式，推荐使用自动创建方式。手动创建Worker线程目录及文件时，需同步进行相关配置，具体要求请参阅创建Worker的注意事项。
使用Worker能力时，构造函数中传入的Worker线程文件的路径在不同版本有不同的规则，详情请参见文件路径注意事项。
Worker创建后需要手动管理生命周期。同时运行的Worker子线程数量最多为64个，并且与napi_create_ark_runtime创建的runtime总数不超过80。详情请参见生命周期注意事项。
不同线程中上下文对象是不同的，因此Worker线程只能使用线程安全的库，例如UI相关的非线程安全库不能在Worker子线程中使用。
单次序列化传输的数据量大小限制为16MB。
使用Worker模块时，API version 18及之后的版本建议在宿主线程中注册onAllErrors回调，以捕获Worker线程生命周期内的各种异常。API version 18之前的版本应注册onerror回调。如果未注册onAllErrors或onerror回调，当Worker线程出现异常时会发生jscrash问题。需要注意的是，onerror接口仅能捕获onmessage回调中的同步异常，捕获异常后，Worker线程将进入销毁流程，无法继续使用。详情请参见onAllErrors接口与onerror接口之间的行为差异。
不支持跨HAP使用Worker线程文件。
引用HAR/HSP中的worker前，需要先配置对HAR/HSP的依赖，详见引用共享包。
不支持在Worker工作线程中使用AppStorage。
从API version 18开始，Worker线程优先级可以在构造函数的参数WorkerOptions中进行指定。
在Worker文件中禁止使用export语法导出任何内容，否则会导致jscrash问题。

创建Worker的注意事项

Worker线程文件需要放在"{moduleName}/src/main/ets/"目录层级之下，否则不会被打包到应用中。有手动和自动两种创建Worker线程目录及文件的方式。

手动创建：开发者手动创建相关目录及文件，通常是在ets目录下创建一个workers文件夹，用于存放worker.ets文件，需要配置build-profile.json5的相关字段信息，确保Worker线程文件被打包到应用中。

Stage模型：

"buildOption": {"sourceOption": {"workers": ["./src/main/ets/workers/worker.ets"]}
}

FA模型：

"buildOption": {"sourceOption": {"workers": ["./src/main/ets/MainAbility/workers/worker.ets"]}
}

自动创建：DevEco Studio支持一键生成Worker，在对应的{moduleName}目录下任意位置，点击鼠标右键 > New > Worker，即可自动生成Worker的模板文件及配置信息，无需再手动在build-profile.json5中进行相关配置。

文件路径注意事项

当使用Worker模块具体功能时，均需先构造Worker实例对象，其构造函数与API版本相关，且构造函数需要传入Worker线程文件的路径（scriptURL）。

// 导入模块
import { worker } from '@kit.ArkTS';// API version 9及之后版本使用：
const worker1: worker.ThreadWorker = new worker.ThreadWorker('entry/ets/workers/worker.ets');
// API version 8及之前版本使用：
const worker2: worker.Worker = new worker.Worker('entry/ets/workers/worker.ets');

Stage模型下的文件路径规则

构造函数中的scriptURL要求如下：

scriptURL的组成包含{moduleName}/ets和相对路径relativePath。
relativePath是Worker线程文件相对于"{moduleName}/src/main/ets/"目录的相对路径。

1）加载Ability中Worker线程文件场景

加载Ability中的worker线程文件，加载路径规则：{moduleName}/ets/{relativePath}。

import { worker } from '@kit.ArkTS';// worker线程文件所在路径："entry/src/main/ets/workers/worker.ets"
const workerStage1: worker.ThreadWorker = new worker.ThreadWorker('entry/ets/workers/worker.ets');// worker线程文件所在路径："testworkers/src/main/ets/ThreadFile/workers/worker.ets"
const workerStage2: worker.ThreadWorker = new worker.ThreadWorker('testworkers/ets/ThreadFile/workers/worker.ets');

2）加载HSP中Worker线程文件场景

加载HSP中worker线程文件，加载路径规则：{moduleName}/ets/{relativePath}。

import { worker } from '@kit.ArkTS';// worker线程文件所在路径： "hsp/src/main/ets/workers/worker.ets"
const workerStage3: worker.ThreadWorker = new worker.ThreadWorker('hsp/ets/workers/worker.ets');

3）加载HAR中Worker线程文件场景

加载HAR中worker线程文件存在以下两种情况：

@标识路径加载形式：所有种类的模块加载本地HAR中的Worker线程文件，加载路径规则：@{moduleName}/ets/{relativePath}。
相对路径加载形式：本地HAR加载该包内的Worker线程文件，加载路径规则：创建Worker对象所在文件与Worker线程文件的相对路径。

当开启useNormalizedOHMUrl（在工程目录中与entry同级别的应用级build-profile.json5文件中，将strictMode属性下的useNormalizedOHMUrl字段配置为true）或HAR包被打包成三方包使用时，HAR包中使用Worker仅支持通过相对路径的加载形式创建。

import { worker } from '@kit.ArkTS';// @标识路径加载形式：
// worker线程文件所在路径: "har/src/main/ets/workers/worker.ets"
const workerStage4: worker.ThreadWorker = new worker.ThreadWorker('@har/ets/workers/worker.ets');// 相对路径加载形式：
// worker线程文件所在路径: "har/src/main/ets/workers/worker.ets"
// 创建Worker对象的文件所在路径："har/src/main/ets/components/mainpage/MainPage.ets"
const workerStage5: worker.ThreadWorker = new worker.ThreadWorker('../../workers/worker.ets');

FA模型下的文件路径规则

构造函数中的scriptURL为：Worker线程文件与"{moduleName}/src/main/ets/MainAbility"的相对路径。

import { worker } from '@kit.ArkTS';// 主要说明以下三种场景：// 场景1： Worker线程文件所在路径："{moduleName}/src/main/ets/MainAbility/workers/worker.ets"
const workerFA1: worker.ThreadWorker = new worker.ThreadWorker('workers/worker.ets', {name:'first worker in FA model'});// 场景2： Worker线程文件所在路径："{moduleName}/src/main/ets/workers/worker.ets"
const workerFA2: worker.ThreadWorker = new worker.ThreadWorker('../workers/worker.ets');// 场景3： Worker线程文件所在路径："{moduleName}/src/main/ets/MainAbility/ThreadFile/workers/worker.ets"
const workerFA3: worker.ThreadWorker = new worker.ThreadWorker('ThreadFile/workers/worker.ets');

生命周期注意事项

Worker的创建和销毁会消耗较多的系统资源，建议开发者合理管理已创建的Worker并重复使用。Worker空闲时仍会占用资源，因此当不需要Worker时，可以调用terminate()接口或close()方法主动销毁Worker。若Worker处于已销毁或正在销毁等非运行状态时，调用其功能接口，会抛出相应的错误。
Worker的数量由内存管理策略决定，设定的内存阈值为1.5GB和设备物理内存的60%中的较小者。在内存允许的情况下，系统最多可以同时运行64个Worker。如果尝试创建的Worker数量超出这一上限，系统将抛出错误：“Worker initialization failure, the number of workers exceeds the maximum.”。实际运行的Worker数量会根据当前内存使用情况实时调整。当所有Worker和主线程的累积内存占用超过设定的阈值时，系统将触发内存溢出（OOM）错误，导致应用程序崩溃。

onAllErrors接口与onerror接口之间的行为差异

异常捕获范围

onAllErrors接口可以捕获Worker线程的onmessage回调、timer回调以及文件执行等流程产生的全局异常。

onerror接口仅能捕获Worker线程的onmessage回调中同步方法产生的异常，无法捕获多线程回调和模块化相关异常。

异常捕获后的线程状态

onAllErrors接口捕获异常后，Worker线程仍然存活并可以继续使用。这使得开发者可以在捕获异常后继续执行其他操作，而不必担心线程被终止。

onerror接口一旦捕获到异常，Worker线程会进入销毁流程，无法继续使用。这意味着在onerror触发后，Worker线程将被终止，后续操作将无法进行。

异常捕获范围

onAllErrors接口适用于捕获Worker线程中所有类型异常的场景，特别是确保异常发生后Worker线程仍能继续运行的复杂场景。

onerror接口适用于只需要捕获onmessage回调中同步异常的简单场景。由于捕获异常后线程会被销毁，适用于不需要继续使用Worker线程的情况。

推荐使用onAllErrors接口，因为它提供了更全面的异常捕获能力，并且不会导致线程终止。

Worker基本用法示例

DevEco Studio支持一键生成Worker，在对应的{moduleName}目录下任意位置，单击鼠标右键 > New > Worker，即可自动生成Worker的模板文件及配置信息。本文以创建“worker”为例。此外，支持手动创建Worker文件，具体方式和注意事项请参阅创建Worker的注意事项。

导入Worker模块。

// Index.ets
import { ErrorEvent, MessageEvents, worker } from '@kit.ArkTS'

在宿主线程中通过调用ThreadWorker的constructor()方法创建Worker对象，并注册回调函数。

// Index.ets
@Entry
@Component
struct Index {@State message: string = 'Hello World';build() {RelativeContainer() {Text(this.message).id('HelloWorld').fontSize(50).fontWeight(FontWeight.Bold).alignRules({center: { anchor: '__container__', align: VerticalAlign.Center },middle: { anchor: '__container__', align: HorizontalAlign.Center }}).onClick(() => {// 创建Worker对象let workerInstance = new worker.ThreadWorker('entry/ets/workers/worker.ets');// 注册onmessage回调，当宿主线程接收到来自其创建的Worker通过workerPort.postMessage接口发送的消息时被调用，在宿主线程执行workerInstance.onmessage = (e: MessageEvents) => {let data: string = e.data;console.info('workerInstance onmessage is: ', data);}// 注册onAllErrors回调，可以捕获Worker线程的onmessage回调、timer回调以及文件执行等流程产生的全局异常，在宿主线程执行workerInstance.onAllErrors = (err: ErrorEvent) => {console.error('workerInstance onAllErrors message is: ' + err.message);}// 注册onmessageerror回调，当Worker对象接收到无法序列化的消息时被调用，在宿主线程执行workerInstance.onmessageerror = () => {console.error('workerInstance onmessageerror');}// 注册onexit回调，当Worker销毁时被调用，在宿主线程执行workerInstance.onexit = (e: number) => {// 如果Worker正常退出，code为0；如果异常退出，code为1console.info('workerInstance onexit code is: ', e);}// 发送消息给Worker线程workerInstance.postMessage('1');})}.height('100%').width('100%')}
}

跨har包加载Worker

创建har详情参考开发静态共享包。

在har中创建Worker线程文件相关内容。

// worker.ets
workerPort.onmessage = (e: MessageEvents) => {console.info('worker thread receive message: ', e.data);workerPort.postMessage('worker thread post message to main thread');
}

在entry模块的oh-package.json5文件中配置har包的依赖。

// 在entry模块配置har包的依赖
{"name": "entry","version": "1.0.0","description": "Please describe the basic information.","main": "","author": "","license": "","dependencies": {"har": "file:../har"}
}

在entry模块中加载har包中的Worker线程文件。

// Index.ets
import { worker } from '@kit.ArkTS';@Entry
@Component
struct Index {@State message: string = 'Hello World';build() {RelativeContainer() {Text(this.message).id('HelloWorld').fontSize(50).fontWeight(FontWeight.Bold).alignRules({center: { anchor: '__container__', align: VerticalAlign.Center },middle: { anchor: '__container__', align: HorizontalAlign.Center }}).onClick(() => {// 通过@标识路径加载形式，加载har中Worker线程文件let workerInstance = new worker.ThreadWorker('@har/ets/workers/worker.ets');workerInstance.onmessage = () => {console.info('main thread onmessage');};workerInstance.postMessage('hello world');})}.height('100%').width('100%')}
}

多级Worker生命周期管理

支持创建多级Worker，即父Worker可以创建子Worker，形成层级线程关系。由于Worker线程的生命周期由开发者自行管理，因此需要正确管理多级Worker的生命周期。如果销毁父Worker时未能终止其子Worker的运行，可能会导致不可预期的结果。因此需确保子Worker的生命周期在父Worker生命周期范围内，销毁父Worker前，先销毁所有子Worker。

推荐使用示例

// 在主线程中创建Worker线程（父Worker），在worker线程中再次创建Worker线程（子Worker）
// main thread
import { worker, MessageEvents, ErrorEvent } from '@kit.ArkTS';// 主线程中创建父worker对象
const parentworker = new worker.ThreadWorker('entry/ets/workers/parentworker.ets');parentworker.onmessage = (e: MessageEvents) => {console.info('主线程收到父worker线程信息 ' + e.data);
}parentworker.onexit = () => {console.info('父worker退出');
}parentworker.onAllErrors = (err: ErrorEvent) => {console.error('主线程接收到父worker报错 ' + err);
}parentworker.postMessage('主线程发送消息给父worker-推荐示例');

// parentworker.ets
import { ErrorEvent, MessageEvents, ThreadWorkerGlobalScope, worker } from '@kit.ArkTS';// 创建父Worker线程中与主线程通信的对象
const workerPort: ThreadWorkerGlobalScope = worker.workerPort;workerPort.onmessage = (e : MessageEvents) => {if (e.data == '主线程发送消息给父worker-推荐示例') {let childworker = new worker.ThreadWorker('entry/ets/workers/childworker.ets');childworker.onmessage = (e: MessageEvents) => {console.info('父Worker收到子Worker的信息 ' + e.data);if (e.data == '子Worker向父Worker发送信息') {workerPort.postMessage('父Worker向主线程发送信息');}}childworker.onexit = () => {console.info('子Worker退出');// 子Worker退出后再销毁父WorkerworkerPort.close();}childworker.onAllErrors = (err: ErrorEvent) => {console.error('子Worker发生报错 ' + err);}childworker.postMessage('父Worker向子Worker发送信息-推荐示例');}
}

// childworker.ets
import { ErrorEvent, MessageEvents, ThreadWorkerGlobalScope, worker } from '@kit.ArkTS';// 创建子Worker线程中与父Worker线程通信的对象
const workerPort: ThreadWorkerGlobalScope = worker.workerPort;workerPort.onmessage = (e: MessageEvents) => {if (e.data == '父Worker向子Worker发送信息-推荐示例') {// 子Worker线程业务逻辑...console.info('业务执行结束，然后子Worker销毁');workerPort.close();}
}

不推荐使用示例

不建议在父Worker销毁后，子Worker仍向父Worker发送消息。

// main thread
import { worker, MessageEvents, ErrorEvent } from '@kit.ArkTS';const parentworker = new worker.ThreadWorker('entry/ets/workers/parentworker.ets');parentworker.onmessage = (e: MessageEvents) => {console.info('主线程收到父Worker信息' + e.data);
}parentworker.onexit = () => {console.info('父Worker退出');
}parentworker.onAllErrors = (err: ErrorEvent) => {console.error('主线程接收到父Worker报错 ' + err);
}parentworker.postMessage('主线程发送消息给父Worker');

// parentworker.ets
import { ErrorEvent, MessageEvents, ThreadWorkerGlobalScope, worker } from '@kit.ArkTS';const workerPort: ThreadWorkerGlobalScope = worker.workerPort;workerPort.onmessage = (e : MessageEvents) => {console.info('父Worker收到主线程的信息 ' + e.data);let childworker = new worker.ThreadWorker('entry/ets/workers/childworker.ets')childworker.onmessage = (e: MessageEvents) => {console.info('父Worker收到子Worker的信息 ' + e.data);}childworker.onexit = () => {console.info('子Worker退出');workerPort.postMessage('父Worker向主线程发送信息');}childworker.onAllErrors = (err: ErrorEvent) => {console.error('子Worker发生报错 ' + err);}childworker.postMessage('父Worker向子Worker发送信息');// 创建子Worker后，销毁父WorkerworkerPort.close();
}

// childworker.ets
import { ErrorEvent, MessageEvents, ThreadWorkerGlobalScope, worker } from '@kit.ArkTS';const workerPort: ThreadWorkerGlobalScope = worker.workerPort;workerPort.onmessage = (e: MessageEvents) => {console.info('子Worker收到信息 ' + e.data);// 父Worker销毁后，子Worker向父Worker发送信息，行为不可预期workerPort.postMessage('子Worker向父Worker发送信息');setTimeout(() => {workerPort.postMessage('子Worker向父Worker发送信息');}, 1000);
}

不建议在父Worker发起销毁操作的执行阶段创建子Worker。在创建子Worker线程之前，需确保父Worker线程始终处于存活状态，不建议在不确定父Worker是否发起销毁操作的情况下创建子Worker。

// main thread
import { worker, MessageEvents, ErrorEvent } from '@kit.ArkTS';const parentworker = new worker.ThreadWorker('entry/ets/workers/parentworker.ets');parentworker.onmessage = (e: MessageEvents) => {console.info('主线程收到父Worker信息' + e.data);
}parentworker.onexit = () => {console.info('父Worker退出');
}parentworker.onAllErrors = (err: ErrorEvent) => {console.error('主线程接收到父Worker报错 ' + err);
}parentworker.postMessage('主线程发送消息给父Worker');

// parentworker.ets
import { ErrorEvent, MessageEvents, ThreadWorkerGlobalScope, worker } from '@kit.ArkTS';const workerPort: ThreadWorkerGlobalScope = worker.workerPort;workerPort.onmessage = (e : MessageEvents) => {console.info('父Worker收到主线程的信息 ' + e.data);// 父Worker销毁后创建子Worker，行为不可预期workerPort.close();let childworker = new worker.ThreadWorker('entry/ets/workers/childworker.ets');// 子Worker线程未确认创建成功前销毁父Worker，行为不可预期// let childworker = new worker.ThreadWorker('entry/ets/workers/childworker.ets');// workerPort.close();childworker.onmessage = (e: MessageEvents) => {console.info('父Worker收到子Worker的信息 ' + e.data);}childworker.onexit = () => {console.info('子Worker退出');workerPort.postMessage('父Worker向主线程发送信息');}childworker.onAllErrors = (err: ErrorEvent) => {console.error('子Worker发生报错 ' + err);}childworker.postMessage('父Worker向子Worker发送信息');
}

// childworker.ets
import { ErrorEvent, MessageEvents, ThreadWorkerGlobalScope, worker } from '@kit.ArkTS';const workerPort: ThreadWorkerGlobalScope = worker.workerPort;workerPort.onmessage = (e: MessageEvents) => {console.info('子Worker收到信息 ' + e.data);
}

TaskPool和Worker的对比

TaskPool和Worker的作用是为应用程序提供一个多线程的运行环境，用于处理耗时的计算任务或其他密集型任务。可以避免任务阻塞宿主线程，从而提高系统性能和资源利用率。

实现特点对比

表1 TaskPool和Worker的实现特点对比

实现	TaskPool	Worker
内存模型	线程间隔离，内存不共享。	线程间隔离，内存不共享。
参数传递机制	采用标准的结构化克隆算法（Structured Clone）进行序列化、反序列化，完成参数传递。支持ArrayBuffer转移、SharedArrayBuffer共享和Sendable引用传递。	采用标准的结构化克隆算法（Structured Clone）进行序列化、反序列化，完成参数传递。支持ArrayBuffer转移、SharedArrayBuffer共享和Sendable引用传递。
参数传递	直接传递，无需封装。	消息对象唯一参数，需要自己封装。
方法调用	直接将@Concurrent修饰的方法传入并调用。	在Worker线程中解析消息并调用对应方法。
返回值	异步调用后默认返回。	主动发送消息，需在onmessage中解析并赋值。
生命周期	TaskPool自行管理生命周期，无需关心任务负载高低。	开发者自行管理Worker的数量及生命周期。
任务池个数上限	自动管理，无需配置。	同个进程下，最多支持同时开启64个Worker线程，实际数量由进程内存决定。
任务执行时长上限	3分钟（不包含Promise和async/await异步调用的耗时，例如网络下载、文件读写等I/O任务的耗时），长时任务无执行时长上限。	无限制。
设置任务的优先级	支持配置任务优先级。	从API version 18 开始，支持配置Worker线程的优先级。
执行任务的取消	支持取消已经发起的任务。	不支持。
线程复用	支持。	不支持。
任务延时执行	支持。	不支持。
设置任务依赖关系	支持。	不支持。
串行队列	支持。	不支持。
任务组	支持。	不支持。
周期任务	支持。	不支持。
异步队列	支持。	不支持。

适用场景对比

TaskPool和Worker均支持多线程并发能力。由于TaskPool的工作线程会绑定系统的调度优先级，并支持负载均衡（自动扩缩容），相比之下，Worker需要开发者自行创建，存在创建耗时。因此，性能方面TaskPool优于Worker，推荐在大多数场景中使用TaskPool。

TaskPool偏向独立任务维度，该任务在线程中执行，无需关注线程的生命周期，超长任务（大于3分钟且非长时任务）会被系统自动回收。而Worker偏向线程的维度，支持长时间占据线程执行，需要开发者主动管理线程生命周期。

常见开发场景及适用说明如下：

运行时间超过3分钟的任务（不包括Promise和async/await异步调用的耗时，如网络下载、文件读写等I/O任务的耗时）：例如后台进行1小时的预测算法训练等CPU密集型任务，需要使用Worker。场景示例可参考常驻任务开发指导。
有关联的一系列同步任务：例如在一些需要创建、使用句柄的场景中，句柄每次创建都是不同的，该句柄需永久保存，保证使用该句柄进行操作，需要使用Worker。场景示例可参考使用Worker处理关联的同步任务。
需要设置优先级的任务：在API version 18 之前，Worker不支持设置调度优先级，需要使用TaskPool。从API version 18 开始，Worker支持设置调度优先级，开发者可以根据使用场景和任务特性选择使用TaskPool或Worker。例如图库直方图绘制场景，后台计算的直方图数据会用于前台界面的显示，影响用户体验，需要高优先级处理，且任务相对独立，推荐使用TaskPool。
需要频繁取消的任务：例如图库大图浏览场景，为提升体验，会同时缓存当前图片左右侧各2张图片，往一侧滑动跳到下一张图片时，要取消另一侧的一个缓存任务，需要使用TaskPool。
大量或调度点分散的任务：例如大型应用的多个模块包含多个耗时任务，不方便使用Worker去做负载管理，推荐使用TaskPool。场景示例可参考批量数据写数据库场景。