飞镖流并等待

Question

我无法理解以下代码的流程。 代码应处理 MERGE_SIZE 行（本次运行为 3 行），将这些行保存到“阶段”文件，然后处理接下来的 3 行，依此类推。

对 savePhase 的调用有一个等待，所以我希望 savePhase 在处理其他行之前完成。

正如您在下面的输出中看到的，每一行都是进程，然后 savePhase 调用完成。

Future _sort() async {
    var completer = Completer<void>();
    var instance = 0;
    var lineCount = MERGE_SIZE;

    var phaseDirectory = Directory.systemTemp.createTempSync();

    var list = <String>[];

    var sentToPhase = false;

    await File(filename)
        .openRead()
        .map(utf8.decode)
        .transform(LineSplitter())
        .forEach((l) async {
      list.add(l);
      print('$l linecount:$lineCount');
      lineCount--;

      if (lineCount == 0) {
        lineCount = MERGE_SIZE;
        instance++;
        sentToPhase = true;
        await savePhase(phaseDirectory, 1, instance, list, lineDelimiter);
        list.clear();
        print('savePhase completed');
      }
    });

哪个输出

9 line linecount:3
8 line linecount:2
7 line linecount:1
6 line linecount:3
5 line linecount:2
4 line linecount:1
3 line linecount:3
2 line linecount:2
1 line linecount:1
savePhase completed
savePhase completed
savePhase completed

这与 openRead 用于传递读取行的流有关吗？

我以为我已经 await 想通了，但显然不是:)

Answer 1

未测试您的程序，但我相当确定您的问题是您希望 forEach() 方法等待每个 Future 在下一次调用之前完成，但情况并非如此。

尝试看看以下解决方案，它们或多或少是相同的问题： Sequential processing of a variable number of async functions in Dart

程序流程

因此，在您的代码中发生的情况是，您正在读取的文件似乎足够小，以至于可以在读取文件时使用的缓冲区之一中一次性读取整个内容。这将解释为什么您会在 savePhase completed 之前看到多个 line linecount 行。

如前所述，Stream 上的 forEach() 方法没有考虑到作为参数给出的方法确实返回了应该等待的 Future。您可以在此处显示的实现中看到： https://api.dart.dev/stable/2.7.0/dart-async/Stream/forEach.html

这意味着调用forEach() 返回的Future 确实在所有行都已处理后才完成，但不会等待为每一行生成的每个Future（请记住，async 方法总是返回一个Future，不管它是否包含await）。

由于您还在每个生成的 Future 之间使用共享变量，因此您也会在这里得到一些时髦的行为，例如共享相同的列表，但之后也会清除列表。所以这里有可能出现错误。