在forEach循环中使用async/await有什么问题吗?我正在尝试循环浏览一系列文件,并等待每个文件的内容。

import fs from 'fs-promise'

async function printFiles () {
  const files = await getFilePaths() // Assume this works fine

  files.forEach(async (file) => {
    const contents = await fs.readFile(file, 'utf8')
    console.log(contents)
  })
}

printFiles()

这段代码确实有效,但这段代码会出错吗?我有人告诉我,你不应该在这样的高阶函数中使用async/await,所以我只想问问这是否有问题。


当前回答

使用Task、futurize和可遍历列表,您可以简单地

async function printFiles() {
  const files = await getFiles();

  List(files).traverse( Task.of, f => readFile( f, 'utf-8'))
    .fork( console.error, console.log)
}

这是你如何设置的

import fs from 'fs';
import { futurize } from 'futurize';
import Task from 'data.task';
import { List } from 'immutable-ext';

const future = futurizeP(Task)
const readFile = future(fs.readFile)

构建所需代码的另一种方法是

const printFiles = files => 
  List(files).traverse( Task.of, fn => readFile( fn, 'utf-8'))
    .fork( console.error, console.log)

或者甚至更注重功能

// 90% of encodings are utf-8, making that use case super easy is prudent

// handy-library.js
export const readFile = f =>
  future(fs.readFile)( f, 'utf-8' )

export const arrayToTaskList = list => taskFn => 
  List(files).traverse( Task.of, taskFn ) 

export const readFiles = files =>
  arrayToTaskList( files, readFile )

export const printFiles = files => 
  readFiles(files).fork( console.error, console.log)

然后从父函数

async function main() {
  /* awesome code with side-effects before */
  printFiles( await getFiles() );
  /* awesome code with side-effects after */
}

如果你真的想在编码上有更多的灵活性,你可以这样做(为了好玩,我使用了建议的Pipe Forward操作符)

import { curry, flip } from 'ramda'

export const readFile = fs.readFile 
  |> future,
  |> curry,
  |> flip

export const readFileUtf8 = readFile('utf-8')

PS-我没有在控制台上尝试这段代码,可能有一些拼写错误。。。正如90年代的孩子们所说的那样,“直式自由泳,从穹顶上跳下来!”-p

其他回答

一个重要的警告是:等待+等待。。of方法和forEach+异步方式实际上有不同的效果。

在真正的for循环中等待将确保所有异步调用都逐一执行。forEach+异步方式将同时发出所有的承诺,这会更快,但有时会不堪重负(如果您执行一些DB查询或访问一些具有容量限制的web服务,并且不想一次发出100000个呼叫)。

如果您不使用async/await,并且希望确保一个接一个地读取文件,那么也可以使用reduce+promise(不太优雅)。

files.reduce((lastPromise, file) => 
 lastPromise.then(() => 
   fs.readFile(file, 'utf8')
 ), Promise.resolve()
)

或者您可以创建一个forEachAsync来帮助,但基本上使用相同的for循环底层。

Array.prototype.forEachAsync = async function(cb){
    for(let x of this){
        await cb(x);
    }
}

若要查看如何出错,请在方法末尾打印console.log。

一般情况下可能出错的事情:

任意顺序。printFiles可以在打印文件之前完成运行。性能差。

这些并不总是错误的,但通常在标准用例中。

通常,使用forEach将导致除最后一个之外的所有结果。它将在不等待函数的情况下调用每个函数,这意味着它将告诉所有函数开始,然后完成,而不等待函数完成。

import fs from 'fs-promise'

async function printFiles () {
  const files = (await getFilePaths()).map(file => fs.readFile(file, 'utf8'))

  for(const file of files)
    console.log(await file)
}

printFiles()

这是本机JS中的一个示例,它将保持顺序,防止函数过早返回,并在理论上保持最佳性能。

这将:

启动所有并行文件读取。通过使用映射将文件名映射到要等待的承诺来保持顺序。按照数组定义的顺序等待每个承诺。

使用此解决方案,第一个文件将在其可用时立即显示,而无需等待其他文件首先可用。

它还将同时加载所有文件,而不必等待第一个文件完成后才能开始第二次文件读取。

这和原始版本的唯一缺点是,如果一次启动多个读取,则由于一次可能发生更多错误,因此处理错误更困难。

对于一次读取一个文件的版本,则会在出现故障时停止,而不会浪费时间尝试读取更多文件。即使有一个精心设计的取消系统,也很难避免它在第一个文件上失败,但也很难读取大部分其他文件。

性能并不总是可预测的。虽然许多系统的并行文件读取速度会更快,但有些系统更倾向于顺序读取。有些是动态的,可能会在负载下发生变化,提供延迟的优化在激烈竞争下并不总能产生良好的吞吐量。

该示例中也没有错误处理。如果有什么东西要求他们要么全部成功展示,要么根本不展示,那它就做不到。

建议在每个阶段使用console.log进行深入实验,并使用假文件读取解决方案(随机延迟)。尽管许多解决方案在简单的情况下似乎都是一样的,但它们都有细微的差异,需要额外的仔细检查才能挤出。

使用此模拟来帮助区分解决方案之间的差异:

(async () => {
  const start = +new Date();
  const mock = () => {
    return {
      fs: {readFile: file => new Promise((resolve, reject) => {
        // Instead of this just make three files and try each timing arrangement.
        // IE, all same, [100, 200, 300], [300, 200, 100], [100, 300, 200], etc.
        const time = Math.round(100 + Math.random() * 4900);
        console.log(`Read of ${file} started at ${new Date() - start} and will take ${time}ms.`)
        setTimeout(() => {
          // Bonus material here if random reject instead.
          console.log(`Read of ${file} finished, resolving promise at ${new Date() - start}.`);
          resolve(file);
        }, time);
      })},
      console: {log: file => console.log(`Console Log of ${file} finished at ${new Date() - start}.`)},
      getFilePaths: () => ['A', 'B', 'C', 'D', 'E']
    };
  };

  const printFiles = (({fs, console, getFilePaths}) => {
    return async function() {
      const files = (await getFilePaths()).map(file => fs.readFile(file, 'utf8'));

      for(const file of files)
        console.log(await file);
    };
  })(mock());

  console.log(`Running at ${new Date() - start}`);
  await printFiles();
  console.log(`Finished running at ${new Date() - start}`);
})();

OP的原始问题

在forEach循环中使用async/await有什么问题吗。。。

在@Bergi选择的答案中,它展示了如何串行和并行处理。然而,并行性还存在其他问题-

订单--@chharvey注意到-

例如,如果一个非常小的文件在一个非常大的文件之前完成了读取,那么它将首先被记录,即使小文件在文件数组中位于大文件之后。

可能一次打开太多文件--Bergi在另一个答案下的评论

同时打开数千个文件以同时读取它们也是不好的。人们总是要评估顺序、并行或混合方法是否更好。

因此,让我们来解决这些问题,展示实际的代码,简洁明了,不使用第三方库。易于剪切、粘贴和修改的东西。

并行读取(一次读取),串行打印(每个文件尽可能早)。

最简单的改进是像@Bergi的回答那样执行完全并行,但做了一个小改动,以便在保持顺序的同时尽快打印每个文件。

async function printFiles2() {
  const readProms = (await getFilePaths()).map((file) =>
    fs.readFile(file, "utf8")
  );
  await Promise.all([
    await Promise.all(readProms),                      // branch 1
    (async () => {                                     // branch 2
      for (const p of readProms) console.log(await p);
    })(),
  ]);
}

上面,两个单独的分支同时运行。

分支1:同时并行读取,分支2:连续读取以强制排序,但等待时间不超过必要

这很容易。

在并发限制下并行读取,串行打印(每个文件尽可能早)。

“并发限制”意味着同时读取的文件不超过N个。就像一家一次只允许这么多顾客进入的商店(至少在新冠疫情期间)。

首先引入了一个helper函数-

function bootablePromise(kickMe: () => Promise<any>) {
  let resolve: (value: unknown) => void = () => {};
  const promise = new Promise((res) => { resolve = res; });
  const boot = () => { resolve(kickMe()); };
  return { promise, boot };
}

函数bootablePromise(kickMe:()=>Promise<any>)需要函数kickMe作为启动任务的参数(在本例中为readFile),但不会立即启动。

bootablePromise返回几个财产

承诺类型承诺引导类型函数()=>void

承诺有两个阶段

承诺开始一项任务作为一个承诺,完成一项已经开始的任务。

当调用boot()时,promise从第一状态转换到第二状态。

bootablePromise用于printFiles--

async function printFiles4() {
  const files = await getFilePaths();
  const boots: (() => void)[] = [];
  const set: Set<Promise<{ pidx: number }>> = new Set<Promise<any>>();
  const bootableProms = files.map((file,pidx) => {
    const { promise, boot } = bootablePromise(() => fs.readFile(file, "utf8"));
    boots.push(boot);
    set.add(promise.then(() => ({ pidx })));
    return promise;
  });
  const concurLimit = 2;
  await Promise.all([
    (async () => {                                       // branch 1
      let idx = 0;
      boots.slice(0, concurLimit).forEach((b) => { b(); idx++; });
      while (idx<boots.length) {
        const { pidx } = await Promise.race([...set]);
        set.delete([...set][pidx]);
        boots[idx++]();
      }
    })(),
    (async () => {                                       // branch 2
      for (const p of bootableProms) console.log(await p);
    })(),
  ]);
}

和以前一样,有两个分支

分支1:用于运行和处理并发。分支2:用于打印

现在的区别是不允许并发运行超过concurrentLimit Promise。

重要的变量是

boots:要调用以强制其相应Promise转换的函数数组。它仅在分支1中使用。set:在随机访问容器中有Promise,这样一旦实现,就可以很容易地删除它们。此容器仅在分支1中使用。bootableProms:这些是与最初在集合中的Promise相同的Promise,但它是一个数组而不是集合,并且该数组从未更改。它仅在分支2中使用。

使用模拟fs.readFile运行,所需时间如下(文件名与时间(毫秒))。

const timeTable = {
  "1": 600,
  "2": 500,
  "3": 400,
  "4": 300,
  "5": 200,
  "6": 100,
};

可以看到这样的测试运行时间,显示并发正在运行--

[1]0--0.601
[2]0--0.502
[3]0.503--0.904
[4]0.608--0.908
[5]0.905--1.105
[6]0.905--1.005

可在typescript游乐场沙盒中执行

今天,我遇到了多种解决方案。在forEach循环中运行异步等待函数。通过构建包装器,我们可以实现这一点。

在这里的链接中提供了关于它如何在内部工作、对于本机forEach以及为什么它不能进行异步函数调用的更多详细说明,以及关于各种方法的其他详细信息

可以通过多种方式实现,如下所示,

方法1:使用包装器。

await (()=>{
     return new Promise((resolve,reject)=>{
       items.forEach(async (item,index)=>{
           try{
               await someAPICall();
           } catch(e) {
              console.log(e)
           }
           count++;
           if(index === items.length-1){
             resolve('Done')
           }
         });
     });
    })();

方法2:使用与Array.prototype的泛型函数相同的方法

EachAsync.js的数组.prototype.for

if(!Array.prototype.forEachAsync) {
    Array.prototype.forEachAsync = function (fn){
      return new Promise((resolve,reject)=>{
        this.forEach(async(item,index,array)=>{
            await fn(item,index,array);
            if(index === array.length-1){
                resolve('done');
            }
        })
      });
    };
  }

用法:

require('./Array.prototype.forEachAsync');

let count = 0;

let hello = async (items) => {

// Method 1 - Using the Array.prototype.forEach 

    await items.forEachAsync(async () => {
         try{
               await someAPICall();
           } catch(e) {
              console.log(e)
           }
        count++;
    });

    console.log("count = " + count);
}

someAPICall = () => {
    return new Promise((resolve, reject) => {
        setTimeout(() => {
            resolve("done") // or reject('error')
        }, 100);
    })
}

hello(['', '', '', '']); // hello([]) empty array is also be handled by default

方法3:

使用Promise.all

  await Promise.all(items.map(async (item) => {
        await someAPICall();
        count++;
    }));

    console.log("count = " + count);

方法4:传统循环或现代循环

// Method 4 - using for loop directly

// 1. Using the modern for(.. in..) loop
   for(item in items){

        await someAPICall();
        count++;
    }

//2. Using the traditional for loop 

    for(let i=0;i<items.length;i++){

        await someAPICall();
        count++;
    }


    console.log("count = " + count);

此解决方案还对内存进行了优化,因此您可以在10000个数据项和请求上运行它。这里的一些其他解决方案将使大型数据集上的服务器崩溃。

在TypeScript中:

export async function asyncForEach<T>(array: Array<T>, callback: (item: T, index: number) => Promise<void>) {
        for (let index = 0; index < array.length; index++) {
            await callback(array[index], index);
        }
    }

如何使用?

await asyncForEach(receipts, async (eachItem) => {
    await ...
})