Akka Streams：运行过程中的流

Question

如何在流程上创建流程？假设我有一个通过标准输入获取输入并通过标准输出发送输出的进程。我想有一个流过它。那么 Flow 将首先启动流程，控制其输入和输出流，然后充当 Map，通过将输入馈送到流程来将输入映射到输出？并最终在流结束时终止进程。另外，在这种情况下，背压控制如何工作？

Answer 1

这是一个非常基本的Flow that is "coupled"，因此它会因输入或输出（接收器/源）的中断而终止。为简单起见，我假设没有标准输入的情况，但您可以调整它以添加不会被忽略的 Sink 和 pipes input into the process：

// 从“ls”命令捕获标准输出 val runCmd：来源[消息，未使用] = { Source.fromIterator[Message](() => Process("ls").lineStream.toIterator.map(line => TextMessage.Strict(line))) } // 创建用于向客户端发送标准输出源的流 val messageStdout: Flow[Any, Message, NotUsed] = Flow .fromSinkAndSourceCoupled(Sink.ignore, runCmd)

至于背压，请注意上面我的代码示例是一个阻塞的map 操作（一次一个line）。随着处理速度的增长，您可以开发一种异步解决方案，该解决方案具有最大并行度，遵循mapasync 定义的背压。