Java 8流和并发写

Question

我有这样的代码

    public static void main(String[] args) throws Exception {
      long start = System.currentTimeMillis();
      List<String> matches = new Vector<>(); // Race condition for ArrayList??
      BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream("AHugeFile.txt")));
      BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(new FileOutputStream("output.txt")));
      reader.lines().parallel()
         .filter(s -> s.matches("someFancyRegEx"))
         .forEach(s -> {
               matches.add(s);
               try {
                  writer.write(s);
                  writer.newLine();
               } catch (Exception e) {
                  System.out.println("error");
               }
            }
         );
      out.println("Processing took " + (System.currentTimeMillis() - start) / 1000 + " seconds and matches " + matches.size());
      reader.close();
      writer.flush();
      writer.close();
   }

我注意到，如果我在第3行用ArrayList替换Vector，每次在匹配中都会得到不同的结果。我只是想让我的手脏流，但假设forEach执行并发试图写入数组列表错过了一些写!对于Vector，结果是一致的。

我有两个问题:

1. 是我的推理关于ArrayList导致一个正确的RACE ?
2. 考虑到'写'也在同一终端操作中写入文件，'写'可能会遗漏一些行吗?在我的测试中，运行该程序几次，结果似乎与写入的正确行数一致。

Answer 1

首先:定义你是否关心行写入的顺序;.forEach()剥离了Spliterator的ORDERED特性。

第二:使用Java 8提供的工具;有两种非常方便的方法:Files.lines()和Files.write()。

第三:正确处理你的资源!在代码中不能保证文件描述符将被正确关闭。

第四:.matches()每次都会重新创建一个Pattern，并且您总是使用相同的正则表达式进行过滤…你在浪费资源。

第五:考虑到BufferedWriter的写方法是同步的，你不会从并行性中获得太多好处。

我是这样做的:

public static void writeFiltered(final Path srcFile, final Path dstFile,
    final String regex)
    throws IOException
{
    final Pattern pattern = Pattern.compile(regex);
    final List<String> filteredLines;
    try (
        // UTF-8 by default
        final Stream<String> srcLines = Files.lines(srcFile);
    ) {
        filteredLines = srcLines.map(pattern::matcher)
            .filter(Matcher::matches)
            .collect(Collectors.toList());
    }
    // UTF-8 by default
    Files.write(dstFile, filteredLines);
}

Answer 2

1. ArrayList不是一个同步集合，所以它会导致RACE条件。所有改变矢量状态的方法都是同步的，所以你在这里没有发现任何问题。

2. BufferedWriter的写方法是同步的，所以所有的写操作在线程之间是一致的。因此，文件中的单个写操作将是线程安全的。但是你需要显式地处理同步，使其在线程间保持一致。

下面是Java 6中write方法的代码片段。

public void write(String s, int off, int len) throws IOException {
    synchronized (lock) {
        ensureOpen();
        int b = off, t = off + len;
        while (b < t) {
            int d = min(nChars - nextChar, t - b);
            s.getChars(b, b + d, cb, nextChar);
            b += d;
            nextChar += d;
            if (nextChar >= nChars)
                flushBuffer();
            }
        }
    }
}