将 Java8 流数组转换为元组流

Question

假设我有一个 Java 8 流数组：Stream<T>[] streams，我想创建一个 Stream，其中新流的每个元素都是一个数组，该数组由每个初始基本流中的一个元素组成（让我们假设它们都是连续的）。

例如，如果我有：

  streams [ 0 ] returning: ( "A", "B", "C" ), 
  streams [ 1 ] returning ( "X", "Y", "Z" ) 
  and streams [ 2 ] as ( "0", "1", "2" )

我想要一个返回的流

  ( { "A", "X", "0" }, { "B", "Y", "1" }, { "C", "Z", "2" } )

是否有一些代码已经实现了这一点？我知道如何做到这一点，这将是pair case 的概括，但我想知道是否已经存在可重用的东西。

编辑：抱歉，我意识到我需要澄清一下：

• 我不想创建整个矩阵，我想要一个动态地一次返回一行的流（第一个 A/X/0，然后是 B/Y/1 等），而不必预先将所有行占用内存。我可以对基本流的大小做出合理的假设（例如，取最小值，一旦有没有更多元素要返回的流就停止）。

• 我知道这可以通过首先将基本流转换为迭代器来实现，然后创建一个新迭代器，next() 从每个下划线迭代器中选择一个元素并返回一个新行。这就是我上面链接的对示例所做的，我可以在自己身上实现它，在这里我试图了解它是否已经在某个库中完成（我知道 JDK 没有这样的功能）。

Answer 1

首先，保留一组流是一个非常糟糕的主意，因为它们不能被重用，而且会使已经很复杂的可能解决方案变得复杂。

不，这在普通的 JDK 中是不可能的。没有zip 功能，我们也没有Tuples，所以恐怕这是你能想到的最好的东西：

Stream[] streams = Stream.of(
  Stream.of("A", "B", "C"),
  Stream.of("X", "Y", "Z"),
  Stream.of("0", "1", "2"))
    .toArray(Stream[]::new);

String[][] arrays = Arrays.stream(streams)
  .map(s -> s.toArray(String[]::new))
  .toArray(String[][]::new);

int minSize = Arrays.stream(arrays)
  .mapToInt(s -> s.length)
  .min().orElse(0);

String[][] zipped = IntStream.range(0, minSize)
  .mapToObj(i -> Arrays.stream(arrays)
  .map(s -> s[i])
    .toArray(String[]::new))
  .toArray(String[][]::new);

首先，我们需要将一个流数组转换为一个数组数组或任何我们可以多次遍历的东西。

其次，如果数组中的流具有不同的长度，您没有指定该怎么做，我假设标准的 zip 行为只要我们可以从每个集合中提取元素，就可以连接元素。

第三，我在这里创建了所有可能的索引流，用于压缩 (IntStream.range(0, minSize)) 并从每个嵌套数组中逐个元素地手动提取。

这里可以在 Optional 上使用 .get() ，因为计算 minSize 可以保证里面会有东西。

假设我们正在处理列表列表，这是一种更合理的方法：

List<List<String>> lists = Arrays.asList(
  Arrays.asList("A", "B", "C"),
  Arrays.asList("X", "Y", "Z"),
  Arrays.asList("0", "1", "2"));

final int minSize = lists.stream()
  .mapToInt(List::size)
  .min().orElse(0);

List<List<String>> result = IntStream.range(0, minSize)
  .mapToObj(i -> lists.stream()
  .map(s -> s.get(i))
    .collect(Collectors.toList()))
  .collect(Collectors.toList());

Java 9 添加的 Stream API 可能会让我们放弃计算 minSize。

如果你希望序列的生成保持lazy，你可以不收集结果：

IntStream.range(0, minSize)
  .mapToObj(i -> lists.stream()
    .map(s -> s.get(i))
    .collect(Collectors.toList()));

Answer 2

如果你真的是指任意数量的 Streams 作为输入 - 我能想到的不是 TupleX，但如果你真的知道传入的流都是相同的大小（没有无限流），那么这可能会满足您的需求：

@SafeVarargs
static <T> Stream<Stream<T>> streamOfStreams(Stream<T>... streams) {

    @SuppressWarnings("unchecked")
    Iterator<T>[] iterators = new Iterator[streams.length];
    for (int i = 0; i < streams.length; ++i) {
        iterators[i] = streams[i].iterator();
    }

    Iterator<T> first = iterators[0];

    Builder<Stream<T>> outer = Stream.builder();
    Builder<T> inner = Stream.builder();
    while (first.hasNext()) {
        for (int i = 0; i < streams.length; ++i) {
            inner.add(iterators[i].next());
        }
        outer.add(inner.build());
        inner = Stream.builder();
    }

    return outer.build();
}

Answer 3

从 Guava 版本 21 开始，您可以使用 Streams.zip 实用方法，它可以满足您的需求，只是它只适用于两个流。

现在，如果您将流数组转换为流流，则可以使用此Streams.zip 方法执行归约：

Stream<List<String>> zipped = Arrays.stream(streams)
    .map(s -> s.map(e -> {
        List<String> l = new ArrayList<>();
        l.add(e);
        return l;
    }))
    .reduce((s1, s2) -> Streams.zip(s1, s2, (l1, l2) -> {
        l1.addAll(l2);
        return l1;
    }))
    .orElse(Stream.empty());

List<List<String>> tuples = zipped.collect(Collectors.toList());

System.out.println(tuples); // [[A, X, 0], [B, Y, 1], [C, Z, 2]]

请注意，在减少之前，您需要将每个Stream<T> 映射到Stream<List<T>>，以便您可以使用List.addAll 压缩流。

---

编辑：上面的代码有效，但我对它的性能和内存占用非常担心，这主要是由于创建了单个元素的多个列表。

也许使用接受身份、累加器和组合器的Stream.reduce 版本效果更好：

Stream<List<String>> zipped = Arrays.stream(streams)
    .reduce(
        IntStream.range(0, streams.length).mapToObj(n -> new ArrayList<>()),
        (z, s) -> Streams.zip(z, s, (l, e) -> {
            l.add(e);
            return l;
        }),
        (s1, s2) -> Streams.zip(s1, s2, (l1, l2) -> {
            l1.addAll(l2);
            return l1;
        }));

List<List<String>> tuples = zipped.collect(Collectors.toList());

System.out.println(tuples); // [[A, X, 0], [B, Y, 1], [C, Z, 2]]

标识需要是n 空列表流，n 是streams 数组的长度，而累加器使用Streams.zip 压缩列表流和元素流。组合器和以前一样：它使用Streams.zip 压缩两个列表流。

Answer 4

好的，好像没有这样的东西，所以我自己写了：

• TupleSpliterator，从拆分器数组开始构建元组拆分器；
• Tuple Stream Builder，它构建一个元组流，从一个流数组开始并利用一个元组迭代器。
• 基于 Spliteraror/Iterator 允许并行（在某些条件下），如果您想要更简单但顺序的东西，也可以使用 TupleIterator。

单元测试中可用的用法示例（here 和 here），这些类是 utility package 的一部分。

编辑：我在 Federico 发表评论后添加了 Spliterator 实现，注意到基于 Iterator 的版本不能并行。