Java 8 流上的缓冲区运算符

Question

我正在尝试编写一个 java 8 流收集器，它反映了rxjava 缓冲区运算符的功能

我有一个工作代码：

// This will gather numbers 1 to 13 and combine them in groups of
// three while preserving the order even if its a parallel stream.
final List<List<String>> triads = IntStream.range(1, 14)
        .parallel()
        .boxed()
        .map(Object::toString)
        .collect(ArrayList::new, accumulator, combiner);

System.out.println(triads.toString())

这里的累加器是这样的：

final BiConsumer<List<List<String>>, String> accumulator = (acc, a) -> {
      StringBuilder stringBuilder = new StringBuilder();
      stringBuilder.append("Accumulator|");
      stringBuilder.append("Before: ").append(acc.toString());
      int accumulatorSize = acc.size();
      if (accumulatorSize == 0) {
        List<String> newList = new ArrayList<>();
        newList.add(a);
        acc.add(newList);
      } else {
        List<String> lastList = acc.get(accumulatorSize - 1);
        if (lastList.size() != 3) {
          lastList.add(a);
        } else {
          List<String> newList = new ArrayList<>();
          newList.add(a);
          acc.add(newList);
        }
      }
      stringBuilder.append("|After: ").append(acc.toString());
      stringBuilder.append("|a: ").append(a);
      System.out.println(stringBuilder.toString());
    };

还有组合器

// Utility method to make first list of size 3 
// by shifting elements from second to first list
final BiConsumer<List<String>, List<String>> fixSize = (l1, l2) -> {
  while(l1.size() != 3 && l2.size() > 0) {
    l1.add(l2.remove(0));
  }
};

final BiConsumer<List<List<String>>, List<List<String>>> combiner = (l1, l2) -> {
  StringBuilder stringBuilder = new StringBuilder();
  stringBuilder.append("Combiner|");
  stringBuilder.append("Before, l1: ").append(l1).append(", l2: ").append(l2);
  if (l1.isEmpty()) {
    // l1 is empty
    l1.addAll(l2);
  } else {
    // l1 is not empty
    List<String> lastL1List = l1.get(l1.size() - 1);
    if (lastL1List.size() == 3) {
      l1.addAll(l2);
    } else {
      if (l2.isEmpty()) {
        // do nothing
      } else {
        List<List<String>> fixSizeList = new ArrayList<>(1 + l2.size());
        fixSizeList.add(lastL1List);
        fixSizeList.addAll(l2);
        for (int i = 0; i < fixSizeList.size() - 1; i++) {
          List<String> x = fixSizeList.get(i), y = fixSizeList.get(i + 1);
          fixSize.accept(x, y);
        }
        l2.stream().filter(l -> !l.isEmpty()).forEach(l1::add);
        // everything is now of size three except, may be last
      }
    }
  }
  stringBuilder.append("|After, l1: ").append(l1).append(", l2: ").append(l2);
  System.out.println(stringBuilder.toString());
};

这会产生以下输出：

Accumulator|Before: []|After: [[12]]|a: 12
Accumulator|Before: []|After: [[2]]|a: 2
Accumulator|Before: []|After: [[11]]|a: 11
Accumulator|Before: []|After: [[6]]|a: 6
Accumulator|Before: []|After: [[4]]|a: 4
Accumulator|Before: []|After: [[1]]|a: 1
Accumulator|Before: []|After: [[13]]|a: 13
Accumulator|Before: []|After: [[8]]|a: 8
Accumulator|Before: []|After: [[3]]|a: 3
Accumulator|Before: []|After: [[5]]|a: 5
Accumulator|Before: []|After: [[10]]|a: 10
Accumulator|Before: []|After: [[7]]|a: 7
Accumulator|Before: []|After: [[9]]|a: 9
Combiner|Before, l1: [[5]], l2: [[6]]|After, l1: [[5, 6]], l2: [[]]
Combiner|Before, l1: [[12]], l2: [[13]]|After, l1: [[12, 13]], l2: [[]]
Combiner|Before, l1: [[2]], l2: [[3]]|After, l1: [[2, 3]], l2: [[]]
Combiner|Before, l1: [[8]], l2: [[9]]|After, l1: [[8, 9]], l2: [[]]
Combiner|Before, l1: [[10]], l2: [[11]]|After, l1: [[10, 11]], l2: [[]]
Combiner|Before, l1: [[4]], l2: [[5, 6]]|After, l1: [[4, 5, 6]], l2: [[]]
Combiner|Before, l1: [[1]], l2: [[2, 3]]|After, l1: [[1, 2, 3]], l2: [[]]
Combiner|Before, l1: [[7]], l2: [[8, 9]]|After, l1: [[7, 8, 9]], l2: [[]]
Combiner|Before, l1: [[10, 11]], l2: [[12, 13]]|After, l1: [[10, 11, 12], [13]], l2: [[13]]
Combiner|Before, l1: [[1, 2, 3]], l2: [[4, 5, 6]]|After, l1: [[1, 2, 3], [4, 5, 6]], l2: [[4, 5, 6]]
Combiner|Before, l1: [[7, 8, 9]], l2: [[10, 11, 12], [13]]|After, l1: [[7, 8, 9], [10, 11, 12], [13]], l2: [[10, 11, 12], [13]]
Combiner|Before, l1: [[1, 2, 3], [4, 5, 6]], l2: [[7, 8, 9], [10, 11, 12], [13]]|After, l1: [[1, 2, 3], [4, 5, 6], [7, 8, 9], [10, 11, 12], [13]], l2: [[7, 8, 9], [10, 11, 12], [13]]
[[1, 2, 3], [4, 5, 6], [7, 8, 9], [10, 11, 12], [13]]

我可能已经完全放弃了流的概念，但是有没有办法简化、优化或重写它？

为了简单起见，这是完整的程序（遗憾的是，stackoverflow 不允许在没有足够描述的情况下按原样发布代码）

http://rextester.com/TCMG7963

Answer 1

这是我想出的¹：

\n

public static <T> Stream<List<T>> buffer(Stream<T> stream, final long count) {\n    final Iterator<T> streamIterator = stream.iterator();\n\n    return StreamSupport.stream(Spliterators.spliteratorUnknownSize(new Iterator<List<T>>() {\n        @Override\n        public boolean hasNext() {\n            return streamIterator.hasNext();\n        }\n\n        @Override\n        public List<T>next() {\n            if (!hasNext()) {\n                throw new NoSuchElementException();\n            }\n            List<T> intermediate = new ArrayList<>();\n            for (long v = 0; v < count && hasNext(); v++) {\n                intermediate.add(streamIterator.next());\n            }\n            return intermediate;\n        }\n    }, 0), false);\n}\n

\n

显然，您传递给该函数的流以后无法修改。

\n

通过测试来展示其功能：

\n

public class Test {\n    public static void main(String[] args) {\n        List<List<Integer>> triads = buffer(IntStream.range(1, 14)\n                                                    .boxed()\n                                                    .parallel(), 3).collect(Collectors.toList());\n\n        System.out.println(triads);\n        System.out.println("Empty stream test");\n        System.out.println(buffer(Stream.<Integer>empty(), 4).collect(Collectors.toList()));\n\n        System.out.println("Intermediate size greater than stream size");\n        System.out.println(buffer(IntStream.range(1, 14)\n                                          .boxed()\n                                          .parallel(), 15).collect(Collectors.toList()));\n\n        System.out.println("Intermediate size same as stream size");\n        System.out.println(buffer(IntStream.range(1, 14)\n                                          .boxed()\n                                          .parallel(), 14).collect(Collectors.toList()));\n\n        System.out.println("Intermediate size is a multiple of stream size");\n        System.out.println(buffer(IntStream.range(0, 14)\n                                          .boxed()\n                                          .parallel(), 7).collect(Collectors.toList()));\n    }\n

\n

输出

\n

/tmp \n\xe2\x9e\x9c javac Test.java\n\n/tmp \n\xe2\x9e\x9c java Test      \n[[1, 2, 3], [4, 5, 6], [7, 8, 9], [10, 11, 12], [13]]\nEmpty stream test\n[]\nIntermediate size greater than stream size\n[[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]]\nIntermediate size same as stream size\n[[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]]\nIntermediate size is a multiple of stream size\n[[0, 1, 2, 3, 4, 5, 6], [7, 8, 9, 10, 11, 12, 13]]\n

\n

¹ _{我最初打算让它返回一个流的 Stream，但我意识到，由于流的惰性本质，使用流迭代器会使这项任务变得比需要的更加困难。}

\n