1# CPU密集型任务开发指导 (TaskPool和Worker) 2 3 4CPU密集型任务是指需要占用系统资源处理大量计算能力的任务,需要长时间运行,这段时间会阻塞线程其它事件的处理,不适宜放在UI主线程进行。例如图像处理、视频编码、数据分析等。 5 6 7基于多线程并发机制处理CPU密集型任务可以提高CPU利用率,提升应用程序响应速度。 8 9 10当任务不需要长时间(3分钟)占据后台线程,而是一个个独立的任务时,推荐使用TaskPool,反之推荐使用Worker。 11 12接下来将以图像直方图处理以及后台长时间的模型预测任务分别进行举例。 13 14 15## 使用TaskPool进行图像直方图处理 16 171. 实现图像处理的业务逻辑。 18 192. 数据分段,通过任务组发起关联任务调度。 20 创建[TaskGroup](../reference/apis-arkts/js-apis-taskpool.md#taskgroup10)并通过[addTask()](../reference/apis-arkts/js-apis-taskpool.md#addtask10)添加对应的任务,通过[execute()](../reference/apis-arkts/js-apis-taskpool.md#taskpoolexecute10)执行任务组,并指定为[高优先级](../reference/apis-arkts/js-apis-taskpool.md#priority),在当前任务组所有任务结束后,会将直方图处理结果同时返回。 21 223. 结果数组汇总处理。 23 24```ts 25import { taskpool } from '@kit.ArkTS'; 26 27@Concurrent 28function imageProcessing(dataSlice: ArrayBuffer): ArrayBuffer { 29 // 步骤1: 具体的图像处理操作及其他耗时操作 30 return dataSlice; 31} 32 33function histogramStatistic(pixelBuffer: ArrayBuffer): void { 34 // 步骤2: 分成三段并发调度 35 let number: number = pixelBuffer.byteLength / 3; 36 let buffer1: ArrayBuffer = pixelBuffer.slice(0, number); 37 let buffer2: ArrayBuffer = pixelBuffer.slice(number, number * 2); 38 let buffer3: ArrayBuffer = pixelBuffer.slice(number * 2); 39 40 let group: taskpool.TaskGroup = new taskpool.TaskGroup(); 41 group.addTask(imageProcessing, buffer1); 42 group.addTask(imageProcessing, buffer2); 43 group.addTask(imageProcessing, buffer3); 44 45 taskpool.execute(group, taskpool.Priority.HIGH).then((ret: Object) => { 46 // 步骤3: 结果数组汇总处理 47 }) 48} 49 50@Entry 51@Component 52struct Index { 53 @State message: string = 'Hello World' 54 55 build() { 56 Row() { 57 Column() { 58 Text(this.message) 59 .fontSize(50) 60 .fontWeight(FontWeight.Bold) 61 .onClick(() => { 62 let buffer: ArrayBuffer = new ArrayBuffer(24); 63 histogramStatistic(buffer); 64 }) 65 } 66 .width('100%') 67 } 68 .height('100%') 69 } 70} 71``` 72 73 74## 使用Worker进行长时间数据分析 75 76本文通过某地区提供的房价数据训练一个简易的房价预测模型,该模型支持通过输入房屋面积和房间数量去预测该区域的房价,模型需要长时间运行,房价预测需要使用前面的模型运行结果,因此需要使用Worker。 77 781. DevEco Studio提供了Worker创建的模板,新建一个Worker线程,例如命名为“MyWorker”。 79 80  81 822. 在宿主线程中通过调用ThreadWorker的[constructor()](../reference/apis-arkts/js-apis-worker.md#constructor9)方法创建Worker对象。 83 84 ```ts 85 // Index.ets 86 import { worker } from '@kit.ArkTS'; 87 88 const workerInstance: worker.ThreadWorker = new worker.ThreadWorker('entry/ets/workers/MyWorker.ts'); 89 ``` 90 913. 在宿主线程中通过调用[onmessage()](../reference/apis-arkts/js-apis-worker.md#onmessage9)方法接收Worker线程发送过来的消息,并通过调用[postMessage()](../reference/apis-arkts/js-apis-worker.md#postmessage9)方法向Worker线程发送消息。 92 例如向Worker线程发送训练和预测的消息,同时接收Worker线程发送回来的消息。 93 94 ```ts 95 // Index.ets 96 let done = false; 97 98 // 接收Worker子线程的结果 99 workerInstance.onmessage = (() => { 100 console.info('MyWorker.ts onmessage'); 101 if (!done) { 102 workerInstance.postMessage({ 'type': 1, 'value': 0 }); 103 done = true; 104 } 105 }) 106 107 workerInstance.onerror = (() => { 108 // 接收Worker子线程的错误信息 109 }) 110 111 // 向Worker子线程发送训练消息 112 workerInstance.postMessage({ 'type': 0 }); 113 ``` 114 1154. 在MyWorker.ts文件中绑定Worker对象,当前线程为Worker线程。 116 117 ```ts 118 // MyWorker.ts 119 import { worker, ThreadWorkerGlobalScope, MessageEvents, ErrorEvent } from '@kit.ArkTS'; 120 121 let workerPort: ThreadWorkerGlobalScope = worker.workerPort; 122 ``` 123 1245. 在Worker线程中通过调用[onmessage()](../reference/apis-arkts/js-apis-worker.md#onmessage9-1)方法接收宿主线程发送的消息内容,并通过调用[postMessage()](../reference/apis-arkts/js-apis-worker.md#postmessage9-2)方法向宿主线程发送消息。 125 例如在Worker线程中定义预测模型及其训练过程,同时与宿主线程进行信息交互。 126 127 ```ts 128 // MyWorker.ts 129 // 定义训练模型及结果 130 let result: Array<number>; 131 // 定义预测函数 132 function predict(x: number): number { 133 return result[x]; 134 } 135 // 定义优化器训练过程 136 function optimize(): void { 137 result = [0]; 138 } 139 // Worker线程的onmessage逻辑 140 workerPort.onmessage = (e: MessageEvents): void => { 141 // 根据传输的数据的type选择进行操作 142 switch (e.data.type as number) { 143 case 0: 144 // 进行训练 145 optimize(); 146 // 训练之后发送宿主线程训练成功的消息 147 workerPort.postMessage({ type: 'message', value: 'train success.' }); 148 break; 149 case 1: 150 // 执行预测 151 const output: number = predict(e.data.value as number); 152 // 发送宿主线程预测的结果 153 workerPort.postMessage({ type: 'predict', value: output }); 154 break; 155 default: 156 workerPort.postMessage({ type: 'message', value: 'send message is invalid' }); 157 break; 158 } 159 } 160 ``` 161 1626. 在Worker线程中完成任务之后,执行Worker线程销毁操作。销毁线程的方式主要有两种:根据需要可以在宿主线程中对Worker线程进行销毁;也可以在Worker线程中主动销毁Worker线程。 163 164 在宿主线程中通过调用[onexit()](../reference/apis-arkts/js-apis-worker.md#onexit9)方法定义Worker线程销毁后的处理逻辑。 165 166 ```ts 167 // Worker线程销毁后,执行onexit回调方法 168 workerInstance.onexit = (): void => { 169 console.info("main thread terminate"); 170 } 171 ``` 172 173 方式一:在宿主线程中通过调用[terminate()](../reference/apis-arkts/js-apis-worker.md#terminate9)方法销毁Worker线程,并终止Worker接收消息。 174 175 ```ts 176 // 销毁Worker线程 177 workerInstance.terminate(); 178 ``` 179 180 方式二:在Worker线程中通过调用[close()](../reference/apis-arkts/js-apis-worker.md#close9)方法主动销毁Worker线程,并终止Worker接收消息。 181 182 ```ts 183 // 销毁线程 184 workerPort.close(); 185 ```