@hongfangze/concurrent 并发

介绍

并发处理任务。

运行方式：

  // 定义并发任务要执行的函数，这里用休眠来模拟http请求需要不同的相应时间
  // 第一个参数是用户传入的业务参数
  // 第二个参数是index，即任务所在索引
  const fn = async (params: any, index: number) => {
    // 记录函数开始时间
    let s = new Date().valueOf();
    // 这里模拟了函数异常
    let e = getRandomNum(1, 3) == 3 ? true : false;
    console.log(`我是第${index + 1}个人，${params.name}，我今年${params.age}岁，我现在睡觉了`);
    // 休眠一定时长
    await sleep(getRandomNum(1000, 10000));
    if (!e) {
      // 正常的返回
      return `${params.name}睡醒了，睡了：${((new Date().valueOf() - s) / 1000)}秒`;
    } else {
      // 函数异常
      throw new Error(`${params.name}睡过去了`);
    }
  }
  // 创建一个任务队列
  const taskInstance = new Task();
  // 定义传入fn的参数，每个元素为一次任务
  const params = [
    { name: "刘一", age: 11 },
    { name: "陈二", age: 12 },
    { name: "张三", age: 13 },
    { name: "李四", age: 14 },
    { name: "王五", age: 15 },
    { name: "赵六", age: 16 },
    { name: "孙七", age: 17 },
    { name: "周八", age: 18 },
    { name: "吴九", age: 19 },
    { name: "郑十", age: 20 },
  ]
  // 监听任务处理情况
  taskInstance.on("process", (index, err, result: any) => {
    console.log(`第${index}个人${params[index].name}已经完成睡觉（某一个任务完成），返回值：`, result);
  });
  taskInstance.on("pause", () => {
    console.log("任务已暂停");
  });
  taskInstance.on("continue", () => {
    console.log("任务已继续");
  });
  taskInstance.on("stop", (results) => {
    console.log("任务已被强行终止，目前已经完成的任务结果：", results);
  });
  taskInstance.on("end", (results) => {
    console.log("所有人都睡完了（任务完成），返回值：", results);
  });
  // 开启3个并发运行
  taskInstance.start(fn, params, 3);
  // await taskInstance.start(fn, params, 3);
  // 等待2秒后暂停任务
  // 注意：暂停的只是还未开始的单个任务，已经在队列中运行的任务仍旧会继续
  await sleep(2000);
  await taskInstance.pause();
  // 等待15秒后继续任务
  // 如果之前的队列已经完成，那么继续从待运行任务中取出对应并发数的任务进行运行，否则仍旧将等待之前队列中的任务完成
  await sleep(15000);
  await taskInstance.continue();
  // 如果3秒后还没结束任务，就强行终止
  // 注意：强行停止任务，只能是还未进入运行队列的任务不在执行，目前已经在运行队列中的任务仍将继续运行结束
  await sleep(3000);
  await taskInstance.stop();

版本迭代记录

2024-03-21 v0.2.1

• 修复因并发数大于任务总数的情况下导致的任务结果未正确返回的Bug。

2024-03-21 v0.2.0

• 增加任务的暂停、继续、强制停止功能。

2024-03-20 v0.1.0

• 增加单个任务结果实时订阅。
• 不兼容以前的版本。