从一个长的流创建流

Question

我想根据Streams的内容将单个Stream拆分为Streams的Stream。生成的Stream应该包含原始流数据的一部分。

我的实际应用程序更复杂（它是对时间间隔列表中的日志行进行分组），但我的问题是如何处理流，因此这里我询问一个简化的示例。

我希望能够拆分流

例如，以下流包含：

{1,1,1,2,2,3,6,7,7,1,1}

需要产生包含以下内容的流：

{{1,1,1}，{3}，{7,7}，{1,1}}

省略偶数，我可以用过滤器开始（或结束）：

Stream<Integer> input = ...;
Straem<Stream<Integer>> output = input.filter(this::isOdd).someOtherOperation();

这是不希望的，因为这将意味着对每个输入值进行两次评估，这是可以接受的，但我更希望避免这种情况。

我当前的解决方案就是迭代流的内容并创建一个列表

我还认为我可以通过编写自己的从流中读取的迭代器来实现这一点，但我不确定这将如何工作。

如何基于原始流的内容将流转换为流的流，而不首先将完整结果存储为列表的列表。




Answer 1

恐怕这是不可行的，至少不是以一种好的方式。即使您将元素映射到流中并减少它们，这些内部流也必须知道它们包含哪些元素，以便它们必须存储一些内容。

最简单的解决方案是使用group pingBy，但是它会将所有结果存储在地图中：

List<Integer> input = asList(1, 1, 1, 2, 2, 2, 3, 6, 7, 7, 1, 1);
Map<Integer, List<Integer>> grouped = input.stream().collect(groupingBy(i -> i));
Stream<Stream<Integer>> streamOfStreams = grouped.values().stream().map(list -> list.stream());

您可以尝试使用reduce操作，但它需要您实现自己的流，您必须在其中存储每个流包含的元素。更不用说实施它需要付出很大的努力。

对于您的情况，我能想到的最佳解决方案是在列表上迭代两次：

public static void main(String[] args) {
    List<Integer> input = asList(1, 1, 1, 2, 2, 2, 3, 6, 7, 7, 1, 1);

    input.stream().distinct().filter(i -> isOdd(i)).forEach(i -> {
        List<Integer> subList = input.stream().filter(j -> Objects.equals(j, i)).collect(toList());
        System.out.println(subList); // do something with the stream instead of collecting to list
    });
}

private static boolean isOdd(Integer i) {
    return (i & 1) == 1;
}

但是请注意，它具有时间复杂度O（n^2）。

编辑：

此解决方案将只包含本地元素组。它只存储当前本地组。

public static void main(String[] args) {
    Stream<Integer> input = Stream.of(1, 1, 1, 2, 2, 2, 3, 6, 7, 7, 1, 1);

    Iterator<Integer> iterator = input.iterator();
    int first;
    int second = iterator.next();

    List<Integer> buffer = new ArrayList<>();
    buffer.add(second);

    do {
        first = second;
        second = iterator.next();

        if (Objects.equals(first, second)) {
            buffer.add(second);
        } else {
            doSomethingWithTheGroup(buffer);
            buffer = new ArrayList<>(); // let GC remove the previous buffer
            buffer.add(second);
        }
    } while (iterator.hasNext());
    doSomethingWithTheGroup(buffer);
}

private static void doSomethingWithTheGroup(List<Integer> buffer) {
    System.out.println(buffer);
}

private static boolean isOdd(Integer i) {
    return (i & 1) == 1;
}

输出：

[1, 1, 1]
[2, 2, 2]
[3]
[6]
[7, 7]
[1, 1]

Answer 2

匿名用户

您可以使用我的StreamEx库。它具有执行作业的groupRuns：

List<Integer> input = Arrays.asList(1, 1, 1, 2, 2, 2, 3, 6, 7, 7, 1, 1);
Stream<Stream<Integer>> streams = StreamEx.of(input).filter(this::isOdd)
    .groupRuns(Integer::equals)
    .map(List::stream);

使用示例：

streams.map(s -> StreamEx.of(s).joining(",")).forEach(System.out::println);

输出：

1,1,1
3
7,7
1,1

与protonpack库类似，内部有一个自定义拆分器，但使用StreamEx可以利用并行处理（protonpack根本不拆分）。

在顺序处理中，一次最多有一个中间列表驻留在内存中（其他列表符合GC条件）。如果您仍然担心内存消耗（例如，您有很长的组），则有另一种解决此任务的方法，因为StreamEx 0.3.3：

Stream<Stream<Integer>> streams = StreamEx.of(input).filter(this::isOdd)
        .runLengths()
        .mapKeyValue(StreamEx::constant);

runLengths方法返回条目流，其中key是元素，value是相邻重复元素的数量。之后，使用StreamEx.constant，这是Stream.generate（（）-

更新：StreamEx 0.3.3已发布，因此第二个解决方案现在也符合条件。

Answer 3

为此，您可能需要实现自己的聚合拆分器。质子包库中已经有了类似的东西（第一个链接重定向到质子包中实现的链接）。

注意，你得到一个Stream

StreamUtils.aggregate(Stream.of(1, 1, 1, 2, 2, 2, 3, 6, 7, 7, 1, 1), 
                      Objects::equals)
           .forEach(System.out::println);

产出：

[1, 1, 1]
[2, 2, 2]
[3]
[6]
[7, 7]
[1, 1]