从流和元素生成流 Java 8

Question

我正在研究一些Java8流特性。我对FP相当熟悉，三十年前写过一些Lisp，我想我可能会尝试做一些新工具不真正针对的事情。不管怎样，如果这个问题很愚蠢，我很高兴知道我的错误。

我将给出一个具体的问题，尽管它实际上是我试图解决的一般概念。

假设我想从流的每三个元素中获取一个流。在常规FP中，我将（近似地）创建一个递归函数，该函数通过在删除两个元素之后将列表的第一个元素与列表的其余元素（调用thyself）连接起来进行操作。这很容易。但要在流中实现这一点，我觉得我需要两种工具之一：

1） 一种让操作从流中提取多个项目进行处理的方法（然后我只需要抓取三个项目，使用第一个项目，然后转储其余项目）

2)一种制造供应商的方法，它接受一个项目和一个流，并创建一个流。然后，我感觉我可以从第一个项目和缩短的流中创建一个下游流，尽管我仍然不清楚这是否会实现必要的递归魔法。

开始编辑

因此，有一些有趣且有用的反馈;谢谢大家。特别是，这些评论帮助我澄清了我的头脑试图更好地解决什么问题。

首先，一个人可以——至少在概念上——拥有/需要序列顺序的知识不应妨碍他允许完全并行的操作。我想到了一个例子，那就是图形绘制人员倾向于做的卷积运算。想象一下模糊图像。每个像素都通过其附近的像素进行修改，但这些像素本身仅被读取，而不是修改。

这是我的理解（当然，在这个阶段非常不稳定！）流机制是虚拟机管理并行的奇妙世界的主要入口点，迭代器仍然是他们一直以来的样子（是吗？否？）如果这是正确的，那么使用迭代器来解决我所讨论的问题域似乎并不太好。

因此，至少在这一点上，创建分块拆分器的建议似乎是最有前途的，但是，天哪，支持该html" target="_blank">示例的代码看起来像是一项艰巨的工作吗！我想我更愿意使用ForkJoin机制，尽管它现在是“旧帽子”：

无论如何，我仍然对人们希望提供的更多见解感兴趣。

结束编辑

有什么想法吗？我是在尝试使用这些流做一些它们原本不打算做的事情，还是我遗漏了一些明显的东西？

干杯，托比。

Answer 1

Java的流与FP的（懒惰）序列完全不同。如果你熟悉Clojure，那么区别就像懒惰seq和reducer之间的区别一样。虽然惰性 seq 单独打包每个元素的所有处理，从而允许获取单独处理的元素，但化简器在单个原子操作中折叠完整序列。

特别是对于您描述的示例，请考虑依赖流分区转换，如此处所述。然后你会很容易做到

partition(originalStream, 3).map(xs -> xs.get(0));

导致流具有原始元素的三分之一。

这将保持效率、惰性和并行性。

Answer 2

My StreamEx 库增强了标准 Stream API。特别是，它添加了headTail方法，该方法允许递归定义自定义操作。它采用一个接收流头（第一个元素）和尾部（其余元素的流）的函数，并应返回将使用的结果流而不是原始流。例如，可以按如下方式定义 every3 操作：

public static <T> StreamEx<T> every3(StreamEx<T> input) {
    return input.headTail(
        (first, tail1) -> tail1.<T>headTail(
            (second, tail2) -> tail2.headTail(
                (third, tail3) -> every3(tail3))).prepend(first));
}

这里还使用了＜code＞prepend

一般来说，使用HeadTail您可以定义几乎任何您想要的中间操作，包括现有操作和新操作。您可以在此处找到一些示例。

请注意，我实现了一些机制，可以在这种递归操作定义中优化尾部，因此在处理长流时，正确定义的操作不会占用整个堆栈。

Answer 3

要记住的一件事是，< code>Stream最初是被设计成一种利用并行处理的方式。这意味着它们有许多相关联的条件，这些条件旨在给予VM很大的自由度来以任何方便的顺序处理元素。这方面的一个例子是坚持归约函数是关联的。另一个是被操纵的局部变量是最终变量。这些类型的条件意味着可以以任何顺序评估和收集流项目。

这样做的自然结果是，流的最佳用例不涉及流值之间的依赖关系。在LISP这样的语言中，将一个整数流映射到它们的累积值这样的事情是微不足道的，但对于Java流来说却是非常不自然的（参见这个问题）。

通过使用< code>sequential强制< code>Stream不要并行，有一些巧妙的方法可以绕过这些限制，但是我的经验是这些方法比它们的价值更麻烦。如果您的问题涉及本质上连续的一系列项目，其中需要状态来处理值，那么我推荐使用传统的集合和迭代。如果流不能被并行化，代码将会变得更加清晰，执行起来也不会更差。

话虽如此，如果你真的想这样做，那么最直接的方法是让一个收集器存储每三个项目，然后再次将它们作为流发送出去：

class EveryThird {

    private final List<Integer> list = new ArrayList<>();
    private int count = 0;

    public void accept(Integer i) {
        if (count++ % 3 == 0)
            list.add(i);
    }

    public EveryThird combine(EveryThird other) {
        list.addAll(other.list);
        count += other.count;
        return this;
    }

    public Stream<Integer> stream() {
        return list.stream();
    }
}

然后可以像这样使用：

IntStream.range(0, 10000)
    .collect(EveryThird::new, EveryThird::accept, EveryThird::combine)
    .stream()

但这并不是收集器设计的真正目的，这是非常低效的，因为它不必要地收集流。如上所述，我的建议是在这种情况下使用传统迭代。