如何在Effect-Ts的Effect.all执行过程中跟踪任务进度？

问题描述 #

我正在使用Effect-Ts，通过Effect.all以指定并发级别执行多个异步操作（Promise），基础实现代码如下：

async function method<R>(
  fn: (() => Promise<R>)[],
  concurrency: number,
) {
  const tasks = fn.map(c =>
    Effect.promise(c)
  )

  let effect = Effect.all(tasks, {
    concurrency: concurrency,
  })
  
  const results = await Effect.runPromise(effect)
  return results
}

希望跟踪任务执行进程，实现类似如下的进度条效果：

Progress: 1/100
Progress: 2/100
...
Progress: 100/100

尝试过多种方法但未能实现需求，请问最适合的实现方式是什么？

实现方案 #

可以利用Effect-Ts的Ref来维护已完成任务的计数，给每个任务添加完成后的回调逻辑来更新进度，具体实现如下：

import { Effect, Ref } from "effect";

async function method<R>(
  fn: (() => Promise<R>)[],
  concurrency: number,
) {
  // 创建可变引用跟踪已完成任务数
  const completedCountRef = await Effect.runPromise(Ref.make(0));
  const totalTasks = fn.length;

  // 给每个任务包装进度更新逻辑
  const tasksWithProgress = fn.map(taskFn =>
    Effect.promise(taskFn).pipe(
      Effect.tap(() =>
        Ref.update(completedCountRef, count => {
          const newCount = count + 1;
          console.log(`Progress: ${newCount}/${totalTasks}`);
          return newCount;
        })
      )
    )
  );

  const effect = Effect.all(tasksWithProgress, {
    concurrency: concurrency,
  });
  
  const results = await Effect.runPromise(effect);
  return results;
}

逻辑说明 #

• 状态管理：用Ref.make(0)创建安全的共享可变状态，避免多任务并发更新时的竞态问题，这是Effect-Ts推荐的可变状态处理方式。
• 进度触发：通过Effect.tap在每个任务成功完成后执行计数更新和进度打印，不干扰任务本身的返回值。
• 并发保持：保留原有的Effect.all并发控制逻辑，确保任务按指定并发级别执行。

如果需要动态覆盖当前行的进度条效果（更美观的终端展示），可以替换console.log为process.stdout.write：

process.stdout.write(`\rProgress: ${newCount}/${totalTasks}`);
// 全部完成后换行
if (newCount === totalTasks) {
  console.log();
}

内容的提问来源于stack exchange，提问作者gixlg