java - 即使抛出异常，如何迭代流？

Question

stream.map(obj -> doMap(obj)).collect(Collectors.toList());

private String doMap(Object obj) {
    if (objectIsInvalid) {
    throw new ParseException("Object could not be parsed");
    }
}

问题:我如何抛出异常并让流迭代知道它不应该中断整个迭代，而是继续下一个元素(并最终记录失败的对象)？

Answer 1

这里有一个奇怪的技巧可以用来改进异常处理。

假设您的映射器函数是这样的:

String doMap(Object obj) {
    if (isInvalid(obj)) {
        throw new IllegalArgumentException("info about obj");
    } else {
        return obj.toString();
    }
}

如果对象有效则返回结果，如果对象无效则抛出异常。不幸的是，如果您将其直接粘贴到管道中，任何错误都会停止管道执行。您想要的是类似于“任一”类型的东西，它可以包含一个值或一个错误指示符(这在 Java 中是一个异常)。

原来 CompletableFuture 可以保存一个值或一个异常。尽管它旨在用于异步处理(此处并未发生)，但我们只需稍微扭曲它即可用于我们的目的。

首先，给定要处理的对象的 stream，我们调用包装在对 supplyAsync 的调用中的映射函数:

 CompletableFuture<String>[] cfArray = 
        stream.map(obj -> CompletableFuture.supplyAsync(() -> doMap(obj), Runnable::run))
              .toArray(n -> (CompletableFuture<String>[])new CompletableFuture<?>[n]);

(不幸的是，通用数组创建会给出未经检查的警告，必须将其抑制。)

奇怪的结构

 CompletableFuture.supplyAsync(supplier, Runnable::run)

在提供的执行器 Runnable::run 上“异步”运行供应商，它只是立即在此线程中运行任务。换句话说，它同步运行供应商。

诀窍在于从该调用返回的 CompletableFuture 实例包含来自供应商的值(如果它正常返回)，或者它包含一个异常(如果供应商抛出异常)。 (我在这里不考虑取消。)然后我们将 CompletableFuture 实例收集到一个数组中。为什么是数组？这是下一部分的设置:

CompletableFuture.allOf(cfArray).join();

这通常会等待 CF 数组完成。由于它们已同步运行，因此它们应该已经全部完成。对于这种情况，重要的是如果数组中的任何 CF 异常完成，join() 将抛出CompletionException。如果连接正常完成，我们可以简单地收集返回值。如果连接抛出异常，我们可以传播它，或者我们可以捕获它并处理存储在数组中的 CF 中的异常。例如，

try {
    CompletableFuture.allOf(cfArray).join();
    // no errors
    return Arrays.stream(cfArray)
                 .map(CompletableFuture::join)
                 .collect(toList());
} catch (CompletionException ce) {
    long errcount =
        Arrays.stream(cfArray)
              .filter(CompletableFuture::isCompletedExceptionally)
              .count();
    System.out.println("errcount = " + errcount);
    return Collections.emptyList();
}

如果全部成功，这将返回一个值列表。如果有任何异常，这将计算异常的数量并返回一个空列表。当然，您可以轻松地做其他事情，比如记录异常的详细信息、过滤掉异常并返回有效值列表等。