TaskPool和Worker的对比 (TaskPool和Worker)

TaskPool（任务池）和Worker的作用是为应用程序提供一个多线程的运行环境，用于处理耗时的计算任务或其他密集型任务。可以有效地避免这些任务阻塞主线程，从而最大化系统的利用率，降低整体资源消耗，并提高系统的整体性能。

本文将从实现特点和适用场景两个方面来进行TaskPool与Worker的比较。

实现特点对比

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

适用场景对比

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

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

常见的一些开发场景及适用具体说明如下：

• 运行时间超过3分钟（不包含Promise和async/await异步调用的耗时，例如网络下载、文件读写等I/O任务的耗时）的任务。例如后台进行1小时的预测算法训练等CPU密集型任务，需要使用Worker。

• 有关联的一系列同步任务。例如在一些需要创建、使用句柄的场景中，句柄创建每次都是不同的，该句柄需永久保存，保证使用该句柄进行操作，需要使用Worker。

• 需要设置优先级的任务。例如图库直方图绘制场景，后台计算的直方图数据会用于前台界面的显示，影响用户体验，需要高优先级处理，需要使用TaskPool。

• 需要频繁取消的任务。例如图库大图浏览场景，为提升体验，会同时缓存当前图片左右侧各2张图片，往一侧滑动跳到下一张图片时，要取消另一侧的一个缓存任务，需要使用TaskPool。

• 大量或者调度点较分散的任务。例如大型应用的多个模块包含多个耗时任务，不方便使用Worker去做负载管理，推荐采用TaskPool。

CPU密集型任务开发指导 (TaskPool和Worker)

CPU密集型任务是指需要占用系统资源处理大量计算能力的任务，需要长时间运行，这段时间会阻塞线程其它事件的处理，不适宜放在主线程进行。例如图像处理、视频编码、数据分析等。

基于多线程并发机制处理CPU密集型任务可以提高CPU利用率，提升应用程序响应速度。

当任务不需要长时间（3分钟）占据后台线程，而是一个个独立的任务时，推荐使用TaskPool，反之推荐使用Worker。接下来将以图像直方图处理以及后台长时间的模型预测任务分别进行举例。

使用TaskPool进行图像直方图处理

1. 实现图像处理的业务逻辑。

2. 数据分段，通过任务组发起关联任务调度。

创建 TaskGroup 并通过 addTask() 添加对应的任务，通过 execute() 执行任务组，并指定为 高优先级 ，在当前任务组所有任务结束后，会将直方图处理结果同时返回。

3. 结果数组汇总处理。

import { taskpool } from '@kit.ArkTS';

@Concurrent
function imageProcessing(dataSlice: ArrayBuffer): ArrayBuffer {
  // 步骤1: 具体的图像处理操作及其他耗时操作
  return dataSlice;
}

function histogramStatistic(pixelBuffer: ArrayBuffer): void {
  // 步骤2: 分成三段并发调度
  let number: number = pixelBuffer.byteLength / 3;
  let buffer1: ArrayBuffer = pixelBuffer.slice(0, number);
  let buffer2: ArrayBuffer = pixelBuffer.slice(number, number * 2);
  let buffer3: ArrayBuffer = pixelBuffer.slice(number * 2);

  let group: taskpool.TaskGroup = new taskpool.TaskGroup();
  group.addTask(imageProcessing, buffer1);
  group.addTask(imageProcessing, buffer2);
  group.addTask(imageProcessing, buffer3);

  taskpool.execute(group, taskpool.Priority.HIGH).then((ret: Object) => {
    // 步骤3: 结果数组汇总处理
  })
}

@Entry
@Component
struct Index {
  @State message: string = 'Hello World'

  build() {
    Row() {
      Column() {
        Text(this.message)
          .fontSize(50)
          .fontWeight(FontWeight.Bold)
          .onClick(() => {
            let buffer: ArrayBuffer = new ArrayBuffer(24);
            histogramStatistic(buffer);
          })
      }
      .width('100%')
    }
    .height('100%')
  }
}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46

使用Worker进行长时间数据分析

本文通过某地区提供的房价数据训练一个简易的房价预测模型，该模型支持通过输入房屋面积和房间数量去预测该区域的房价，模型需要长时间运行，房价预测需要使用前面的模型运行结果，因此需要使用Worker。

1. DevEco Studio提供了Worker创建的模板，新建一个Worker线程，例如命名为“MyWorker”。

2. 在主线程中通过调用ThreadWorker的 constructor() 方法创建Worker对象，当前线程为宿主线程。

// Index.ets
import { worker } from '@kit.ArkTS';

const workerInstance: worker.ThreadWorker = new worker.ThreadWorker('entry/ets/workers/MyWorker.ts');

• 1
• 2
• 3
• 4

3. 在宿主线程中通过调用 onmessage() 方法接收Worker线程发送过来的消息，并通过调用 postMessage() 方法向Worker线程发送消息。

例如向Worker线程发送训练和预测的消息，同时接收Worker线程发送回来的消息。

// Index.ets
let done = false;

// 接收Worker子线程的结果
workerInstance.onmessage = (() => {
  console.info('MyWorker.ts onmessage');
  if (!done) {
    workerInstance.postMessage({ 'type': 1, 'value': 0 });
    done = true;
  }
})

workerInstance.onerror = (() => {
  // 接收Worker子线程的错误信息
})

// 向Worker子线程发送训练消息
workerInstance.postMessage({ 'type': 0 });

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18

4. 在MyWorker.ts文件中绑定Worker对象，当前线程为Worker线程。

// MyWorker.ts
import { worker, ThreadWorkerGlobalScope, MessageEvents, ErrorEvent } from '@kit.ArkTS';

let workerPort: ThreadWorkerGlobalScope = worker.workerPort;

• 1
• 2
• 3
• 4

5. 在Worker线程中通过调用 onmessage() 方法接收宿主线程发送的消息内容，并通过调用 postMessage() 方法向宿主线程发送消息。

例如在Worker线程中定义预测模型及其训练过程，同时与主线程进行信息交互。

// MyWorker.ts
// 定义训练模型及结果
let result: Array;
// 定义预测函数
function predict(x: number): number {
 return result[x];
}
// 定义优化器训练过程
function optimize(): void {
 result = [0];
}
// Worker线程的onmessage逻辑
workerPort.onmessage = (e: MessageEvents): void => {
 // 根据传输的数据的type选择进行操作
 switch (e.data.type as number) {
  case 0:
  // 进行训练
   optimize();
  // 训练之后发送主线程训练成功的消息
   workerPort.postMessage({ type: 'message', value: 'train success.' });
   break;
  case 1:
  // 执行预测
   const output: number = predict(e.data.value as number);
  // 发送主线程预测的结果
   workerPort.postMessage({ type: 'predict', value: output });
   break;
  default:
   workerPort.postMessage({ type: 'message', value: 'send message is invalid' });
   break;
 }
}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32

6. 在Worker线程中完成任务之后，执行Worker线程销毁操作。销毁线程的方式主要有两种：根据需要可以在宿主线程中对Worker线程进行销毁；也可以在Worker线程中主动销毁Worker线程。

在宿主线程中通过调用 onexit() 方法定义Worker线程销毁后的处理逻辑。

// Worker线程销毁后，执行onexit回调方法
workerInstance.onexit = (): void => {
 console.info("main thread terminate");
}

• 1
• 2
• 3
• 4

方式一：在宿主线程中通过调用 terminate() 方法销毁Worker线程，并终止Worker接收消息。

// 销毁Worker线程
workerInstance.terminate();

• 1
• 2

方式二：在Worker线程中通过调用 close() 方法主动销毁Worker线程，并终止Worker接收消息。

// 销毁线程
workerPort.close();

• 1
• 2

鸿蒙全栈开发全新学习指南

为了积极培养鸿蒙生态人才，让大家都能学习到鸿蒙开发最新的技术，针对一些在职人员、0基础小白、应届生/计算机专业、鸿蒙爱好者等人群，整理了一套纯血版鸿蒙（HarmonyOS Next）全栈开发技术的学习路线【包含了大厂APP实战项目开发】。

本路线共分为四个阶段：

第一阶段：鸿蒙初中级开发必备技能

第二阶段：鸿蒙南北双向高工技能基础：gitee.com/MNxiaona/733GH

第三阶段：应用开发中高级就业技术

第四阶段：全网首发-工业级南向设备开发就业技术：gitee.com/MNxiaona/733GH

鸿蒙开发面试真题（含参考答案）:gitee.com/MNxiaona/733GH

写在最后

• 如果你觉得这篇内容对你还蛮有帮助，我想邀请你帮我三个小忙：
• 点赞，转发，有你们的 『点赞和评论』，才是我创造的动力。
• 关注小编，同时可以期待后续文章ing?，不定期分享原创知识。
• 想要获取更多完整鸿蒙最新学习资源，请移步前往小编：gitee.com/MNxiaona/733GH