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