为什么原始流（primitive Stream）没有collect(Collector)方法？

Question

我正在为初学者编写一个库，因此我试图尽可能地保持API的简洁性。

我的库需要做的事情之一是对大量的int或long类型的集合执行一些复杂的计算。有许多场景和业务对象需要用户从中计算这些值，因此我认为最好的方式是使用流来允许用户将业务对象映射到IntStream或LongStream，然后在收集器内部计算这些计算结果。

但是，IntStream和LongStream只有三个参数的collect方法：

collect(Supplier<R> supplier, ObjIntConsumer<R> accumulator, BiConsumer<R,R> combiner)

而且它没有像Stream<T>那样简单的collect(Collector)方法。

因此，无法执行以下操作：

Collection<T> businessObjs = ...
MyResult result = businessObjs.stream()
                              .mapToInt( ... )
                              .collect( new MyComplexComputation(...));

我需要提供类似这样的供应商、蓄电池和组合器：

MyResult result = businessObjs.stream()
                              .mapToInt( ... )
                              .collect( 
                                  ()-> new MyComplexComputationBuilder(...),
                                  (builder, v)-> builder.add(v),
                                  (a,b)-> a.merge(b))
                              .build(); //prev collect returns Builder object

这对我那些新手用户来说太过复杂了，而且很容易出错。

我的解决方法是创建静态方法，接受IntStream或LongStream作为输入，并为您隐藏集合器的创建和执行。

public static MyResult compute(IntStream stream, ...){
       return .collect( 
                        ()-> new MyComplexComputationBuilder(...),
                        (builder, v)-> builder.add(v),
                        (a,b)-> a.merge(b))
               .build();
}

但这不遵循使用流的常规约定：

IntStream tmpStream = businessObjs.stream()
                              .mapToInt( ... );

 MyResult result = MyUtil.compute(tmpStream, ...);

因为你必须保存一个临时变量并将其传递给静态方法，或者在静态调用中创建流，这可能会与我的计算中的其他参数混淆。是否有更简洁的方法，在仍然使用IntStream或LongStream的情况下实现这一点？

- dkatzel

---

不幸的是，我的建议是使用 Stream<Integer>。你可以通过 mapToObj(Function.identity()) 从 IntStream 中获取它。 - Dmitry Ginzburg

编译器可能会消除装箱/拆箱，如果它可以将转换路径从转换到您的消费者内联。只需像使用IntStream一样编写基于int的接口，然后查看它是否会生成任何垃圾。 - the8472

@DmitryGinzburg，IntStream#boxed() 提供了相同的功能。 - the8472

我们也不应该过于信任编译器优化。 - ZhongYu

对你来说很好。对我来说，这主要是基于信仰的 - 我相信我可以这样写，因为编译器可以优化它；我相信不应该那样做，因为编译器不太可能优化它。- 我太懒了，无法测试每种实现策略的排列组合。 - ZhongYu

Answer 1

事实上，我们确实原型化了一些Collector.OfXxx专业化。除了更多的专业类型带来的明显烦恼之外，我们发现如果没有一整套基于原始类型专业化的集合（就像Trove或GS-Collections，但JDK没有），这并不是非常有用的。例如，如果没有IntArrayList，那么Collector.OfInt仅将装箱推到另一个地方——从收集器推到容器，这样做既没有很大的优势，还增加了许多API接口。

- Brian Goetz

---

所以您认为没有性能差异吗？如果我使用IntStream，我不知道会有任何装箱。 - dkatzel

你正在关注错误的流端。当操作int时，IntStream不进行装箱。但是，你能将int放入哪些结果容器中而不进行装箱呢？不包括ArrayList<Integer>、HashSet<Integer, Integer>等。如果没有丰富的适用于int的收集器，Collector.OfInt就没有用处。 - Brian Goetz

@dkatzel 对的，IntStream及其相关类的整个设计就是为了避免在每个操作中都进行装箱(boxing)，而您已经理解了这一点。如果我们有原始(primitive)友好的集合类，那么Collector.OfXxx可能会更有意义，但我们没有这些，所以我们只专注于实现那些真正有实际价值的专门化(specialization)。 - Brian Goetz

@bayou.io 你不必被说服。问题是“为什么他们没有这样做”，答案是“我们制作了原型，并得出结论，花费、成本、复杂性与回报之间的平衡不成立”。 - Brian Goetz

@davidalayachew 是的，但需要经过几个步骤。专门的泛型将在后续步骤中出现，所以你的答案最终是正确的，但不是立即有效的。（一旦我们拥有专门的泛型，还需要额外的努力来推动像IntStream这样的手动特化逐渐退出。）这一切需要一段时间才能完全实现。 - undefined

Answer 2

也许，如果使用方法引用而不是lambda表达式，原始流收集所需的代码就不会显得那么复杂。

MyResult result = businessObjs.stream()
                              .mapToInt( ... )
                              .collect( 
                                  MyComplexComputationBuilder::new,
                                  MyComplexComputationBuilder::add,
                                  MyComplexComputationBuilder::merge)
                              .build(); //prev collect returns Builder object

在Brian对于这个问题的权威回答中，他提到了另外两个Java集合框架，它们具有实际可用于原始流收集方法的原始集合。我觉得举一些如何使用这些框架中的原始容器与原始流的例子可能会很有用。下面的代码也适用于并行流。

// Eclipse Collections
List<Integer> integers = Interval.oneTo(5).toList();

Assert.assertEquals(
        IntInterval.oneTo(5),
        integers.stream()
                .mapToInt(Integer::intValue)
                .collect(IntArrayList::new, IntArrayList::add, IntArrayList::addAll));

// Trove Collections

Assert.assertEquals(
        new TIntArrayList(IntStream.range(1, 6).toArray()),
        integers.stream()
                .mapToInt(Integer::intValue)
                .collect(TIntArrayList::new, TIntArrayList::add, TIntArrayList::addAll));

注意：我是Eclipse Collections的贡献者。

- Donald Raab

Answer 3

我已在我的库StreamEx中实现了基本收集器（自版本0.3.0起）。有接口IntCollector, LongCollector和DoubleCollector，它们扩展了Collector接口并专门用于处理原始类型。在组合过程中还有一个额外的小差异，例如像IntStream.collect这样的方法接受一个BiConsumer而不是BinaryOperator。请保留HTML标签，不做解释。

有一些预定义的集合方法，可以将数字与字符串连接、存储为基本数组、转换为 BitSet、查找最小值、最大值、总和、计算摘要统计信息、执行按组分组和分区操作。当然，您也可以定义自己的收集器。以下是几个使用示例（假定您有包含输入数据的 int[] input 数组）。

使用分隔符将数字连接为字符串：

String nums = IntStreamEx.of(input).collect(IntCollector.joining(","));

按最后一位数字分组：

Map<Integer, int[]> groups = IntStreamEx.of(input)
      .collect(IntCollector.groupingBy(i -> i % 10));

分别求正数和负数的和：

Map<Boolean, Integer> sums = IntStreamEx.of(input)
      .collect(IntCollector.partitioningBy(i -> i > 0, IntCollector.summing()));

这里有一个简单的基准测试，用于比较这些收集器和通常的对象收集器。

请注意，我的库不提供（也不会在未来提供）任何用户可见的原始映射等数据结构，因此分组是在常规HashMap中执行的。但是，如果您使用Trove/GS/HFTC/等库，则可以编写额外的原始收集器，以获得更高的性能。

- Tagir Valeev

Answer 4

如果您缺少某些方法，请将原始流转换为装箱对象流。

MyResult result = businessObjs.stream()
                          .mapToInt( ... )
                          .boxed()
                          .collect( new MyComplexComputation(...));

或者一开始就不使用原始流，一直使用 Integer。

MyResult result = businessObjs.stream()
                          .map( ... )     // map to Integer not int
                          .collect( new MyComplexComputation(...));

- John Kugelman

---

有一个特定的原因使用原始流。再次装箱或使用装箱的原始类型会打破这个目的。否则，还有另一种使用for循环的替代方案。 - SijuMathew

Answer 5

Geotz先生提供了为什么决定不包括专门的收集器的明确答案，但我想进一步调查这个决定对性能的影响。

我想把我的结果发布为一个答案。

我使用jmh微基准测试框架来计算使用两种类型的收集器计算不同大小的集合（1、100、1000、100000和100万）所需的时间：

@BenchmarkMode(Mode.AverageTime)
@OutputTimeUnit(TimeUnit.NANOSECONDS)
@State(Scope.Thread)
public class MyBenchmark {

@Param({"1", "100", "1000", "100000", "1000000"})
public int size;

List<BusinessObj> seqs;

@Setup
public void setup(){
    seqs = new ArrayList<BusinessObj>(size);
    Random rand = new Random();
    for(int i=0; i< size; i++){
        //these lengths are random but over 128 so no caching of Longs
        seqs.add(BusinessObjFactory.createOfRandomLength());
    }
}
@Benchmark
public double objectCollector() {       

    return seqs.stream()
                .map(BusinessObj::getLength)
                .collect(MyUtil.myCalcLongCollector())
                .getAsDouble();
}

@Benchmark
public double primitiveCollector() {

    LongStream stream= seqs.stream()
                                    .mapToLong(BusinessObj::getLength);
    return MyUtil.myCalc(stream)        
                        .getAsDouble();
}

public static void main(String[] args) throws RunnerException{
    Options opt = new OptionsBuilder()
                        .include(MyBenchmark.class.getSimpleName())
                        .build();

    new Runner(opt).run();
}

}

以下是结果：

# JMH 1.9.3 (released 4 days ago)
# VM invoker: /Library/Java/JavaVirtualMachines/jdk1.8.0_31.jdk/Contents/Home/jre/bin/java
# VM options: <none>
# Warmup: 20 iterations, 1 s each
# Measurement: 20 iterations, 1 s each
# Timeout: 10 min per iteration
# Threads: 1 thread, will synchronize iterations
# Benchmark mode: Average time, time/op
# Benchmark: org.sample.MyBenchmark.objectCollector

# Run complete. Total time: 01:30:31

Benchmark                        (size)  Mode  Cnt          Score         Error  Units
MyBenchmark.objectCollector           1  avgt  200        140.803 ±       1.425  ns/op
MyBenchmark.objectCollector         100  avgt  200       5775.294 ±      67.871  ns/op
MyBenchmark.objectCollector        1000  avgt  200      70440.488 ±    1023.177  ns/op
MyBenchmark.objectCollector      100000  avgt  200   10292595.233 ±  101036.563  ns/op
MyBenchmark.objectCollector     1000000  avgt  200  100147057.376 ±  979662.707  ns/op
MyBenchmark.primitiveCollector        1  avgt  200        140.971 ±       1.382  ns/op
MyBenchmark.primitiveCollector      100  avgt  200       4654.527 ±      87.101  ns/op
MyBenchmark.primitiveCollector     1000  avgt  200      60929.398 ±    1127.517  ns/op
MyBenchmark.primitiveCollector   100000  avgt  200    9784655.013 ±  113339.448  ns/op
MyBenchmark.primitiveCollector  1000000  avgt  200   94822089.334 ± 1031475.051  ns/op

正如您所看到的，原始的Stream版本略微更快，但即使集合中有100万个元素，平均只快了0.05秒。

对于我的API，我宁愿遵循更清晰的Object Stream约定并使用Boxed版本，因为这只是一个很小的性能损失。

感谢所有为解决此问题提供见解的人。

- dkatzel

---

这取决于你要解决的任务。在你的情况下，你的计算可能相当复杂，所以装箱开销不重要。我几乎在我的库中实现了原始收集器，并且性能提升可以相当显着，从“按最后一位数字分组”的任务提高了30％，到字符串连接提高了2倍，到“按符号求和”的任务提高了5倍。查看我的基准测试和结果 这里。 - Tagir Valeev

@TagirValeev 是的，你可能是正确的。大部分时间都花在我的计算中，其中已经取消了所有装箱操作。 - dkatzel

尝试测量 IntStream.sum() 和 Stream<Integer>.reduce(0, Integer::sum, Integer::sum) 之间在速度和可扩展性方面的差异。你会发现在这些情况下，无论是速度还是并行加速效果都巨大，这就是为什么针对基本操作进行原始类型优化是有道理的。但随着操作变得更加复杂，好处会降低，装箱变得更加可接受。 - Brian Goetz