HarmonyOS鸿蒙多线程并发之TaskPool快速上手

多线程并发

对于刚接触多线程编程的新手来说，理解进程、线程、并发和并行等基础概念是至关重要的。在深入探讨多线程并发之前，让我们先统一这些基础概念的认识，为后续学习打下坚实的基础。

基本概念

进程

• 进程是操作系统分配资源（CPU、内存、文件等）和独立运行程序的基本单位，每个进程拥有隔离的内存空间和系统资源，是正在执行的程序的实例。

线程

• 线程是操作系统能够调度的最小执行单元，是进程（Process）中的一个独立控制流。

• 多线程 是指一个进程中运行多个线程，共享进程资源（如文件句柄、内存），但各自独立执行。

并发和并行

• 并发：多个任务交替执行（单核 CPU 分时调度），看似“同时”运行，实则
• 并行：多个任务真正同时执行（需多核 CPU 或多台机器）

[!TIP]

学习多线程的核心目的是 通过并发执行任务来提升程序性能与响应能力，充分利用CPU的计算能力。

线程并发模型

线程并发模型是协调多线程执行的编程范式，旨在提升性能但需解决竞态与死锁问题。常见的并发模型有基于内存共享的模型和基于消息通信的模型。

内存共享模型

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

消息通信模型

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

Actor并发模型与内存共享并发模型相比，不同线程间的内存是隔离的，因此不会发生线程竞争同一内存资源的情况。无需处理内存上锁问题，从而提高开发效率。Actor并发模型中，不同Actor之间不共享内存，需通过消息传递机制传递任务和结果。

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

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

[!Tip]

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

TaskPool

任务池（taskpool）的作用是为应用程序提供多线程运行环境，降低资源消耗并提升系统性能，且您无需关心线程实例的生命周期。您可以使用任务池API创建后台任务（Task），并进行如执行任务或取消任务等操作。理论上，任务池API允许创建的任务数量不受限制，但由于内存限制，不建议这样做。此外，不建议在任务中执行阻塞操作，尤其是无限期阻塞操作，因为长时间的阻塞操作会占用工作线程，可能阻塞其他任务的调度，影响应用性能。

您所创建的同一优先级任务的执行顺序可以由您决定，任务真实执行的顺序与您调用任务池API提供的任务执行接口顺序一致。任务默认优先级是MEDIUM。

当同一时间待执行的任务数量大于任务池工作线程数量，任务池会根据负载均衡机制进行扩容，增加工作线程数量，减少整体等待时长。同样，当执行的任务数量减少，工作线程数量大于执行任务数量，部分工作线程处于空闲状态，任务池会根据负载均衡机制进行缩容，减少工作线程数量。

TaskPool运作机制

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

TaskPool运作机制示意图

@Concurrent装饰器

@Concurrent用于装饰一个函数为并发函数，使用Taskpool时，执行的并发函数必须用该装饰器修饰，否则无法通过校验。

@Concurrent允许装饰普通函数和async函数，禁止标注箭头函数、类方法。不支持类成员函数或者匿名函数。

示例1：普通并发函数

import { taskpool } from '@kit.ArkTS'

//示例：并发函数为一个计算两数之和的普通函数，`taskpool`执行该函数并返回结果。
@Concurrent
function concurrentFunc(num1: number, num2: number): number {
return num1 + num2;
}

//执行并发函数，方式一
taskpool.execute<[number, number], number>(concurrentFunc, 1, 1).then((result)=>{
console.log(`task1 resut is ${result}`) //输出：taskpool resut is 2
})

//执行并发函数，方式二
let task = new taskpool.Task(concurrentFunc, 2, 2)
taskpool.execute<[number, number], number>(task).then((result)=>{
console.log(`task1 resut is ${result}`) //输出：taskpool resut is 4
})

//值并发函数，方式3
async function executeConcurrentFunc(){
try {
let result = await taskpool.execute(concurrentFunc, 3, 3)
console.log(`task1 resut is ${result}`) //输出：taskpool resut is 6
} catch (e) {
console.log(`task1 error is ${e}`) //如果并发函数执行错误，会抛出异常，被catch捕获
}
}

executeConcurrentFunc()

输出结果如下

task1 resut is 2
task1 resut is 4
task1 resut is 6

示例2：async并发函数

import { taskpool } from '@kit.ArkTS'

//示例：async并发函数
@Concurrent
async function asyncConcurrentFunc(array: number[]): Promise<number> {
console.log(`task2…start`)
let sum = 0
for (let i = 1; i <= array.length; i++) {
console.log(`task2…${i}`)
sum += i
}
console.log(`task2…end`)
return await Promise.resolve(sum) //await作用：等待执行完后再返回结果. 此处不能直接返回 Promise，否则会出异常
}

//执行async并发函数
taskpool.execute<[number[]], number>(asyncConcurrentFunc,[1,2,3,4,5,6]).then((result) => {
console.log(`task2 resut is ${result}`) //输出：task2 resut is 7
})
.catch((error: Error) => {
console.log(`task2 error is ${error}`) //
})

输出结果如下

task2…start
task2...1
task2...2
…
task2...100
task2…end
task2 resut is 21