• js异步控制器，实现异步同步代码交替穿插执行

    一、js代码中发起请求，读取文件等操作都是异步执行，如果我们希望将多个异步函数的结果都收集起来，统一处理，这应该怎么做呢？

    第一种方法，需要用async和await关键字将异步函数强制为同步状态，这样就可以同步获取每一个异步函数的结果了。（耗时长）

    第二种就是使用Promise对象，将每个函数都装到Promise对象中，按照顺序放到.then方法中，变成串行执行，（耗时长）

    第三种方法，需要定义全局变量，标记每个异步函数的执行结果状态，定义全局容器收集异步函数执行结果，定义全局检查函数检查执行结果状态，然后在每一个异步函数的完成的时候修改该状态，并且调用检查方法，在全部结束的时候表示异步函数结果收集完成。（代码复杂度高）

    以上方式即可解决

    二、如果希望多个异步并行，全部结束之后又需要连续几个异步串行，之后又希望多个异步函数并行，又希望多个异步串行等等，这样的话，通过上面的方法，也可以解决，但代码都会变得特别复杂，难以维护，怎么办？

    于是乎，想到应该得有这么一个异步函数控制器，想串行就串行，想并行就并行，而且代码复杂度还得降低，网上搜索半天没有找到现成的代码，或者就是晦涩难懂。

    思路如下：

    每一个异步函数都看做成一个任务，当然同步函数亦可，这里借鉴批量程序的任务单元概念：

    一个任务定义简单的生命周期，即任务初始化，任务执行中，任务结束。

    这里先定义好枚举常量，增强代码易读性

    1. /**
    2.  * 任务执行状态枚举类
    3.  * @type {{INIT: number, RUNNING: number, FINISH: number}}
    4.  */
    5. const TaskStatusEnum = {
    6.     INIT: -1, //初始化
    7.     RUNNING: 0, //执行中
    8.     FINISH: 1 //已完成
    9. }

    我们现在面向异步函数，如何知道异步函数这个任务什么时候结束？代码执行完最后一行就算结束？不对吧，代码执行完最后一行，异步函数似乎并没有结束。

    而且js中也没办法去检测一个函数到底有没有结束，就算可以，那什么时机去检测呢？定时轮训可以不，异步的不确定性，定多少时间合适呢？1秒检测一次？还是1毫秒一次？检测频率越慢，这中间会造成延迟越大，检测越快，越影响代码执行效率，因为js是单线程。

    我在刚学js的时候，会碰到这些疑问，原因是习惯性按照同步的思路去思考问题。

    但js有个概念：事件驱动。比如：根据工作任务分配原则，A同事的工作完成后，我们要让A主动汇报自己已经完成工作了，而不是每次让我们一次又一次观察A的情况。A的汇报就是事件驱动，我们给A交代了一个事情（每次工作完要主动汇报）。

    回归正题：我们需要让异步函数工作完成，主动通知自己工作已经完成。这里又是一个面向对象编程的思想，我们定义一个任务类：

    类的实例化被看做任务初始化

    类中提供一个开始执行的方法，控制任务的执行时机，修改自己的生命周期为正在执行

    类中还提供一个结束的方法，用于任务执行过程中随时可主动修改自己的生命周期为已结束

    代码如下：

    1. /**
    2.  * 任务单元
    3.  * @param taskId 任务id
    4.  * @param task   任务函数
    5.  * @param group   任务组对象
    6.  * @constructor
    7.  */
    8. function TaskUnit(taskId, task, group) {
    9.     this.taskId = taskId;  //任务唯一标识
    10.     this.status = TaskStatusEnum.INIT; //执行状态：-1初始化  0执行中  1已完成
    11.     this.task = task; //任务内容
    12.     /**
    13.      * 执行任务内容
    14.      */
    15.     this.execute = () => {
    16.         this.status = TaskStatusEnum.RUNNING;
    17.         this.task(this.end);
    18.     }
    19.     /**
    20.      * 任务完成回调
    21.      */
    22.     this.end = () => {
    23.         if (this.status === TaskStatusEnum.RUNNING) {
    24.             this.status = TaskStatusEnum.FINISH
    25.             group.check();
    26.         }
    27.     }
    28. }

    额(⊙o⊙)…目测这个类，好像和Promise长得差不多。少了异常捕获catch和finally两个函数，这两个不是我任务单元所关心的，因此这些东西交给实际任务内容去处理。

    多的是什么呢？任务并不是实例化之后被立刻调用，而是通过execute函数，让任务可在任意时机被调用，不受代码位置的限制。

    这样的好处是什么？我可以在一瞬间添加无数个任务，但执行的时候我可以控制想让哪几个任务一起执行，就可以一起执行，不想让谁执行，谁就执行不了，想什么时机执行就什么时机执行。实现异步串行。

    那可能有人会问，就凭这两段代码好像做不到异步串行吧。确实，这怎么解决？

    有一个方法就是可以给任务加标识，即标识为1的都可以同时执行，标识为2的必须等所有的1结束了才能执行。标识为2的任务怎么知道标识1的任务什么时候结束呢？当然要让标识1的任务主动通知，设置检查函数，每一个标识1的任务结束后都主动调用检查函数，发现全部执行完毕后，就开始调用标识2的任务，这样就实现了异步串行了。

    补：因为js是单线程，不会存在线程并发资源抢占的问题，一定有一个任务会检测到所有任务都结束了。在多线程编程的语言中，需要在检查的函数中加把锁。

    加标识的方法存在局限性，哪里有那么多标识可加，代码得多混乱.....

    因此我们又引入了任务组的概念：即可以同时一起执行的任务把它们归成一组，这样就会分成很多组，这个时候把任务组看成一个不可分割的整体，保证任务组的有序执行即可。如何保证有序，构建任务组队列，或者任务组链表，依次执行即可。

    再分析任务组内的所有任务如何保证全部执行完毕：让每个任务结束都通知任务组检查，总会有一个任务检查的时候发现任务都结束了

    代码如下：

    1. /**
    2.  * 任务组
    3.  * @constructor
    4.  */
    5. function TaskGroup() {
    6.     this.nextGroup=null;//下一个任务组
    7.     this.taskList=[];//任务列表
    8.     this.add=(task)=>{
    9.         this.taskList.push(task);
    10.     }
    11.     this.setNextGroup=(group)=>{
    12.         this.nextGroup=group;
    13.     }
    14.     /**
    15.      * 启动任务组
    16.      */
    17.     this.start=()=>{
    18.         for (let i = 0; i < this.taskList.length; i++) {
    19.             const task=this.taskList[i];
    20.             if (task.status === TaskStatusEnum.INIT) {
    21.                 task.execute();//执行
    22.             }
    23.         }
    24.     }
    25.     /**
    26.      * 检查任务
    27.      */
    28.     this.check=()=>{
    29.         for (let i = 0; i < this.taskList.length; i++) {
    30.             if (this.taskList[i].status !== TaskStatusEnum.FINISH) {
    31.                 return; //发现还有任务没有执行完成
    32.             }
    33.         }
    34.         if(this.nextGroup)this.nextGroup.start();//任务全部执行完成,进行下一个任务组
    35.     }
    36.  
    37. }

    最后创建一个任务中心，用来提供对外接口，将任务封装成一个个任务，组成任务组，控制启动结束等等。

    三、总代码如下：asyncController.js

    1. /**
    2.  * 任务执行状态枚举类
    3.  * @type {{INIT: number, RUNNING: number, FINISH: number}}
    4.  */
    5. const TaskStatusEnum = {
    6.     INIT: -1, //初始化
    7.     RUNNING: 0, //执行中
    8.     FINISH: 1 //已完成
    9. }
    10.  
    11. /**
    12.  * 任务单元
    13.  * @param taskId 任务id
    14.  * @param task   任务函数
    15.  * @param group   任务组对象
    16.  * @constructor
    17.  */
    18. function TaskUnit(taskId, task, group) {
    19.     this.taskId = taskId;  //任务唯一标识，暂时没用到
    20.     this.status = TaskStatusEnum.INIT; //执行状态：-1初始化  0执行中  1已完成
    21.     this.task = task; //任务内容
    22.     /**
    23.      * 执行任务内容
    24.      */
    25.     this.execute = () => {
    26.         this.status = TaskStatusEnum.RUNNING;
    27.         this.task(this.end);
    28.     }
    29.     /**
    30.      * 任务完成回调
    31.      */
    32.     this.end = () => {
    33.         if (this.status === TaskStatusEnum.RUNNING) {
    34.             this.status = TaskStatusEnum.FINISH
    35.             group.check();
    36.         }
    37.     }
    38. }
    39.  
    40. /**
    41.  * 任务组
    42.  * @constructor
    43.  */
    44. function TaskGroup() {
    45.     this.nextGroup=null;//下一个任务组
    46.     this.taskList=[];//任务列表
    47.     this.add=(task)=>{
    48.         this.taskList.push(task);
    49.     }
    50.     this.setNextGroup=(group)=>{
    51.         this.nextGroup=group;
    52.     }
    53.     /**
    54.      * 启动任务组
    55.      */
    56.     this.start=()=>{
    57.         for (let i = 0; i < this.taskList.length; i++) {
    58.             const task=this.taskList[i];
    59.             if (task.status === TaskStatusEnum.INIT) {
    60.                 task.execute();//执行
    61.             }
    62.         }
    63.     }
    64.     /**
    65.      * 检查任务
    66.      */
    67.     this.check=()=>{
    68.         for (let i = 0; i < this.taskList.length; i++) {
    69.             if (this.taskList[i].status !== TaskStatusEnum.FINISH) {
    70.                 return; //发现还有任务没有执行完成
    71.             }
    72.         }
    73.         //任务全部执行完成,进行下一个任务组
    74.         this.taskList=[] //清空当前任务组，让这些任务尽快回收
    75.         if(this.nextGroup){
    76.             let nextGroupTemp=this.nextGroup;
    77.             nextGroupTemp.start();
    78.             this.nextGroup=null;//取消引用，让当前任务组尽快回收
    79.         }
    80.     }
    81.  
    82. }
    83.  
    84. /**
    85.  * 异步函数控制器，
    86.  * .and() 添加异步并行函数
    87.  * .next() 添加同步串行函数
    88.  */
    89. module.exports = function () {
    90.     this.queue = [];
    91.     this.nowTaskGroup=null;//当前任务组
    92.     this.startCount = 0;//加入任务数
    93.  
    94.     /**
    95.      * 调用该函数表示添加并行任务
    96.      * @param task 任务
    97.      */
    98.     this.and = (task) => {
    99.         if(this.nowTaskGroup==null){
    100.             this.nowTaskGroup=new TaskGroup();
    101.         }
    102.         this.nowTaskGroup.add(new TaskUnit(++this.startCount, task, this.nowTaskGroup))
    103.         return this
    104.     }
    105.     /**
    106.      * 调用该函数表示添加串行任务
    107.      * @param task 任务
    108.      */
    109.     this.next = (task) => {
    110.         if(this.nowTaskGroup!=null) this.queue.push(this.nowTaskGroup);//防止上一个添加的任务是并行的
    111.         this.nowTaskGroup=new TaskGroup();
    112.         this.nowTaskGroup.add(new TaskUnit(++this.startCount, task, this.nowTaskGroup));
    113.         this.queue.push(this.nowTaskGroup);
    114.         this.nowTaskGroup=null;//当前任务添加结束清空
    115.         return this
    116.     }
    117.     /**
    118.      * 调用该函数表示任务添加完毕，开始执行任务
    119.      * @param endTask 任务全部结束后回调
    120.      */
    121.     this.finish = (endTask) => {
    122.         if(this.nowTaskGroup!=null) this.queue.push(this.nowTaskGroup);
    123.         this.nowTaskGroup=new TaskGroup();
    124.         this.nowTaskGroup.add(new TaskUnit(++this.startCount, endTask, this.nowTaskGroup));
    125.         this.queue.push(this.nowTaskGroup);
    126.         this.nowTaskGroup=null;//当前任务添加结束清空
    127.  
    128.         //组装成单向链表
    129.         for(let i=0;i<this.queue.length-1;i++){
    130.             this.queue[i].setNextGroup(this.queue[i+1]);
    131.         }
    132.         this.queue[0].start();//启动链表首个任务组
    133.         this.queue=[] //清空任务，为下一波任务做准备
    134.     }
    135. }

    四、使用样例：

    1. const Controller=require('./src/util/asyncController')
    2.  
    3. const controller=new Controller();
    4. controller.and(end=>{
    5.     setTimeout(()=>{
    6.         console.log("并行1")
    7.         end();
    8.     },2000);
    9. }).and(end=>{
    10.     setTimeout(()=>{
    11.         console.log("并行2")
    12.         end();
    13.     },2000);
    14. }).and(end=>{
    15.     setTimeout(()=>{
    16.         console.log("并行3")
    17.         end();
    18.     },2000);
    19. }).next(end=>{
    20.     setTimeout(()=>{
    21.         console.log("串行1")
    22.         end();
    23.     },2000);
    24. }).next(end=>{
    25.     setTimeout(()=>{
    26.         console.log("串行2")
    27.         end();
    28.     },2000);
    29. }).and(end=>{
    30.     setTimeout(()=>{
    31.         console.log("并行4")
    32.         end();
    33.     },2000);
    34. }).and(end=>{
    35.     setTimeout(()=>{
    36.         console.log("并行5")
    37.         end();
    38.     },2000);
    39. }).next(end=>{
    40.     setTimeout(()=>{
    41.         console.log("串行3")
    42.         end();
    43.     },2000);
    44. }).finish(()=>{
    45.     setTimeout(()=>{
    46.         console.log("结束")
    47.     },2000);
    48. })

     执行结果

    "C:\Program Files\nodejs\node.exe" main.js
    并行1
    并行2
    并行3
    串行1
    串行2
    并行4
    并行5
    串行3
    结束

    Process finished with exit code 0

    --------------------------------------------------

    2024/01/06更新

    代码格式优化

    1. /**
    2.  * 任务执行状态枚举类
    3.  * @type {{INIT: number, RUNNING: number, FINISH: number}}
    4.  */
    5. const TaskStatusEnum = {
    6.     INIT: -1, //初始化
    7.     RUNNING: 0, //执行中
    8.     FINISH: 1 //已完成
    9. }
    10.  
    11. /**
    12.  * 任务单元
    13.  */
    14. class TaskUnit {
    15.     /**
    16.      * 任务唯一标识，暂时没用到
    17.      * @type {String}
    18.      */
    19.     taskId;  //
    20.     /**
    21.      * 任务状态
    22.      * @type {Number}
    23.      */
    24.     status;
    25.     /**
    26.      * 任务执行函数
    27.      * @type {Function}
    28.      */
    29.     task; //任务内容
    30.     /**
    31.      * 任务所在组
    32.      * @type {TaskGroup}
    33.      */
    34.     group;
    35.  
    36.     /**
    37.      * 任务单元
    38.      * @param taskId 任务id
    39.      * @param task   任务函数
    40.      * @param group   任务组对象
    41.      * @constructor
    42.      */
    43.     constructor(taskId, task, group) {
    44.         this.taskId = taskId;
    45.         this.status = TaskStatusEnum.INIT;
    46.         this.task = task;
    47.         this.group = group;
    48.     }
    49.  
    50.     /**
    51.      * 执行任务内容
    52.      */
    53.     execute = () => {
    54.         this.status = TaskStatusEnum.RUNNING;
    55.         return this.task(this.end);
    56.     }
    57.     /**
    58.      * 任务完成回调
    59.      */
    60.     end = () => {
    61.         if (this.status === TaskStatusEnum.RUNNING) {
    62.             this.status = TaskStatusEnum.FINISH
    63.         }
    64.         return this.group.check();
    65.     }
    66.  
    67. }
    68.  
    69.  
    70. /**
    71.  * 任务组
    72.  * 任务组之间是串行执行的
    73.  * 一个任务组内部的任务是并行执行的，直至所有任务都完成，任务组结束
    74.  */
    75. class TaskGroup {
    76.     /**
    77.      * 下一个任务组
    78.      * @type {TaskGroup}
    79.      */
    80.     nextGroup;
    81.     /**
    82.      * 当前任务组所有任务
    83.      * @type {[TaskUnit]}
    84.      */
    85.     taskList = [];
    86.  
    87.     /**
    88.      * 任务组完成后回调
    89.      * @type {Function}
    90.      */
    91.     finish;
    92.  
    93.     /**
    94.      *
    95.      * @param {Function} finish
    96.      */
    97.     setFinishCallback(finish) {
    98.         this.finish = finish;
    99.     }
    100.  
    101.     add = (task) => {
    102.         this.taskList.push(task);
    103.     }
    104.  
    105.     setNextGroup = (group) => {
    106.         this.nextGroup = group;
    107.     }
    108.     /**
    109.      * 启动任务组
    110.      */
    111.     start = () => {
    112.         this.taskList.forEach(e=>{
    113.             if (e.status === TaskStatusEnum.INIT) {
    114.                 e.execute();//执行
    115.             }
    116.         })
    117.     }
    118.  
    119.     /**
    120.      * 检查任务
    121.      */
    122.     check() {
    123.         /**  检查任务是否均执行完成  **/
    124.         for (let i = 0; i < this.taskList.length; i++) {
    125.             if (this.taskList[i].status !== TaskStatusEnum.FINISH) {
    126.                 return; //发现还有任务没有执行完成
    127.             }
    128.         }
    129.         this.taskList = null;
    130.         /** 务全部执行完成,进行下一个任务组，没有下一个任务组，就执行完成的回调函数 **/
    131.         if (this.nextGroup) {
    132.             let nextGroupTemp = this.nextGroup;
    133.             this.nextGroup = null;
    134.             return nextGroupTemp.start();
    135.         } else {
    136.             if (this.finish){
    137.                 return this.finish(); //所有任务组都结束了
    138.             }
    139.         }
    140.     }
    141. }
    142.  
    143. /**
    144.  * 同步控制器
    145.  */
    146. class Asyncor {
    147.     /**
    148.      * 任务组队列
    149.      * @type {[TaskGroup]}
    150.      */
    151.     queue = [];
    152.     /**
    153.      * 当时任务组
    154.      * @type {TaskGroup}
    155.      */
    156.     nowTaskGroup;
    157.     /**
    158.      * 加入任务数
    159.      * @type {number}
    160.      */
    161.     startCount = 0;
    162.  
    163.     /**
    164.      * 向当前任务组添加并行任务，没有任务组则创建一个
    165.      * @param task 任务
    166.      */
    167.     add = (task) => {
    168.         if (this.nowTaskGroup == null) {
    169.             this.nowTaskGroup = new TaskGroup();
    170.         }
    171.         if (task) {
    172.             /**  添加任务 */
    173.             this.nowTaskGroup.add(new TaskUnit(++this.startCount, task, this.nowTaskGroup))
    174.         }
    175.         return this
    176.     }
    177.     /**
    178.      * 本着上一步必须完成之后才能进行下一步的原则，当前函数为新的下一步。
    179.      * 应将之前添加的任务与当前任务分离开，创建一个新的任务组，添加当前步骤的任务
    180.      * 可通过add函数在当前步骤中继续追加并行任务
    181.      * 或者继续调用该next进行添加下下一步
    182.      * @param task 任务
    183.      */
    184.     next = (task) => {
    185.         /** 看看有没有上一步任务，有就分开 */
    186.         if (this.nowTaskGroup != null) {
    187.             this.queue.push(this.nowTaskGroup);
    188.         }
    189.         /**  每次调用该函数都会创建一个单独的任务组，与上一步任务隔离开  */
    190.         this.nowTaskGroup = new TaskGroup();
    191.         if (task) {
    192.             /**  添加任务 */
    193.             this.nowTaskGroup.add(new TaskUnit(++this.startCount, task, this.nowTaskGroup));
    194.         }
    195.         return this
    196.     }
    197.     /**
    198.      * 调用该函数表示任务添加完毕，开始执行任务
    199.      * @param {Function}finish 任务全部结束后回调
    200.      */
    201.     start = (finish) => {
    202.         /**  将启动前最后的任务组打包  **/
    203.         if (this.nowTaskGroup != null) {
    204.             this.queue.push(this.nowTaskGroup);
    205.         }
    206.         this.nowTaskGroup = null;
    207.  
    208.         /**  所有任务组组装成单向链表  */
    209.         for (let i = 0; i < this.queue.length - 1; i++) {
    210.             this.queue[i].setNextGroup(this.queue[i + 1]);
    211.         }
    212.  
    213.         /**  给最后一个任务组 设置全部完成的回调函数  **/
    214.         this.queue[this.queue.length - 1].setFinishCallback(finish);
    215.  
    216.         /**  启动链表首个任务组 */
    217.         const firstGroup = this.queue[0];
    218.         /**  这里清空任务队列，可以使当前Asyncor对象可以复用   */
    219.         this.queue = []
    220.         return firstGroup.start();//启动链表首个任务组
    221.     }
    222.  
    223. }
    224.  
    225. module.exports = Asyncor
    226.

    提供三个函数

    asyncor.add()
asyncor.next()
asyncor.start()
    1. asyncor.add(function (end) {
    2.     setTimeout(()=>{
    3.         //自己的异步代码
    4.         end();//通知控制器当前工作已完成
    5.     })
    6. })
    7.  
    8. asyncor.next(function (end) {
    9.     setTimeout(()=>{
    10.         //自己的异步代码
    11.         end();//通知控制器当前工作已完成
    12.     })
    13. })
    14.  
    15. asyncor.start(function () {
    16.     //整个任务都完成之后会调用这里，可以在这里自定义功能
    17. });

    支持链式调用

    asyncor.add().add().next().next().add().next().add().start()

    如何使用？如图表示任务计划执行情况，每一个单元格表示一个任务，横向的表示并发执行，纵向表示任务批次，批次需要串行执行。

    第1批次123
    第2批次45
    第3批次6
    第4批次7
    第5批次8910

    对应代码示例

    1. //先创建对象
    2. const asyncor=new Asyncor();
    3.  
    4.  
    5.  
    6. /** 第一批次  */
    7. asyncor.add(end=>{
    8.     //任务1
    9.     console.log("任务1")
    10.  
    11.     end();
    12. }).add(end=>{
    13.     //任务2
    14.     console.log("任务2")
    15.  
    16.     end();
    17. }).add(end=>{
    18.     //任务3
    19.     console.log("任务3")
    20.  
    21.     end();
    22. });
    23.  
    24.  
    25. /** 第二批次 */
    26. asyncor.next(end=>{
    27.     //任务4
    28.     console.log("任务4")
    29.  
    30.     end();
    31. }).add(end=>{
    32.     //任务5
    33.     console.log("任务5")
    34.  
    35.     end();
    36. });
    37.  
    38.  
    39.  
    40. /** 第三批次  */
    41. asyncor.next(end=>{
    42.     //任务6
    43.     console.log("任务6")
    44.  
    45.     end();
    46. });
    47.  
    48.  
    49.  
    50. /** 第四批次 */
    51. asyncor.next(end=>{
    52.     //任务7
    53.     console.log("任务7")
    54.  
    55.     end();
    56. });
    57.  
    58.  
    59.  
    60. /** 第五批次  */
    61. asyncor.next(end=>{
    62.     //任务8
    63.     console.log("任务8")
    64.  
    65.     end();
    66. }).add(end=>{
    67.     //任务9
    68.     console.log("任务9")
    69.  
    70.     end();
    71. }).add(end=>{
    72.     //任务10
    73.     console.log("任务10")
    74.  
    75.     end();
    76. });
    77.  
    78.  
    79.  
    80. /** 启动  */
    81. asyncor.start(()=>{
    82.     console.log("执行完成");
    83. });
    84.  
    85.  
    86.  
    87. 输出：
    88. 任务1
    89. 任务2
    90. 任务3
    91. 任务4
    92. 任务5
    93. 任务6
    94. 任务7
    95. 任务8
    96. 任务9
    97. 任务10
    98. 执行完成
    99.

  • 相关阅读:
    opencv之坑（八）——putText中文乱码解决
    浅谈web前端工程师hr面试经典问题20+
    Java笔记（一）：volatile、synchronized关键字
    C语言内存函数
    分拆计划陷入困境，英特尔还能重回巅峰吗?
    LeetCode 150. 逆波兰表达式求值
    Spring Cloud
    【JUC】循环屏障 CyclicBarrier 详解
    js中call、apply和bind：
    vcpkg安装第三方库，报错fatal error RC1107: invalid usage； use RC /? for Help
  • 原文地址：https://blog.csdn.net/qq_43319748/article/details/132746961