使用 parallelStream 时抛出 InterruptedException - Java [重复]

Posted 2023-04-18

【中文标题】使用 parallelStream 时抛出 InterruptedException - Java [重复]

【英文标题】：Throw InterruptedException when using parallelStream - Java [duplicate]

【发布时间】：2018-07-26 01:08:42

【问题描述】：

我有一个嵌套for循环的方法如下：

public MinSpecSetFamily getMinDomSpecSets() throws InterruptedException 
    MinSpecSetFamily result = new MinSpecSetFamily();
    ResourceType minRT = this.getFirstEssentialResourceType();
    if (minRT == null || minRT.noSpecies()) 
        System.out.println("There is something wrong with the "
                + "minimal rticator, such as adjacent to no species. ");
    
    for (Species spec : minRT.specList) 
        ArrayList<SpecTreeNode> leafList = this.getMinimalConstSpecTreeRootedAt(spec).getLeaves();
        for (SpecTreeNode leaf : leafList) 
            result.addSpecSet(new SpecSet(leaf.getAncestors()));
        
    
    return result;

这工作正常，但应用程序对性能至关重要，因此我将方法修改为使用parallelStream()，如下所示：

public MinSpecSetFamily getMinDomSpecSets() throws InterruptedException 
    ResourceType minRT = this.getFirstEssentialResourceType();
    if (minRT == null || minRT.noSpecies()) 
        System.out.println("There is something wrong with the "
                + "minimal rticator, such as adjacent to no species. ");
    

    MinSpecSetFamily result = minRT.specList.parallelStream()
            .flatMap(spec -> getMinimalConstSpecTreeRootedAt(spec).getLeaves().parallelStream())
            .map(leaf -> new SpecSet(leaf.getAncestors()))
            .collect(MinSpecSetFamily::new, MinSpecSetFamily::addSpecSet, MinSpecSetFamily::addMSSF);
    return result;

在我想在 'getLeaves()' 方法中引入 InterruptedException 之前，这一直很好。现在 parallelStream 版本将无法编译，因为它说我有一个未报告的 InterruptedException 必须被捕获或声明为抛出。我认为这是因为parallelStream 在多个线程上运行。我的 IDE 建议的 try/catch 块组合无法解决问题。

Interrupt parallel Stream execution发布的第二个解决方案 建议我可以使用ForkJoinPool 解决问题，但我一直无法弄清楚如何修改我的方法以使用这种方法。

【问题讨论】：

An InterruptedException 是一个检查异常，因此是编译问题。并行运行它实际上并不重要，您需要一种方法将此异常从流内部传递到流外部。

【参考方案1】：

如果你想坚持你当前的设计，你只需要捕获异常：

.flatMap(spec -> 
     try 
       return getMinimalConstSpecTreeRootedAt(spec).getLeaves().parallelStream();
      catch (InterruptedException e) 
       // return something else to indicate interruption
       // maybe an empty stream?
     
 ).map(...)

请注意，并行流的并行流可能是不必要的，仅并行化***流在性能方面可能就足够了。

【讨论】：

在您的回答中 parallelStream 之后有一个额外的 ) ，但是谢谢，这完成了工作。另外，我按照建议将第二个 parallelStream 更改为 .stream()。

再想一想，这并不能解决问题。如果我在 catch 块中返回 null，则线程会继续为每个额外的“规范”运行。如果任何规范为null，我可以跳出parallelStream并从整个方法中返回null吗？

@SteveW 你可能会抛出一个未经检查的异常（但我不能 100% 确定它是否会停止整个流，但感觉有点 hacky）。听起来你最好使用不同的设计，例如使用 ExecutorService：提交任务并存储期货，然后在期货上调用 .get() 并做你的事情。如果任何任务被打断，您将知道并能够停止处理。

好的，那条评论让我有点过头了。但是，如果您建议手动创建所需的线程，我认为我做不到。我用 parallelStream 替换了嵌套的 for 循环（本质上是串行的），速度提高了 50%。如果我不使用 parallelStream，就我所见，我可能最好恢复到嵌套的 for 循环。