在Java8中过滤和映射parallelStream后如何收集结果？

Question

我想从另一个集合中提取一个对象集合。要过滤的对象必须是特定类型（或子类型）并且必须与给定的Shape 相交。我想用parallelStream做它

我有以下代码：

public class ObjectDetector {
...
    public ObjectDetector(final Collection<WorldObject> objects,
        final BiFunction<Shape, Shape, Boolean> isIntersecting) {
    ...
    }

    public List<ISensor> getSonarObjects(final Shape triangle) {
        return selectIntersecting(triangle, ISensor.class);
    }

    private <T> List<T> selectIntersecting(Shape triangle, Class<T> type) {
        return objects.parallelStream()
                .filter(o -> type.isInstance(o) && isIntersecting.apply(o.getShape(), triangle))
                .map(o -> type.cast(o)).collect(Collectors.toList());

有问题的部分在List<T> selectIntersecting(Shape triangle, Class<T> type) 方法中，其中objects 是Collection，isIntersecting 是BiFunction<Shape,Shape,Boolean>。

当我使用stream() 而不是parallelStream() 时，我的所有测试都是绿色的。所以我可以假设过滤和映射逻辑工作正常。但是，当我尝试使用 parallelStream() 时，我的测试出乎意料地失败了。我能够观察到的唯一一致性是返回的List<T> 的size() 小于或等于（但当然永远不会大于）我期望的大小。

例如一个失败的测试用例：

int counter = 0;
public BiFunction<Shape, Shape, Boolean> every2 = (a, b) -> {
    counter++;
    return counter % 2 == 0 ? true : false;
};

@Test
public void getEvery2Sonar() {
    assertEquals("base list size must be 8",8,list.size());
    ObjectDetector detector = new ObjectDetector(list, every2);
    List<ISensor> sonarable = detector.getSonarObjects(triangle);
    assertEquals("number of sonar detectables should be 3", 3, sonarable.size());
}

而测试结果是：

失败的测试：getEvery2Sonar(hu.oe.nik.szfmv.environment.ObjectDetectorTest)：声纳可检测的数量应为 3 预期： 但为：

在我的理解中 - 正如它所写的 here - 可以将 parallelStream 收集到非并发 Collection 中。

我也尝试在并行教程page上找到一些线索，但我仍然一无所知。

有人可以解释一下我做错了什么吗？

Answer 1

您的谓词函数有副作用 - 这对于 parallelStream 来说会很糟糕，因为输入流中的评估顺序是不确定的，而且您没有锁定您的可变状态。

确实，filter 的文档声明^*谓词必须是 stateless。

我不确定您要在这里实现什么行为，所以我不确定合适的“修复”可能是什么。

---

_{* 没有双关语。}