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同时执行，试图写到ArrayList，而ArrayList漏掉了一些写！对于向量，结果是一致的。

我有两个问题：

1. 我对ArrayList引起比赛的推理正确吗？
2. 考虑到“写”也是在同一个终端操作中写入文件，“写”可能会遗漏一些行吗？在我的测试中，运行程序几次，结果似乎与所写出的正确行数一致。

Answer 1

1. ArrayList不是一个同步的集合，所以是的，它会导致争用条件。更改向量状态的所有方法都是同步的，因此您在那里没有发现任何问题。
2. BufferedWriter的写方法是同步的，所以所有的写在线程之间都是一致的。因此，文件中的单个写入操作将是线程安全的。但是，您需要显式地处理同步，以使其在线程之间保持一致。

以下是Java 6中写方法的代码片段。

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();
            }
        }
    }
}

Answer 2

首先要做的是:定义您是否关心写行的顺序； (been there, done that)。

第二:使用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);
}