HarmonyOS7 主线程扛不住了怎么办？TaskPool 和 Worker 并行实战

文章目录前言TaskPool 和 Worker 的定位Concurrent 注解的规则TaskPool 实战图片批量压缩取消任务Worker 实战数据流实时处理TaskPool 的任务编排选型的经验前言鸿蒙的 UI 跑在主线程上这一点和 Android、iOS 一样。只要你在主线程做了耗时操作——大数据解析、图片处理、复杂计算——界面就开始掉帧。用户滑动列表卡一下点开详情页顿一下体验直接打折。HarmonyOS 提供了两套多线程方案TaskPool和Worker。很多开发者分不清该用哪个这篇把两者的定位讲清楚再给两个实战案例。TaskPool 和 Worker 的定位TaskPool适合一次性任务。你把一个函数扔进去它在线程池里找个空闲线程跑完把结果返回。不用管线程创建和销毁框架全包了。Worker适合长期驻留的后台线程。你创建一个 Worker它有自己的生命周期可以反复和主线程通信。适合需要持续处理数据的场景比如实时数据流处理、WebSocket 长连接的数据解析。简单说任务跑完就走用 TaskPool任务需要长期存在用 Worker。90% 的场景 TaskPool 就够用。Concurrent 注解的规则TaskPool 和 Worker 都需要用Concurrent装饰器标记函数。这个注解告诉编译器这个函数会在别的线程执行。有几条硬性限制函数必须是普通函数不能是箭头函数、不能是类方法函数体里不能引用外部变量闭包不行参数和返回值必须是可序列化的基本类型、ArrayBuffer、Sendable 对象不能用Sendable之外的自定义类// 正确 ✓ConcurrentfunctionprocessData(data:ArrayBuffer):number[]{// 只能在这里使用 data 参数不能引用外部变量constresult:number[][]// ... 处理逻辑returnresult}// 错误 ✗ 引用了外部变量letglobalConfig{threshold:0.5}ConcurrentfunctionprocessData(data:number[]):number[]{returndata.filter(vvglobalConfig.threshold)// 编译报错}这些限制看着烦但理解了就好函数会被序列化到另一个线程执行它必须自给自足。TaskPool 实战图片批量压缩场景用户选了 20 张图片要上传需要先压缩到 500KB 以下。在主线程做会卡死 UI用 TaskPool 并行处理。先定义压缩函数import{image}fromkit.ImageKitConcurrentfunctioncompressImage(buffer:ArrayBuffer,maxWidth:number,quality:number):ArrayBuffer{// 解码constpixelMapimage.createPixelMapFromSurface(buffer)constinfopixelMap.getImageInfoSync()// 计算缩放比例constratioMath.min(1,maxWidth/info.size.width)constnewWidthMath.floor(info.size.width*ratio)constnewHeightMath.floor(info.size.height*ratio)// 缩放pixelMap.scaleSync(newWidth,newHeight)// 压缩编码constpackerimage.createImagePacker()constresultpacker.packingSync(pixelMap,{format:image/jpeg,quality:quality})pixelMap.release()packer.release()returnresult}在 UI 层调用 TaskPool 并行执行import{taskpool}fromkit.ArkTSComponentstruct ImageUploader{StateselectedImages:ArrayBuffer[][]Stateprogress:string0/0StateisProcessing:booleanfalseasynccompressAll(){this.isProcessingtrueconsttotalthis.selectedImages.lengthletcompleted0constresults:ArrayBuffer[][]// 创建所有任务consttasksthis.selectedImages.map(buffer{returntaskpool.execute(compressImage,buffer,1080,// maxWidth80// quality)asPromiseArrayBuffer})// 并行执行逐个收集结果for(consttaskoftasks){try{constresultawaittask results.push(result)completedthis.progress${completed}/${total}}catch(err){Logger.error(Compress failed,err)completedthis.progress${completed}/${total}}}this.isProcessingfalse// results 里就是压缩后的图片数据awaitthis.uploadImages(results)}build(){Column({space:16}){Button(this.isProcessing?压缩中${this.progress}:开始压缩).enabled(!this.isProcessing).onClick(()this.compressAll())}}}TaskPool 会自动把任务分配到多个线程充分利用多核 CPU。20 张图片可能同时有 4-6 张在并行压缩比串行快好几倍。取消任务用户点了取消或者页面退出了正在跑的任务要能停掉// 创建任务时拿到 Task 对象consttasknewtaskpool.Task(compressImage,buffer,1080,80)taskpool.execute(task)// 取消taskpool.cancel(task)注意cancel只能取消还没开始执行的任务。已经在跑的取消不掉只能等它跑完。如果确实需要中断正在执行的逻辑得在函数内部加检查点ConcurrentfunctionlongRunningTask(data:number[],signal:boolean[]):number[]{constresult:number[][]for(leti0;idata.length;i){// 每处理 100 个检查一下取消信号if(i%1000signal[0]){break}result.push(data[i]*2)}returnresult}这里signal用数组包一层是因为要满足可序列化的要求Sendable的SharedArrayBuffer也可以实现类似效果。Worker 实战数据流实时处理场景App 通过蓝牙连接传感器每秒收到几十条数据需要做滤波和聚合后再更新 UI。这种持续的数据流处理适合用 Worker。先创建 Worker 文件DataProcessor.etsimport{worker}fromkit.ArkTSconstworkerPortworker.workerPort// 滑动窗口用于滤波letwindow:number[][]constWINDOW_SIZE10workerPort.onmessage(event:MessageEvents){constmsgevent.dataif(msg.typedata){constrawValue:numbermsg.value// 滑动窗口滤波window.push(rawValue)if(window.lengthWINDOW_SIZE){window.shift()}// 计算窗口均值constavgwindow.reduce((a,b)ab,0)/window.length// 检测异常值偏离均值超过 3 倍标准差constvariancewindow.reduce((sum,v)sumMath.pow(v-avg,2),0)/window.lengthconststdDevMath.sqrt(variance)constisAnomalyMath.abs(rawValue-avg)stdDev*3// 回传处理结果workerPort.postMessage({type:processed,value:avg,raw:rawValue,isAnomaly:isAnomaly,timestamp:Date.now()})}elseif(msg.typereset){window[]workerPort.postMessage({type:reset_done})}}// Worker 初始化完成通知workerPort.postMessage({type:ready})主线程里创建和使用 Workerimport{worker}fromkit.ArkTSComponentstruct SensorDashboard{privateworkerInstance:worker.ThreadWorker|nullnullStatecurrentValue:number0StateisAnomaly:booleanfalseStatedataHistory:number[][]aboutToAppear(){// 创建 Workerthis.workerInstancenewworker.ThreadWorker(entry/ets/workers/DataProcessor.ets)this.workerInstance.onmessage(event:MessageEvents){constmsgevent.dataif(msg.typeready){Logger.info(Worker is ready)}elseif(msg.typeprocessed){this.currentValuemsg.valuethis.isAnomalymsg.isAnomalythis.dataHistory.push(msg.value)// 只保留最近 100 个数据点用于图表if(this.dataHistory.length100){this.dataHistory.shift()}}}}aboutToDisappear(){// 页面销毁时终止 Workerthis.workerInstance?.terminate()this.workerInstancenull}// 蓝牙数据回调里把原始数据推给 WorkeronBluetoothDataReceived(value:number){this.workerInstance?.postMessage({type:data,value:value})}build(){Column({space:12}){Text(当前值:${this.currentValue.toFixed(2)}).fontSize(24).fontColor(this.isAnomaly?#FF4D4F:#333333)Text(this.isAnomaly?⚠ 检测到异常值:数据正常).fontColor(this.isAnomaly?#FF4D4F:#52C41A)// 这里可以放一个 Canvas 折线图参考第 29 篇// LineChart({ data: this.dataHistory, ... })}.padding(20)}}Worker 的关键优势在于它一直活着。不像 TaskPool 每次任务都要调度Worker 创建后就常驻后台数据来一条处理一条没有调度开销。对于高频数据流这个差异很明显。TaskPool 的任务编排TaskPool 还有个实用功能——可以控制任务的优先级和执行顺序// 设置优先级HIGH 优先执行consthighPriorityTasknewtaskpool.Task(importantWork,data)taskpool.execute(highPriorityTask,taskpool.Priority.HIGH)// Promise.all 并行等待所有任务完成constresultsawaitPromise.all([taskpool.execute(taskA,data1),taskpool.execute(taskB,data2),taskpool.execute(taskC,data3),])// 串行执行前一个完成再跑下一个constresult1awaittaskpool.execute(step1,input)constresult2awaittaskpool.execute(step2,result1)constfinalResultawaittaskpool.execute(step3,result2)串行执行虽然不是并行但把每步放到 TaskPool 里跑至少不会阻塞主线程UI 还是流畅的。选型的经验能用 TaskPool 就别用 Worker。TaskPool 用起来简单不用管生命周期线程池自动管理适合绝大多数场景。Worker 只在这些场景用需要维护内部状态的后台任务、高频数据流处理、需要长连接的后台处理。别在多线程里操作 UI。不管是 TaskPool 还是 Worker都只能处理数据不能碰 UI 组件。处理完把结果传回主线程主线程更新State驱动 UI 刷新。传参尽量用 ArrayBuffer。大数据量传输时对象序列化开销不小。能用ArrayBuffer传就用ArrayBuffer零拷贝快得多。多线程这块不难难的是判断要不要用。很多时候你觉得需要多线程其实是因为数据结构没设计好或者做了不必要的重复计算。先优化主线程逻辑真卡了再上多线程。