Gulp流水线中异步函数会阻塞后续文件处理？求原理解析

我了解到Gulp的核心特性是模拟并行处理，因此原本预期：即便Gulp流水线中存在异步函数，后续文件的处理也会在前一个文件的异步函数完成前启动。

假设现有两个文件：Foo.txt和Bar.txt，其中Foo.txt先进入Gulp流水线。我预期Bar.txt的处理会在myAsyncFunction完成Foo.txt的处理前启动。

测试代码 #

const gulp = require('gulp');
const through = require('through2');
const PluginError = require('plugin-error');

// 定义用于Transform流的异步函数
async function myAsyncFunction(data) {
  console.log("Checkpoint 4");
  // 模拟异步操作
  await new Promise((resolve) => setTimeout(resolve, 5000));

  // 处理数据
  const transformedData = data.toString().toUpperCase();

  console.log("Checkpoint 5");
  return transformedData;
}

// 使用through2创建Transform流
const myTransformStream = through.obj(async function (file, encoding, callback) {
  console.log("Checkpoint 1");

  if (file.isNull()) {
    console.log("Checkpoint 2-A");
    // 空文件直接传递
    return callback(null, file);
  }

  console.log("Checkpoint 2-B");

  if (file.isBuffer()) {
    console.log("Checkpoint 3");

    try {
      // 将文件内容转为字符串并传入异步函数
      const transformedData = await myAsyncFunction(file.contents.toString());

      console.log("Checkpoint 6");
      console.log("=============");

      // 创建处理后的新文件
      const transformedFile = file.clone();
      transformedFile.contents = Buffer.from(transformedData);

      // 将处理后的文件传入下一个流
      this.push(transformedFile);
    } catch (err) {
      // 处理异步操作中的错误
      this.emit('error', new PluginError('gulp-my-task', err));
    }
  }

  callback();
});

// 定义使用Transform流的Gulp任务
gulp.task('my-task', function () {
  return gulp.src('src/*.txt')
      .pipe(myTransformStream)
      .pipe(gulp.dest('dist'));
});

实际输出 #

Checkpoint 1  
Checkpoint 2-B
Checkpoint 3  
Checkpoint 4  
Checkpoint 5
Checkpoint 6  
============= 
Checkpoint 1  
Checkpoint 2-B
Checkpoint 3  
Checkpoint 4  
Checkpoint 5
Checkpoint 6
=============

预期输出 #

Checkpoint 1  
Checkpoint 2-B
Checkpoint 3  
Checkpoint 4
Checkpoint 1  // <- 开始处理下一个文件  
Checkpoint 2-B
Checkpoint 3  
Checkpoint 4    
Checkpoint 5
Checkpoint 6  
============= 
Checkpoint 5
Checkpoint 6
=============

原理解释 #

核心原因：Transform流的串行机制与代码逻辑问题 #

1. through2 Transform流的默认行为：through2创建的Transform流默认是串行处理文件的——它必须等到当前文件对应的callback()被调用后，才会从上游读取下一个文件进入流中处理。
2. async/await的阻塞影响：你的代码中，await myAsyncFunction会阻塞当前async函数的执行，必须等5秒的异步操作完成后，才会继续执行后续代码并调用callback()。这就导致第一个文件的整个异步流程完全走完，才会触发第二个文件的处理，最终表现为串行执行。

实现并行处理的修正思路 #

要实现预期的并行处理，需要提前调用callback()，让流立即开始处理下一个文件，同时在异步操作完成后再将处理后的文件推入流中。修改后的核心代码如下：

const myTransformStream = through.obj(function (file, encoding, callback) {
  console.log("Checkpoint 1");

  if (file.isNull()) {
    console.log("Checkpoint 2-A");
    return callback(null, file);
  }

  console.log("Checkpoint 2-B");

  if (file.isBuffer()) {
    console.log("Checkpoint 3");

    // 不等待异步完成，直接调用callback，让下一个文件进入处理流程
    callback();

    // 异步处理逻辑独立执行，完成后再推送结果
    myAsyncFunction(file.contents.toString())
      .then(transformedData => {
        console.log("Checkpoint 6");
        console.log("=============");
        const transformedFile = file.clone();
        transformedFile.contents = Buffer.from(transformedData);
        this.push(transformedFile);
      })
      .catch(err => {
        this.emit('error', new PluginError('gulp-my-task', err));
      });
  } else {
    callback();
  }
});

这种修改方式会让多个文件的异步操作同时进行，输出结果将符合预期。需要注意的是，这种并行处理可能会打乱文件的输出顺序，如果需要保持原文件顺序，还需要额外的顺序控制逻辑。

内容的提问来源于stack exchange，提问作者Takeshi Tokugawa YD