Dart流控制器如何实现并发

Question

假设我有一个 StreamController，并且我想使用此控制器添加事件以处理来自应用程序的多个请求（并行）。

我实现的内容如下：

在

main.dart

    final controller = StreamController<Req>.broadcast();

    final resStream = controller.stream.asyncMap<Res>((request) {
      switch (request.method) {
        case "get":
          return  Future.delayed(
            const Duration(seconds: 10),
            () => const Res('response after 10s'),
          );
        case "post":
        default:
          return  Future.delayed(
            const Duration(seconds: 5),
            () => const Res('response after 5s'),
          );
      }
    });

    resStream.listen((event) {
      print('+++ response from stream: ${event.value}');
    });

    controller.add(const Req('get'));
    controller.add(const Req('post'));

班级

req.dart

&

res.dart

class Req {
  const Req(this.method);

  final String method;
}

class Res {
  const Res(this.value);

  final String value;
}

在我运行该代码之后。我意识到输出的顺序与我之前添加到 StreamController 中的顺序相同。

+++ response from stream: response after 10s
+++ response from stream: response after 5s

在我看来，

Req('post')

的输出必须首先出现，然后再出现到另一个，如下所示：

+++ response from stream: response after 5s
+++ response from stream: response after 10s

任何人都可以帮助我解决这种情况。非常感谢！

Answer 1

asyncMap

方法正在对

await

值序列调用

Future

，按添加到流中的顺序一次调用每个值。 这是对结果进行序列化，因此每个

await

必须在下一个

Future

完成之前完成。

await

API 文档中简要提到了

Future<>

对象的处理

https://api.dart.dev/stable/dart-async/Stream/asyncMap.html

让它们同时运行的一种方法是将多个

asyncMap

对象收集到一个

Future<Res>

中，然后在其上使用

List<Future<Res>>

。 示例代码中的延迟将并行运行。