我有一个用Java 8流表示的数据集：
Stream<T> stream = ...;
我可以看到如何过滤它以获得随机子集 - 例如
Random r = new Random();
PrimitiveIterator.OfInt coin = r.ints(0, 2).iterator();
Stream<T> heads = stream.filter((x) -> (coin.nextInt() == 0));
我也可以看到如何将这个流减少，例如，得到代表数据集的两个随机一半的列表，然后将它们转换回流。 但是，有没有一种直接从初始流生成两个流的方法？类似于
(heads, tails) = stream.[some kind of split based on filter]
感谢任何见解。
这可以使用一个收集器来完成。
Collectors.partitioningBy()工厂函数。这将创建一个Map<Boolean, List>，并根据 Predicate 将项目放入其中一个列表。
注意：由于流需要完整消耗，因此无法在无限流上工作。而且由于流已经被消费，该方法只是把它们放在列表中，而不是创建一个新的带有内存的流。如果需要将列表作为输出，请随时对其进行流操作。
另外，在您提供的仅包含头信息的示例中也不需要迭代器。
Random r = new Random();
Map<Boolean, List<String>> groups = stream
.collect(Collectors.partitioningBy(x -> r.nextBoolean()));
System.out.println(groups.get(false).size());
System.out.println(groups.get(true).size());
Collectors.groupingBy() 工厂方法。Map<Object, List<String>> groups = stream
.collect(Collectors.groupingBy(x -> r.nextInt(3)));
System.out.println(groups.get(0).size());
System.out.println(groups.get(1).size());
System.out.println(groups.get(2).size());
如果流不是Stream，而是原始流之一，比如IntStream，那么这个.collect(Collectors)方法就不可用了。你必须手动地完成，没有收集器工厂。它的实现看起来像这样：
[自 2020-04-16 起的示例 2.0]
IntStream intStream = IntStream.iterate(0, i -> i + 1).limit(100000).parallel();
IntPredicate predicate = ignored -> r.nextBoolean();
Map<Boolean, List<Integer>> groups = intStream.collect(
() -> Map.of(false, new ArrayList<>(100000),
true , new ArrayList<>(100000)),
(map, value) -> map.get(predicate.test(value)).add(value),
(map1, map2) -> {
map1.get(false).addAll(map2.get(false));
map1.get(true ).addAll(map2.get(true ));
});
stream.boxed().collect(...);！它会按照所述进行操作：将原始的 IntStream 转换为装箱的 Stream<Integer> 版本。 - YoYo(map, x) -> map.get(p.test(x)).add(x)代替(map, x) -> { boolean partition = p.test(x); List<Integer> list = map.get(partition); list.add(x); }。此外，我不明白为什么collect操作不能是线程安全的。它的工作方式与Collectors.partitioningBy(p)非常接近，并且正常工作。但是，当不使用boxed()时，我会使用IntPredicate而不是Predicate<Integer>，以避免两次装箱。 - Holger我自己遇到了这个问题，我认为分叉流有一些可以证明有效的用例。我编写了下面的代码作为消费者，以便它不做任何事情，但您可以将其应用于函数和其他任何可能遇到的内容。
class PredicateSplitterConsumer<T> implements Consumer<T>
{
private Predicate<T> predicate;
private Consumer<T> positiveConsumer;
private Consumer<T> negativeConsumer;
public PredicateSplitterConsumer(Predicate<T> predicate, Consumer<T> positive, Consumer<T> negative)
{
this.predicate = predicate;
this.positiveConsumer = positive;
this.negativeConsumer = negative;
}
@Override
public void accept(T t)
{
if (predicate.test(t))
{
positiveConsumer.accept(t);
}
else
{
negativeConsumer.accept(t);
}
}
}
personsArray.forEach(
new PredicateSplitterConsumer<>(
person -> person.getDateOfBirth().isPresent(),
person -> System.out.println(person.getName()),
person -> System.out.println(person.getName() + " does not have Date of birth")));
流应该只被操作一次（调用中间或终端的流操作）。这排除了例如“分叉”流的操作，其中相同的源提供了两个或更多的管道，或者对同一流进行多次遍历。
如果您真正想实现这种类型的行为，可以使用peek或其他方法来解决。在这种情况下，您应该复制您的流并适当地筛选每个副本，而不是尝试从同一原始流源备份两个流。
然而，您可能需要重新考虑是否使用Stream是您的最佳方案。
List<Stream> forkStream(Stream s)，但我的结果流至少部分由集合支持，而不是直接由底层流支持，与filter不同，后者不是终端流操作。 - Trevor FreemanRandom r = new Random();
PrimitiveIterator.OfInt coin = r.ints(0, 2).iterator();
List<Long> list = Stream.iterate(0, i -> coin.nextInt())
.limit(100).collect(teeing(
filtering(i -> i == 1, counting()),
filtering(i -> i == 0, counting()),
(heads, tails) -> {
return(List.of(heads, tails));
}));
System.err.println("heads:" + list.get(0) + " tails:" + list.get(1));
获取示例： 正面：51 反面：49
Stream，这是没有意义的——如果你需要在迭代一个流时同时生成另一个流，那么该怎么办呢？流只能被操作一次。stream.forEach((x) -> ((x == 0) ? heads : tails).add(x));
stream.collect(...)，它配备了预定义的线程安全的Collectors，即使在非线程安全的集合上也能很好地工作（没有同步锁争用）。@MarkJeronimus 给出了最佳答案。 - YoYocount()）都将“消耗”S0中的所有元素。因此，Sb失去了其数据源。tee()方法，它将流复制到两个流中。但现在已经被移除了。不完全准确，但您可以通过调用Collectors.groupingBy()来完成需要的操作。您可以创建一个新的集合，然后在该新集合上实例化流。
这是我能想到的最好的答案了。
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
public class Test {
public static <T, L, R> Pair<L, R> splitStream(Stream<T> inputStream, Predicate<T> predicate,
Function<Stream<T>, L> trueStreamProcessor, Function<Stream<T>, R> falseStreamProcessor) {
Map<Boolean, List<T>> partitioned = inputStream.collect(Collectors.partitioningBy(predicate));
L trueResult = trueStreamProcessor.apply(partitioned.get(Boolean.TRUE).stream());
R falseResult = falseStreamProcessor.apply(partitioned.get(Boolean.FALSE).stream());
return new ImmutablePair<L, R>(trueResult, falseResult);
}
public static void main(String[] args) {
Stream<Integer> stream = Stream.iterate(0, n -> n + 1).limit(10);
Pair<List<Integer>, String> results = splitStream(stream,
n -> n > 5,
s -> s.filter(n -> n % 2 == 0).collect(Collectors.toList()),
s -> s.map(n -> n.toString()).collect(Collectors.joining("|")));
System.out.println(results);
}
}
这个程序接收一串整数，并将它们在数字5处分割。对于大于5的数字，仅筛选偶数并将它们放入列表中。对于其余数字，使用 | 符号将它们连接起来。
输出结果：
([6, 8],0|1|2|3|4|5)
这并不是理想的做法，因为它会将所有内容收集到中间集合中，打破流程（而且参数太多了！）
我在寻找过滤流中特定元素并将其记录为错误的方法时，偶然发现了这个问题。因此，我并不需要太多地拆分流，只需要使用不显眼的语法将一个预期的终止操作附加到谓词上即可。以下是我想出来的代码：
public class MyProcess {
/* Return a Predicate that performs a bail-out action on non-matching items. */
private static <T> Predicate<T> withAltAction(Predicate<T> pred, Consumer<T> altAction) {
return x -> {
if (pred.test(x)) {
return true;
}
altAction.accept(x);
return false;
};
/* Example usage in non-trivial pipeline */
public void processItems(Stream<Item> stream) {
stream.filter(Objects::nonNull)
.peek(this::logItem)
.map(Item::getSubItems)
.filter(withAltAction(SubItem::isValid,
i -> logError(i, "Invalid")))
.peek(this::logSubItem)
.filter(withAltAction(i -> i.size() > 10,
i -> logError(i, "Too large")))
.map(SubItem::toDisplayItem)
.forEach(this::display);
}
}
使用 Lombok 的简短版本
import java.util.function.Consumer;
import java.util.function.Predicate;
import lombok.RequiredArgsConstructor;
/**
* Forks a Stream using a Predicate into postive and negative outcomes.
*/
@RequiredArgsConstructor
@FieldDefaults(makeFinal = true, level = AccessLevel.PROTECTED)
public class StreamForkerUtil<T> implements Consumer<T> {
Predicate<T> predicate;
Consumer<T> positiveConsumer;
Consumer<T> negativeConsumer;
@Override
public void accept(T t) {
(predicate.test(t) ? positiveConsumer : negativeConsumer).accept(t);
}
}
Stream转换为多个Stream而不进行中间转换的可能性，尽管我认为达到这个问题的人实际上正在寻找实现此目标的方法，而不考虑这种限制，这是Mark的答案。这可能是因为标题中的问题与描述中的问题不同。 - devildelta