利用并行流高效处理文件比较

Question

因此，我有多个txt文件，比如txt1,txt2,...，每行都有4到22个字符之间的文本，还有另一个具有类似值的txt文件，比如bigText。我们的目标是检查bigTxt中发生在任何txt文件中的所有值，并输出这些值(我们保证，如果bigTxt的任何行都在任何txt文件中，则与该行的匹配只发生一次)。到目前为止，我找到的最好的解决方案是有效的，但效率有点低。基本上，看起来是这样的：

txtFiles.parallelStream().forEach(file->{
   List<String> txtList = listOfLines of this txtFile;
   streamOfLinesOfBigTxt.forEach(line->{
         if(txtList.contains(line)){
            System.out.println(line);
            //it'd be great if we could just stop this forEach loop here
            //but that seems hardish
         }
   });
});

(注意:我尝试在这里使用Honza的“坏主意”解决方案来突破forEach：Break or return from Java 8 stream forEach?，但这肯定不是我想要的，因为它实际上使代码慢了一点或大致相同)，小问题是，即使在一个文件找到了bigTxt文件和其他txt文件之间的一行的匹配之后，其他txt文件仍然试图使用该行搜索检查(尽管我们已经找到了一个匹配项，这就足够了)。我试图阻止这种情况的方法是首先在bigTxt行上迭代(不是并行的，而是并行地遍历每个txt文件)，使用java的anyMatch和我得到的“流已经被修改或关闭”类型的错误，我后来理解这是因为anyMatch正在终止。因此，在一个txt文件的一行上只调用anyMatch一次之后，该流就不再适用于我以后的处理了。我想不出一种正确使用findAny的方法，我也不认为allMatch是我想要的，因为并不是来自bigTxt的每个值都一定在一个txt文件中。这方面的任何(并行)解决方案(甚至不严格包括Java 8中的内容)都是受欢迎的。谢谢。

Answer 1

如果streamOfLinesOfBigTxt是一个Stream，那么当您尝试使用外部流的forEach多次处理该流时，您将在问题中发布的代码中得到相同的错误。还不清楚为什么您没有注意到这一点，但也许您总是在程序开始处理第二个文件之前就停止了它？毕竟，用两个行数的乘积线性地搜索大文件规模的每一行的行的List所需的时间。

当您说要“检查bigTxt中发生在任何txt文件中的所有值并输出这些值”时，您可以直接地这样做：

Files.lines(Paths.get(bigFileLocation))
     .filter(line -> txtFiles.stream()
                 .flatMap(path -> {
                         try { return Files.lines(Paths.get(path)); }
                         catch (IOException ex) { throw new UncheckedIOException(ex); }
                     })
                 .anyMatch(Predicate.isEqual(line)) )
    .forEach(System.out::println);

这会造成短路，但仍然存在着处理时间与n×m相适应的问题.更糟糕的是，它将重新打开并反复读取the文件。

如果您想避免这种情况，在RAM中存储数据是不可避免的。如果存储它们，首先可以选择支持比线性查找更好的存储：

Set<String> matchLines = txtFiles.stream()
    .flatMap(path -> {
        try { return Files.lines(Paths.get(path)); }
        catch (IOException ex) { throw new UncheckedIOException(ex); }
    })
    .collect(Collectors.toSet());

Files.lines(Paths.get(bigFileLocation))
     .filter(matchLines::contains)
     .forEach(System.out::println);

现在，该文件的执行时间与所有文件的行数之和而不是产品的行数之和进行缩放。但是它需要为txtFiles的所有不同行临时存储。

如果大文件在一起比其他文件有更少的不同行，并且顺序并不重要，那么您可以将大文件的行存储在一个集合中，然后动态检查txtFiles的行。

Set<String> matchLines
    = Files.lines(Paths.get(bigFileLocation)).collect(Collectors.toSet());

txtFiles.stream()
        .flatMap(path -> {
            try { return Files.lines(Paths.get(path)); }
            catch (IOException ex) { throw new UncheckedIOException(ex); }
        })
        .filter(matchLines::contains)
        .forEach(System.out::println);

这依赖于以下属性:所有匹配的行在所有这些文本文件中都是唯一的，正如您在问题中所述。

我不认为，这里的并行处理会有任何好处，因为I/O速度将主导执行。