从并行流中捕获异常

Question

我有一堆列作为 csv 文件中的字符串数组。现在我想解析它们。由于这种解析需要日期解析和其他不那么快的解析技术，我正在考虑并行性（我计时了，这需要一些时间）。我的简单方法：

Stream.of(columns).parallel().forEach(column -> 
    result[column.index] = parseColumn(valueCache[column.index], column.type));

Columns 包含ColumnDescriptor 元素，它只有两个属性，要解析的列索引和定义如何解析它的类型。没有其他的。 result 是一个 Object 数组，它采用结果数组。

现在的问题是 parse 函数抛出一个 ParseException，我在调用堆栈中处理得更远。由于我们在这里是并行的，它不能只是被抛出。处理这个问题的最佳方法是什么？

我有这个解决方案，但我有点畏缩阅读它。有什么更好的方法？

final CompletableFuture<ParseException> thrownException = new CompletableFuture<>();
Stream.of(columns).parallel().forEach(column -> {
    try {
        result[column.index] = parseColumn(valueCache[column.index], column.type);
    } catch (ParseException e) {
        thrownException.complete(e);
    }});

if(thrownException.isDone())
    //only can be done if there is a value set.
    throw thrownException.getNow(null);

注意：我不需要所有的例外。如果我按顺序解析它们，我也只会得到一个。所以没关系。

Answer 1

问题是你的错误前提“因为我们在这里是并行的，所以不能随便抛出。”没有规范禁止在并行处理中抛出异常。您可以像在顺序流中一样在并行流中抛出该异常，如果它是已检查异常，则将其包装在未检查异常中。

如果线程中至少抛出一个异常，forEach 调用会将其（或其中之一）传播给调用者。

您可能遇到的唯一问题是，当前实现在遇到异常时不会等待所有线程完成。这可以解决使用

try {
    Arrays.stream(columns).parallel()
        .forEach(column -> 
            result[column.index] = parseColumn(valueCache[column.index], column.type));
} catch(Throwable t) {
    ForkJoinPool.commonPool().awaitQuiescence(1, TimeUnit.MINUTES);
    throw t;
}

但通常情况下，您不需要它，因为在特殊情况下您不会访问并发处理的结果。

Answer 2

我觉得问题比较多，串行解析的时候一般是怎么做的呢？

您是否在第一个异常处停止，并停止整个过程？在这种情况下，将异常包装在运行时异常中，让流中止并抛出它。捕获包装器异常，解包并处理。

您会跳过不良记录吗？然后要么 1. 在某个地方跟踪 List 中的错误，要么 2. 创建一个可以保存解析结果或错误的包装器对象（不要跟踪异常本身，只跟踪描述错误所需的最小值）。

之后检查第一个选项的列表中是否有错误，或者为第二个选项显示不同的错误记录。