有没有办法用 Akka 流将 Path 元素流展平为内容行流？答案

【问题标题】：Is there any way to flatten a stream of Path elements to a stream of content lines with Akka streams?有没有办法用 Akka 流将 Path 元素流展平为内容行流？
【发布时间】：2015-05-17 17:44:12
【问题描述】：

我正在尝试创建一个处理 Path 元素的 javadsl Flow。它应该发出与 Path 元素关联的文件的内容行。换句话说，我认为我需要将 Path 元素流扁平化为 String 元素流。

Flow 确实有 flatten 方法，但它涉及使用 FlattenStrategy，我不确定如何在我的情况下使用它。

final Flow<Path, String, BoxedUnit> toFileLines = Flow
    .<Path>create()
    .flatten(FlattenStrategy.

非常感谢任何帮助！

编辑1：因此，据我了解，使用 StreamReader 读取文件并在阅读器到达“\ n”时发出新字符串可能是个好主意。所以现在的问题是如何从一个单一的转换方法发出多个元素。像这样

final Flow<Path, String, BoxedUnit> toFileLines = Flow
    .<Path>create()
    .mapAsync(
            //create streamreader
            //while streamreader has.next
                //read line until \n
                //emit line
    );

这可能吗？

【问题讨论】：

试着把它想象成一个多步骤的过程：0.) 你得到一个路径流作为输入 1.) 流式传输每个文件的内容 2.) 将每个文件拆分成行 3.) 连接所有文件中的所有行。 FlattenStrategy 将仅用于最后一步。它指定来自传入源的内容应如何“展平”，即是否应将它们连接或合并（尚未实现）。
@jrudolph 感谢您的回复。我对您摘要的第 1 步和第 2 步感兴趣。我应该如何以简洁且最好是并行的方式流式传输文件的内容（字符串行）？这些文件可以达到很大的大小，我想懒惰地阅读它们而不是一次阅读整个文件
例如可以使用Files.lines 的Java8 流实现吗？
你可以做 1.) 使用SynchronousFileSource。 2.) 即将推出的 RC4 将更容易，其中将包括用于拆分字符串流的组件：github.com/akka/akka/pull/17446/files

标签： stream akka akka-stream

【解决方案1】：

我建议使用Flow.flatMapConcat。首先，您必须编写一个可以从文件中生成行的 Source，然后将这些 Source 放在一起 flatMapConcat。还有一点我从docs 得到的使用框架的 ByteString 解析：

import java.io.File;
import akka.stream.javadsl.FileIO;
import akka.stream.javadsl.Flow;
import akka.stream.io.Framing;
import akka.util.ByteString;

int maxLineSize = 1024;

final ByteString delim = ByteString.fromString("\r\n");

final Flow<String, String, BoxedUnit> pathsToContents = 
  Flow.of(String.class)
      .flatMapConcat(path -> FileIO.fromFile(new File(path))
                                   .via(Framing.delimiter(delim, maxLineSize,true))
                                   .map(byteStr -> byteStr.utf8String()));

【讨论】：