CPU密集型任務(wù)開發(fā)指導(dǎo)

使用TaskPool進(jìn)行圖像直方圖處理

1. 實(shí)現(xiàn)圖像處理的業(yè)務(wù)邏輯。

2. 數(shù)據(jù)分段，將各段數(shù)據(jù)通過不同任務(wù)的執(zhí)行完成圖像處理。

   創(chuàng)建Task，通過execute()執(zhí)行任務(wù)，在當(dāng)前任務(wù)結(jié)束后，會(huì)將直方圖處理結(jié)果同時(shí)返回。

3. 結(jié)果數(shù)組匯總處理。

import taskpool from '@ohos.taskpool';

@Concurrent
function imageProcessing(dataSlice: ArrayBuffer) {
  // 步驟1: 具體的圖像處理操作及其他耗時(shí)操作
  return dataSlice;
}

function histogramStatistic(pixelBuffer: ArrayBuffer) {
  // 步驟2: 分成三段并發(fā)調(diào)度
  let number = pixelBuffer.byteLength / 3;
  let buffer1 = pixelBuffer.slice(0, number);
  let buffer2 = pixelBuffer.slice(number, number * 2);
  let buffer3 = pixelBuffer.slice(number * 2);

  let task1 = new taskpool.Task(imageProcessing, buffer1);
  let task2 = new taskpool.Task(imageProcessing, buffer2);
  let task3 = new taskpool.Task(imageProcessing, buffer3);

  taskpool.execute(task1).then((ret: ArrayBuffer[]) => {
    // 步驟3: 結(jié)果處理
  });
  taskpool.execute(task2).then((ret: ArrayBuffer[]) => {
    // 步驟3: 結(jié)果處理
  });
  taskpool.execute(task3).then((ret: ArrayBuffer[]) => {
    // 步驟3: 結(jié)果處理
  });
}

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

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

使用Worker進(jìn)行長時(shí)間數(shù)據(jù)分析

本文通過某地區(qū)提供的房價(jià)數(shù)據(jù)訓(xùn)練一個(gè)簡易的房價(jià)預(yù)測(cè)模型，該模型支持通過輸入房屋面積和房間數(shù)量去預(yù)測(cè)該區(qū)域的房價(jià)，模型需要長時(shí)間運(yùn)行，房價(jià)預(yù)測(cè)需要使用前面的模型運(yùn)行結(jié)果，因此需要使用Worker。

1. DevEco Studio提供了Worker創(chuàng)建的模板，新建一個(gè)Worker線程，例如命名為“MyWorker”。

   

2. 在主線程中通過調(diào)用ThreadWorker的constructor()方法創(chuàng)建Worker對(duì)象，當(dāng)前線程為宿主線程。

   import worker from '@ohos.worker';
   
   const workerInstance = new worker.ThreadWorker('entry/ets/workers/MyWorker.ts');

3. 在宿主線程中通過調(diào)用onmessage()方法接收Worker線程發(fā)送過來的消息，并通過調(diào)用postMessage()方法向Worker線程發(fā)送消息。

   例如向Worker線程發(fā)送訓(xùn)練和預(yù)測(cè)的消息，同時(shí)接收Worker線程發(fā)送回來的消息。

   // 接收Worker子線程的結(jié)果
   workerInstance.onmessage = function(e) {
     // data：Worker線程發(fā)送的信息
     let data = e.data;
     console.info('MyWorker.ts onmessage');
   }
   
   workerInstance.onerror = function (d) {
     // 接收Worker子線程的錯(cuò)誤信息
   }
   
   // 向Worker子線程發(fā)送訓(xùn)練消息
   workerInstance.postMessage({ 'type': 0 });
   // 向Worker子線程發(fā)送預(yù)測(cè)消息
   workerInstance.postMessage({ 'type': 1, 'value': [90, 5] });

4. 在MyWorker.ts文件中綁定Worker對(duì)象，當(dāng)前線程為Worker線程。

   import worker, { ThreadWorkerGlobalScope, MessageEvents, ErrorEvent } from '@ohos.worker';
   
   let workerPort: ThreadWorkerGlobalScope = worker.workerPort;

5. 在Worker線程中通過調(diào)用onmessage()方法接收宿主線程發(fā)送的消息內(nèi)容，并通過調(diào)用postMessage()方法向宿主線程發(fā)送消息。

   例如在Worker線程中定義預(yù)測(cè)模型及其訓(xùn)練過程，同時(shí)與主線程進(jìn)行信息交互。

   import worker, { ThreadWorkerGlobalScope, MessageEvents, ErrorEvent } from '@ohos.worker';
   
   let workerPort: ThreadWorkerGlobalScope = worker.workerPort;
   
   // 定義訓(xùn)練模型及結(jié)果 
   let result;
   
   // 定義預(yù)測(cè)函數(shù)
   function predict(x) {
     return result[x];
   }
   
   // 定義優(yōu)化器訓(xùn)練過程
   function optimize() {
     result = {};
   }
   
   // Worker線程的onmessage邏輯
   workerPort.onmessage = function (e: MessageEvents) {
     let data = e.data
     // 根據(jù)傳輸?shù)臄?shù)據(jù)的type選擇進(jìn)行操作
     switch (data.type) {
       case 0:
       // 進(jìn)行訓(xùn)練
         optimize();
       // 訓(xùn)練之后發(fā)送主線程訓(xùn)練成功的消息
         workerPort.postMessage({ type: 'message', value: 'train success.' });
         break;
       case 1:
       // 執(zhí)行預(yù)測(cè)
         const output = predict(data.value);
       // 發(fā)送主線程預(yù)測(cè)的結(jié)果
         workerPort.postMessage({ type: 'predict', value: output });
         break;
       default:
         workerPort.postMessage({ type: 'message', value: 'send message is invalid' });
         break;
     }
   }

6. 在Worker線程中完成任務(wù)之后，執(zhí)行Worker線程銷毀操作。銷毀線程的方式主要有兩種：根據(jù)需要可以在宿主線程中對(duì)Worker線程進(jìn)行銷毀；也可以在Worker線程中主動(dòng)銷毀Worker線程。

   在宿主線程中通過調(diào)用onexit()方法定義Worker線程銷毀后的處理邏輯。

   // Worker線程銷毀后，執(zhí)行onexit回調(diào)方法
   workerInstance.onexit = function() {
     console.info("main thread terminate");
   }

   方式一：在宿主線程中通過調(diào)用terminate()方法銷毀Worker線程，并終止Worker接收消息。

   // 銷毀Worker線程
   workerInstance.terminate();

   方式二：在Worker線程中通過調(diào)用close()方法主動(dòng)銷毀Worker線程，并終止Worker接收消息。

   // 銷毀線程
   workerPort.close();