使用非阻塞 IO 对文件收集和聚合结果报告进行操作

Question

我想对任意大的文件集执行一些任意昂贵的工作。我想实时报告进度，然后在处理完所有文件后显示结果。如果没有与我的表达式匹配的文件，我想抛出一个错误。

想象一下编写一个测试框架，加载所有测试文件，执行它们（没有特定的顺序），实时报告进度，然后在所有测试完成后显示汇总结果。

使用阻塞语言（例如 Ruby）编写此代码非常简单。

事实证明，我在节点中执行这个看似简单的任务时遇到了麻烦，同时也真正利用了异步、基于事件的 IO。

我的第一个设计是串行执行每个步骤。

1. 加载所有文件，创建要处理的文件集合 处理
2. 集合中的每个文件
3. 处理所有文件后报告结果

这种方法确实有效，但对我来说似乎不太正确，因为它会导致计算成本更高我的程序的一部分等待所有文件 IO 完成。 Node 的设计目的不就是要避免这种等待吗？

我的第二个设计是处理在磁盘上异步找到的每个文件。为了便于讨论，让我们想象一个看起来像这样的方法：

eachFileMatching(path, expression, callback) {
  // recursively, asynchronously traverse the file system,
  // calling callback every time a file name matches expression.
}

该方法的使用者看起来像这样：

eachFileMatching('test/', /_test.js/, function(err, testFile) {
  // read and process the content of testFile
});

虽然这种设计感觉像是一种非常“节点”的 IO 工作方式，但它存在两个主要问题（至少在我可能错误的实现中）：

1. 我不知道所有文件何时已处理完毕，所以我不知道何时组装和发布结果。
2. 由于文件读取是非阻塞且递归的，因此我正在努力了解如何知道是否未找到文件。

我希望我只是做错了什么，并且其他人可以使用一些相当简单的策略来使第二种方法发挥作用。

尽管这个示例使用了测试框架，但我还有许多其他项目遇到了同样的问题，并且我想任何编写一个相当复杂的应用程序来访问节点中的文件系统的人也会遇到这种情况。

Answer 1

“读取并处理 testFile 的内容”是什么意思？

我不明白为什么您不知道所有文件何时处理完毕。你不使用Streams吗？一个流有多个事件，而不仅仅是数据。如果您处理 end 事件，那么您将知道每个文件何时完成。

例如，您可能有一个文件名列表，为每个文件设置处理，然后当您收到结束事件时，从列表中删除文件名。当列表为空时，您就完成了。或者创建一个包含名称和完成状态的 FileName 对象。当您收到 end 事件时，更改状态并减少文件名计数器。当计数器归零时，您就完成了，或者如果您不确定，您可以扫描所有 FileName 对象以确保它们的状态已完成。

您还可能有一个计时器来定期检查计数器，如果它在一段时间内没有变化，则报告处理可能卡在状态未完成的 FileName 对象上。

...我刚刚在另一个问题中遇到了这种情况，接受的答案（加上 github 链接）很好地解释了它。查看for循环事件驱动代码？

Answer 2

事实证明，我能够构建的最小工作解决方案比我希望的要复杂得多。

以下是适合我的代码。它可能可以被清理或变得稍微更具可读性，我对这样的反馈不感兴趣。

如果有一种明显不同的方法来解决这个问题，那就是更简单和/或更有效，我很有兴趣听到它。确实令我惊讶的是，解决这个看似简单的需求竟然需要如此大量的代码，但这也许就是有人发明阻塞io的原因？

复杂性实际上在于满足以下所有要求：

• 在找到文件时对其进行处理
• 了解搜索何时完成
• 了解是否未找到文件

以下是代码：

/**
 * Call fileHandler with the file name and file Stat for each file found inside
 * of the provided directory.
 *
 * Call the optionally provided completeHandler with an array of files (mingled
 * with directories) and an array of Stat objects (one for each of the found
 * files.
 *
 * Following is an example of a simple usage:
 *
 *   eachFileOrDirectory('test/', function(err, file, stat) {
 *     if (err) throw err;
 *     if (!stat.isDirectory()) {
 *       console.log(">> Found file: " + file);
 *     }
 *   });
 *
 * Following is an example that waits for all files and directories to be 
 * scanned and then uses the entire result to do something:
 *
 *   eachFileOrDirectory('test/', null, function(files, stats) {
 *     if (err) throw err;
 *     var len = files.length;
 *     for (var i = 0; i < len; i++) {
 *       if (!stats[i].isDirectory()) {
 *         console.log(">> Found file: " + files[i]);
 *       }
 *     }
 *   });
 */
var eachFileOrDirectory = function(directory, fileHandler, completeHandler) {
  var filesToCheck = 0;
  var checkedFiles = [];
  var checkedStats = [];

  directory = (directory) ? directory : './';

  var fullFilePath = function(dir, file) {
    return dir.replace(/\/$/, '') + '/' + file;
  };

  var checkComplete = function() {
    if (filesToCheck == 0 && completeHandler) {
      completeHandler(null, checkedFiles, checkedStats);
    }
  };

  var onFileOrDirectory = function(fileOrDirectory) {
    filesToCheck++;
    fs.stat(fileOrDirectory, function(err, stat) {
      filesToCheck--;
      if (err) return fileHandler(err);
      checkedFiles.push(fileOrDirectory);
      checkedStats.push(stat);
      fileHandler(null, fileOrDirectory, stat);
      if (stat.isDirectory()) {
        onDirectory(fileOrDirectory);
      }
      checkComplete();
    });
  };

  var onDirectory = function(dir) {
    filesToCheck++;
    fs.readdir(dir, function(err, files) {
      filesToCheck--;
      if (err) return fileHandler(err);
      files.forEach(function(file, index) {
        file = fullFilePath(dir, file);
        onFileOrDirectory(file);
      });
      checkComplete();
    });
  }

  onFileOrDirectory(directory);
};

Answer 3

执行此操作的两种方法，第一种方法可能是串行考虑的，类似于

var files = [];
doFile(files, oncomplete);

function doFile(files, oncomplete) {
  if (files.length === 0) return oncomplete();
  var f = files.pop();
  processFile(f, function(err) {
    // Handle error if any
    doFile(files, oncomplete); // Recurse
  });
};

function processFile(file, callback) {
  // Do whatever you want to do and once 
  // done call the callback
  ...
  callback();
};

第二种方法，我们称之为并行，它是类似的，并且总结如下：

var files = [];
doFiles(files, oncomplete);

function doFiles(files, oncomplete) {
  var exp = files.length;
  var done = 0;
  for (var i = 0; i < exp; i++) {
    processFile(files[i], function(err) {
      // Handle errors (but still need to increment counter)
      if (++done === exp) return oncomplete();      
    });
  }
};

function processFile(file, callback) {
  // Do whatever you want to do and once 
  // done call the callback
  ...
  callback();
};

现在看起来很明显您应该使用第二种方法，但您会发现对于 IO 密集型操作并行化时你并没有真正获得任何性能提升。第一种方法的一个缺点是递归可能会破坏堆栈跟踪。

吉

多