同时将流中的对象添加到两个不同的列表中

Question

如何同时将一个流中的对象添加到两个不同的列表中

目前我正在做以下操作：

body.getSurroundings().parallelStream()
                .filter(o -> o.getClass().equals(ResourcePoint.class))
                .map(o -> (ResourcePoint)o)
                .filter(o -> !resourceMemory.contains(o))
                .forEach(resourceMemory::add);

我想将来自我的流的对象添加到一个名为“resourceMemory”的链表中，但同时我也想将相同的对象添加到另一个列表中，但我找不到相应的语法。这是否可能，还是我需要为每个列表拥有两份代码副本？

- Flux

---

你可以使用 peek() 或自定义收集器。 - Peter Lawrey

你是指 o instanceof ResourcePoint 还是 o.getClass() == ResourcePoint.class？决定一个，但不要使用 o.getClass().equals(ResourcePoint.class)，那会混淆实际意图。此外，当使用并行流时，你的代码存在多个问题。请仔细阅读 https://docs.oracle.com/javase/8/docs/api/?java/util/stream/package-summary.html。 - Holger

严格回答这个问题是.forEach(o -> { resourceMemory.add(o); myOtherList.add(o); })。但请注意Holger的评论。为了修复这段代码，您需要更多地说明您真正想要实现什么。 - Tunaki

Answer 1

在尝试扩展您的代码之前，您应该先了解几个基本错误。

首先，forEach不能保证元素处理的特定顺序，因此即使是对于顺序流，它也很可能不是添加到List的正确工具。但是，如果使用并行流将其用于添加到诸如LinkedList之类的非线程安全集合，则完全是错误的，因为此操作将被同时执行。

但是，即使resourceMemory是线程安全的集合，您的代码仍然有问题，因为filter条件与终端操作之间存在干扰。 .filter(o -> !resourceMemory.contains(o))查询您正在修改的相同列表，并且不难理解即使是使用线程安全集合也会出现问题：

两个或更多线程可能会处理过滤器并发现元素未包含在列表中，然后所有这些线程都会添加该元素，这与您明显的无重复意图相矛盾。

您可以使用forEachOrdered，它将按顺序且非并发地执行操作：

body.getSurroundings().parallelStream()
    .filter(o -> o instanceof ResourcePoint)
    .map(o -> (ResourcePoint)o)
    .forEachOrdered(o -> {// not recommended, just for explanation
        if(!resourceMemory.contains(o))
            resourceMemory.add(o);
    });

这个方法是可行的，你可以在这个操作中添加到另一个列表中，但这远不是推荐的编码风格。此外，这个终端操作会与所有处理线程同步，这将破坏任何并行处理的潜在好处，特别是当流管道中最耗时的操作是在 LinkedList 上调用 contains 方法时，它只能单线程运行。

正确的收集流元素到列表中的方法是使用 collect 方法，正如其名称所示：

List<ResourcePoint> resourceMemory
    =body.getSurroundings().parallelStream()
        .filter(o -> o instanceof ResourcePoint)
        .map(o -> (ResourcePoint)o)
        .distinct()                    // no duplicates
        .collect(Collectors.toList()); // collect into a list

这并不返回一个 LinkedList，但你应该仔细重新考虑是否真的需要一个 LinkedList。在 99% 的情况下，你并不需要它。如果你确实需要一个 LinkedList，你可以用 Collectors.toCollection(LinkedList::new) 替换 Collectors.toList()。

现在，如果你真的必须添加到一个已经包含元素的、由你无法控制的现有列表中，你应该考虑上面提到的事实，即你必须确保对非线程安全列表的单线程访问，因此从并行流中进行这样的操作没有任何好处。在大多数情况下，更有效的做法是让流独立于该列表工作，并在单线程步骤中添加结果。

Set<ResourcePoint> newElements=
    body.getSurroundings().parallelStream()
        .filter(o -> o instanceof ResourcePoint)
        .map(o -> (ResourcePoint)o)
        .collect(Collectors.toCollection(LinkedHashSet::new));
newElements.removeAll(resourceMemory);
resourceMemory.addAll(newElements);

在这里，我们使用LinkedHashSet来收集元素，这意味着维护遇到的顺序并排序出新元素中的重复项，然后在新元素上使用removeAll来删除目标列表中现有的元素（这里我们从临时集合的哈希集特性中受益），最后将新元素添加到目标列表中。正如解释的那样，对于一个非线程安全的目标集合，这必须是单线程完成的。

使用此解决方案将newElements添加到另一个目标集合很容易，比编写自定义收集器在流处理期间生成两个列表要简单得多。但请注意，上述流操作太便宜了，无法假设从并行处理中获得任何好处。您需要非常大量的元素才能弥补初始的多线程开销。甚至可能不存在某个数字，使其值得付出。

- Holger

Answer 2

代替

.forEach(resourceMemory::add)

你可以调用

.forEach(o -> {
   resourceMemory.add(o);
   otherResource.add(o);
 })

或者将加法操作放在一个单独的方法中，这样您就可以提供一个方法引用。

.forEach(this::add)

void add(ResourcePoint p) {
   resourceMemory.add(o);
   otherResource.add(o);
}

请记住，使用并行流时，插入顺序可能会在每次运行时不同。

- Gerald Mücke

---

@Holger，你的评论可能放错了位置。我认为它并不是指上面的答案。 - Markus

@Markus，不，Holger的评论是正确的，我编辑了我的答案，损坏的部分已经被删除了，如果代码有问题就修复/删除它 :) - Gerald Mücke

应强调的是，传递给 forEach 的操作不仅以任意顺序被调用，而且是并发调用的，因此，这只适用于线程安全的集合。 - Holger