鸿蒙多线程应用-taskPool

耶耶耶耶耶  金牌会员 | 2025-2-21 02:26:49 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 948|帖子 948|积分 2844

 并发模型

      并发模型是用来实现差别应用场景中并发任务的编程模型，常见的并发模型分为基于内存共享的并发模型和基于消息通信的并发模型。
        Actor并发模型作为基于消息通信并发模型的典型代表，不需要开发者去面对锁带来的一系列复杂偶发的问题，同时并发度也相对较高，因此得到了广泛的支持和利用。
       当前鸿蒙ArkTS提供了TaskPool和Worker两种并发本领，TaskPool和Worker都基于Actor并发模型实现。
       内存共享并发模型指多线程同时执行任务，这些线程依赖同一内存并且都有权限访问，线程访问内存前需要抢占并锁定内存的利用权，没有抢占到内存的线程需要等待其他线程开释利用权再执行。
       Actor并发模型每一个线程都是一个独立Actor，每个Actor有自己独立的内存，Actor之间通过消息通报机制触发对方Actor的举动，差别Actor之间不能直接访问对方的内存空间。Actor并发模型对比内存共享并发模型的上风在于差别线程间内存隔离，不会产生差别线程竞争同一内存资源的问题。开发者不需要考虑对内存上锁导致的一系列功能、性能问题，提拔了开发效率。
       由于Actor并发模型线程之间不共享内存，需要通过线程间通信机制传输并发任务和任务结果。
TaskPool简介

      任务池（TaskPool）作用是为应用步伐提供一个多线程的运行环境，降低团体资源的斲丧、进步体系的团体性能，且您无需关心线程实例的生命周期。
     TaskPool支持开发者在宿主线程封装任务抛给任务队列，体系选择合适的工作线程，进行任务的分发及执行，再将结果返回给宿主线程。接口直观易用，支持任务的执行、取消，以及指定优先级的本领，同时通过体系统一线程管理，联合动态调理及负载均衡算法，可以节约体系资源。体系默认会启动一个任务工作线程，当任务较多时会扩容，工作线程数量上限跟当前设备的物理核数相关，详细数量内部管理，保证最优的调理及执行效率，长时间没有任务分发时会缩容，减少工作线程数量。
TaskPool留意事项



  • 实现任务的函数需要利用@Concurrent装饰器标注，且仅支持在.ets文件中利用。
  • 从API version 11开始，跨并发实例通报带方法的实例对象时，该类必须利用装饰器@Sendable装饰器标注，且仅支持在.ets文件中利用。
  • 任务函数在TaskPool工作线程的执行耗时不能超过3分钟（不包含Promise和async/await异步调用的耗时，例如网络下载、文件读写等I/O任务的耗时），否则会被逼迫退出。
  • 实现任务的函数入参需满足序列化支持的范例，详情请参见线程间通信对象。
  • ArrayBuffer参数在TaskPool中默认转移，需要设置转移列表的话可通过接口setTransferList()设置。
  • 由于差别线程中上下文对象是差别的，因此TaskPool工作线程只能利用线程安全的库，例如UI相关的非线程安全库不能利用。
  • 序列化传输的数据量巨细限定为16MB。
  • Priority的IDLE优先级是用来标记需要在背景运行的耗时任务（例如数据同步、备份），它的优先级别是最低的。这种优先级标记的任务只会在全部线程都空闲的环境下触发执行，并且只会占用一个线程来执行。
  • Promise不支持跨线程通报，如果TaskPool返回pending或rejected状态的Promise，会返回失败；对于fulfilled状态的Promise，TaskPool会解析返回的结果，如果结果可以跨线程通报，则返回成功。
  • 不支持在TaskPool工作线程中利用AppStorage。
TaskPool应用实例

       生产者消费者模型应用taskPool的详细代码实现
1.生产者
  2. import { taskpool } from '@kit.ArkTS';
  3. import { stingToUint8, uint8TransformString } from './utils';
  5. @Concurrent
  6. export async function producer(ArrayBuffer: Int32Array, dataBuffer: Uint8Array, newStr: string) {
  7.   let i32a = ArrayBuffer;
  8.   let array = dataBuffer
  9.   if (array[array.length-1] !== 0) {
  10.     taskpool.Task.sendData(false)
  11.     let runner = new taskpool.SequenceRunner()
  12.     console.log("-----atomics-producer-push-fal-" + newStr)
  13.     return
  14.   }
  15.   let jsonStr: string = uint8TransformString(array)
  17.   let arr: string[] = []
  18.   try {
  19.     arr= JSON.parse(jsonStr) as string[]
  20.   } catch (e) {
  21.     taskpool.Task.sendData(false)
  22.     return
  23.   }
  25.   arr.push(newStr)
  26.   let newArrJson = JSON.stringify(arr) ?? ''
  27.   //console.log("newArrJson" + newArrJson)
  28.   let isFinish = stingToUint8(newArrJson,array,4)
  29.   if (!isFinish) {
  30.     arr.pop()
  31.     let newArrJson1 = JSON.stringify(arr) ?? ''
  32.     stingToUint8(newArrJson1,array,4)
  33.     taskpool.Task.sendData(false)
  34.     console.log("-----atomics-producer-push-fal-" + newStr)
  35.   }else{
  36.     console.log("-----atomics-producer-push-sec-" + newStr)
  37.   }
  39.   Atomics.notify(i32a, 0, 1)
  40.   Promise.resolve()
  41. }
复制代码
2.消费者
  2. import { getStringArrayFromJson, testMethod, uint8TransformString} from './utils';
  3. import { buffer, taskpool } from '@kit.ArkTS';
  4. import { ThreadUtils } from './ThreadUtils';
  6. @Concurrent
  7. export async function consumerTask(ArrayBuffer: Int32Array, dataBuffer: Uint8Array): Promise<void> {
  8.   let i32a = ArrayBuffer;
  9.   let array = dataBuffer
  11.   while (true) {
  12.     let jsonStr: string = uint8TransformString(array)
  13.     let arr = getStringArrayFromJson(jsonStr)
  15.     if (arr.length == 0) {
  16.       Atomics.wait(i32a, 0, 0);
  17.     } else {
  18.       let i = 4
  19.       for (let index = 0; index < array.byteLength; index++) {
  20.         if (i >= array.byteLength) {
  21.           break
  22.         }
  23.         Atomics.store(array, i++, 0)
  24.       }
  25.       taskpool.Task.sendData(true)
  26.       let writeResult: boolean = true
  27.       while ((writeResult == true || writeResult == false)) {
  28.         let ele = arr.shift()
  29.         if (!ele) {
  30.           break
  31.         }
  32.         writeResult = await ThreadUtils.getInstance().writeToFile(ele)
  33.         console.log('-----atomics-consumer-' + ele)
  34.       }
  36.     }
  37.   }
  38. }
复制代码
3.字符串和字节码相互转换工具
  1. export function testMethod(str: string) {
  2.   console.log('--test-function-str-' + str)
  3. }
  4. export function uint8TransformString(array:Uint8Array): string{
  5.   let jsonStr: string = JSON.stringify([])
  6.   let tempArr: number[] = []
  7.   let j = 0
  8.   for (let index = 0; index < array.length; index++) {
  9.     if (array[index] == 0) {
  10.       continue
  11.     }
  12.     tempArr[j++] = array[index]
  13.   }
  15.   let temp = new Uint8Array(tempArr)
  16.   if (temp.byteLength > 0) {
  17.     let str = '';
  18.     for (let i = 0; i < temp.length; ) {
  19.       let byte1 = temp[i];
  20.       let codePoint: number
  22.       if (byte1 >> 7 === 0) { // 1字节
  23.         codePoint = byte1;
  24.         i += 1;
  25.       } else if (byte1 >> 5 === 0b110) { // 2字节
  26.         codePoint = ((byte1 & 0b11111) << 6) | (temp[i + 1] & 0b111111);
  27.         i += 2;
  28.       } else if (byte1 >> 4 === 0b1110) { // 3字节
  29.         codePoint = ((byte1 & 0b1111) << 12) | ((temp[i + 1] & 0b111111) << 6) | (temp[i + 2] & 0b111111);
  30.         i += 3;
  31.       } else {
  32.         // 错误处理：不支持的字节序列
  33.         i += 1; // 跳过当前字节
  34.         continue;
  35.       }
  36.       str += String.fromCodePoint(codePoint)
  37.       console.info('字节流转成可理解的字符串:' + str);
  39.     }
  40.     jsonStr = str
  41.   }
  44.   return jsonStr
  45. }
  46. //
  47. export function stingToUint8(json: string, array:Uint8Array,formIndex: number = 0) : boolean{
  48.   let i = formIndex
  49.   let isFinish = true
  50.   for (let index = 0; index < json.length; index++) {
  51.     if (i >= array.byteLength) {
  52.       if (index < json.length - 1) {
  53.         isFinish = false
  54.       }
  55.       break
  56.     }
  57.     const element = json.charCodeAt(index);
  58.     if (element > 0x7FF) {
  59.       Atomics.store(array, i++, (0xE0 | (element >> 12)))
  60.       Atomics.store(array, i++, (0x80 | ((element >> 6) & 0x3F)))
  61.       Atomics.store(array, i++, (0x80 | (element & 0x3F)))
  62.     } else if (element > 0x7F) {
  63.       Atomics.store(array, i++, (0xC0 | (element >> 6)))
  64.       Atomics.store(array, i++, (0x80 | (element & 0x3F)))
  65.     } else {
  66.       Atomics.store(array, i++, (element))
  67.     }
  68.   }
  69.   //剩余空间赋值0
  70.   for (let index = i; index < array.length; index++) {
  71.     array[index] = 0
  72.   }
  73.   return isFinish
  74. }
复制代码
4.单例工具
  1. import { taskpool } from '@kit.ArkTS';
  2. import { it } from '@ohos/hypium';
  3. import { consumerTask } from './consumer';
  4. import { producer } from './product';
  6. export class ThreadUtils {
  8.   private tempLogList: Array<string> = new Array()
  10.   private static instance: ThreadUtils
  11.   private sab :SharedArrayBuffer
  12.   private ui8 :Uint8Array
  13.   private i32a :Int32Array
  15.   private constructor(bufferSize:number = 1024) {
  16.     this.sab = new SharedArrayBuffer(bufferSize)
  17.     this.ui8 = new Uint8Array(this.sab)
  18.     this.i32a = new Int32Array(this.sab)
  19.     this.startConsumer()
  20.   };
  22.   writeLog(log: string) {
  23.     if (this.flag) {
  24.       this.tempLogList.push(log)
  25.     }else {
  26.       this.product(log)
  27.     }
  29.   }
  31.   public static getInstance(bufferSize:number = 1024): ThreadUtils {
  32.     if (!ThreadUtils.instance) {
  33.       ThreadUtils.instance = new ThreadUtils(bufferSize);
  34.     }
  35.     return ThreadUtils.instance;
  36.   }
  39.   async writeToFile(content: string): Promise<boolean> {
  40.     return new Promise((resolve, reject) => {
  41.       setTimeout(() => {
  43.         console.log("日志写入完成=" + content)
  44.         console.log('pop element=' + content)
  45.         resolve(true)
  46.       }, 4000)
  47.     })
  48.   }
  50.   lastTask:taskpool.Task | undefined
  51.   flag = false
  52.   async product(log: string):Promise<boolean> {
  53.     return new Promise<boolean>((resolve,reject)=>{
  54.       let newLog = log
  55.       let task = new taskpool.Task(producer, this.i32a, this.ui8, newLog)
  56.       if (this.lastTask) {
  57.         task.addDependency(this.lastTask)
  58.       }
  60.       this.lastTask = task
  61.       task.onReceiveData((success: boolean) => {
  62.         if (!success) {
  63.           this.flag = true
  64.           this.tempLogList.unshift(log)
  65.           resolve(false)
  66.         }
  67.       })
  68.       taskpool.execute(task).then(()=>{
  69.         console.log('------taskpool.execute.then-----')
  70.         resolve(true)
  71.       });
  72.     })
  75.   }
  77.   isWhile = false
  78.   async startConsumer() {
  79.     let task = new taskpool.Task(consumerTask, this.i32a, this.ui8)
  80.     task.onReceiveData(async (hasSpace: boolean) => {
  81.       if (hasSpace) {
  82.         this.flag = false
  83.         if (this.tempLogList.length > 0 && this.isWhile == false){
  84.           let item = this.tempLogList.shift()
  85.           console.log('---item---'+ item)
  86.           this.isWhile = true
  87.           let com = true
  88.           while (item && this.flag == false && com){
  89.             com = await this.product(item)
  90.             item = this.tempLogList.shift()
  91.           }
  92.           this.isWhile = false
  93.         }
  95.       }
  96.     })
  97.     taskpool.execute(task)
  98.   }
  100. }
复制代码
5.页面UI应用
  1. import { buffer, taskpool } from '@kit.ArkTS';
  2. import { consumerTask } from '../consumer';
  3. import { producer } from '../product';
  4. import { router } from '@kit.ArkUI';
  5. import { ThreadUtils } from '../ThreadUtils';
  8. @Entry
  9. @Component
  10. struct Index {
  11.   timer = -1
  12.   count = 0
  13.   logTool = ThreadUtils.getInstance(32)
  14.   aboutToAppear(): void {
  16.   }
  20.   @State inputText:string =''
  21.   build() {
  22.     Column({space: 20}) {
  24.       TextInput({text: $$this.inputText})
  25.         .width('80%')
  26.       Button() {
  27.         Text("生产日志").padding(10)
  28.       }
  29.       .backgroundColor(Color.Gray)
  30.       .onClick(async () => {
  32.         this.timer = setInterval(()=>{
  33.           this.logTool.writeLog ('item' + this.count)
  34.           this.count += 1
  35.         },1000)
  37.       })
  38.       Button() {
  39.         Text("停止生产").padding(10)
  40.       }
  41.       .backgroundColor(Color.Gray)
  42.       .onClick(async () => {
  43.         clearInterval(this.timer)
  44.         // router.pushUrl({
  45.         //   url: 'pages/TaskPoolPage'
  46.         // })
  48.       })
  50.     }
  51.     .alignItems(HorizontalAlign.Center)
  52.     .justifyContent(FlexAlign.Center)
  53.     .height('100%')
  54.     .width('100%')
  55.   }
  56. }
复制代码


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

耶耶耶耶耶

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

运维 CIO 存储 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表