使用 Java Streams 从文本文件中一次读取 X 行？

Question

我有一个“普通旧文本文件”，其中行以换行符结尾。由于任意原因，我需要一次读取和解析此文本文件 4 行（一般为 X）行。

我想使用 Java 流来完成这项任务，并且我知道我可以像这样将文件转换为流：

try (Stream<String> stream = Files.lines(Paths.get("file.txt""))) {
    stream.forEach(System.out::println);
} catch (IOException e) {
    e.printStackTrace();
}

但是如何使用 Java 的 Stream API 将文件“捆绑”成 4 行连续的组？

Answer 1

如果您想坚持使用流，我看到的唯一解决方案是编写您自己的自定义收集器。它不是为此目的而设计的，但您可以使用它。

private static final class CustomCollector {

    private List<String> list = new ArrayList<>();

    private List<String> acumulateList = new ArrayList<>();

    public void accept(String str) {
        acumulateList.add(str);
        if (acumulateList.size() == 4) { // acumulate 4 strings
            String collect = String.join("", acumulateList);
            // I just joined them in on string, you can do whatever you want
            list.add(collect);
            acumulateList = new ArrayList<>();
        }
    }

    public CustomCollector combine(CustomCollector other) {
        throw new UnsupportedOperationException("Parallel Stream not supported");
    }

    public List<String> finish() {
        if(!acumulateList.isEmpty()) {
            list.add(String.join("", acumulateList));
        }
        return list;
    }

    public static Collector<String, ?, List<String>> collector() {
        return Collector.of(CustomCollector::new, CustomCollector::accept, CustomCollector::combine, CustomCollector::finish);
    }
}

然后像这样使用它：

stream.collect(CustomCollector.collector());

Answer 2

如果您愿意使用RxJava，您可以使用它的buffer 功能：

Stream<String> stream = Files.lines(Paths.get("file.txt"))

Observable.fromIterable(stream::iterator)
          .buffer(4)                      // Observable<List<String>>
          .map(x -> String.join(", ", x)) // Observable<String>
          .forEach(System.out::println);

buffer 创建一个Observable 来收集特定大小列表中的元素。在上面的示例中，我通过map 添加了另一个转换以使列表更易于打印，但您可以根据需要转换Observable。例如，如果您有一个方法processChunk，它以List<String> 作为参数并返回String，您可以这样做：

Observable<String> fileObs =
    Observable.fromIterable(stream::iterator)
              .buffer(4)
              .map(x -> processChunk(x));

Answer 3

有一种方法可以使用标准 Java 8 Stream API 将文件内容分区并处理为 n-size 块。您可以使用Collectors.groupingBy() 将您的文件内容划分为块 - 您可以将它们收集为Collection<List<String>>，或者您可以在收集所有行时应用一些处理（例如，您可以将它们加入单个字符串）。

看看下面的例子：

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

public class ReadFileWithStream {

    public static void main(String[] args) throws IOException {
        // Path to a file to read
        final Path path = Paths.get(ReadFileWithStream.class.getResource("/input.txt")‌​.toURI());
        final AtomicInteger counter = new AtomicInteger(0);
        // Size of a chunk
        final int size = 4;

        final Collection<List<String>> partitioned = Files.lines(path)
                .collect(Collectors.groupingBy(it -> counter.getAndIncrement() / size))
                .values();

        partitioned.forEach(System.out::println);
    }
}

我的输入文件contains some numbers (one number at a line)，当我运行以下代码时，我得到如下内容：

[0, 0, 0, 2]
[0, -3, 2, 0]
[1, -3, -8, 0]
[2, -12, -11, -11]
[-8, -1, -8, 0]
[2, -1, 2, -1]
... and so on

Collectors.groupingBy() 还允许我使用不同的下游收集器。默认情况下使用Collectors.toList()，所以我的结果累积到List<String>，我得到Collection<List<String>>作为最终结果。

假设我想读取 4 大小的块，并且我想将块中的所有数字相加。在这种情况下，我将使用Collectors.summingInt() 作为我的下游函数，返回结果为Collection<Integer>：

final Collection<Integer> partitioned = Files.lines(path)
        .collect(Collectors.groupingBy(it -> counter.getAndIncrement() / size, Collectors.summingInt(Integer::valueOf)))
        .values();

输出：

2
-1
-10
-32
-17
2
-11
-49
... and so on

最后但并非最不重要。 Collectors.groupingBy() 返回一个映射，其中值按特定键分组。这就是为什么最后我们调用Map.values() 来获取此映射中包含的值的集合。

希望对你有帮助。

Answer 4

这是java.util.Scanner 的工作。在 Java 9 中，您可以简单地使用

try(Scanner s = new Scanner(PATH)) {
    s.findAll("(.*\\R){1,4}")
     .map(mr -> Arrays.asList(mr.group().split("\\R")))
     .forEach(System.out::println);
}

对于 Java 8，您可以使用 findAll 的后端端口 this answer。为该方法添加import static 后，您可以像这样使用它

try(Scanner s = new Scanner(PATH)) {
    findAll(s, Pattern.compile("(.*\\R){1,4}"))
        .map(mr -> Arrays.asList(mr.group().split("\\R")))
        .forEach(System.out::println);
}

请注意，匹配操作的结果是一个字符串，最多包含四行（最后一行少）。如果这适合您的后续操作，您可以跳过将该字符串拆分为单独的行。

您甚至可以使用MatchResult 的属性对块进行更复杂的处理，例如

try(Scanner s = new Scanner(PATH)) {
    findAll(s, Pattern.compile("(.*)\\R(?:(.*)\\R)?(?:(.*)\\R)?(?:(.*)\\R)?"))
        .flatMap(mr -> IntStream.rangeClosed(1, 4)
                           .mapToObj(ix -> mr.group(ix)==null? null: ix+": "+mr.group(ix)))
        .filter(Objects::nonNull)
        .forEach(System.out::println);
}

Answer 5

这是使用 Guava 的 Iterators.partition 方法的简单方法：

try (Stream<String> stream = Files.lines(Paths.get("file.txt""))) {

    Iterator<List<String>> iterator = Iterators.partition(stream.iterator(), 4);

    // iterator.next() returns each chunk as a List<String>

} catch (IOException e) {
    // handle exception properly
}

这仅适用于顺序处理，但如果您从磁盘读取文件，我很难想象并行处理有什么好处......

---

编辑：如果您愿意，可以将其再次转换为流，而不是使用迭代器：

Stream<List<String>> targetStream = StreamSupport.stream(
      Spliterators.spliteratorUnknownSize(iterator, Spliterator.ORDERED),
      false);