將任意源流劃分為固定大小的批次是不可能的，因為這會加重并行處理。 并行處理時，您可能不知道拆分后第一個子任務中有多少個元素，因此您無法為下一個子任務創建分區，直到完全處理第一個子任務。

但是，可以從隨機訪問ofSubLists創建分區流。例如，在我的ofSubLists庫中可以使用這種功能：

List input = Arrays.asList(...);

Stream> stream = StreamEx.ofSubLists(input, partitionSize);

或者，如果您真的想要流：

Stream> stream = StreamEx.ofSubLists(input, partitionSize).map(List::stream);

如果您不想依賴第三方庫，則可以手動實現此類ofSubLists方法：

public static Stream> ofSubLists(List source, int length) {

if (length <= 0)

throw new IllegalArgumentException("length = " + length);

int size = source.size();

if (size <= 0)

return Stream.empty();

int fullChunks = (size - 1) / length;

return IntStream.range(0, fullChunks + 1).mapToObj(

n -> source.subList(n * length, n == fullChunks ? size : (n + 1) * length));

}

該實現看起來有些長，但是考慮了一些極端情況，例如接近MAX_VALUE列表大小。

如果您想為無序流提供并行友好的解決方案(因此您不必擔心將哪些流元素合并為單個批處理)，則可以使用這樣的收集器(感謝@sibnick的啟發)：

public static Collector unorderedBatches(int batchSize,

Collector, A, R> downstream) {

class Acc {

List cur = new ArrayList<>();

A acc = downstream.supplier().get();

}

BiConsumer accumulator = (acc, t) -> {

acc.cur.add(t);

if(acc.cur.size() == batchSize) {

downstream.accumulator().accept(acc.acc, acc.cur);

acc.cur = new ArrayList<>();

}

};

return Collector.of(Acc::new, accumulator,

(acc1, acc2) -> {

acc1.acc = downstream.combiner().apply(acc1.acc, acc2.acc);

for(T t : acc2.cur) accumulator.accept(acc1, t);

return acc1;

}, acc -> {

if(!acc.cur.isEmpty())

downstream.accumulator().accept(acc.acc, acc.cur);

return downstream.finisher().apply(acc.acc);

}, Collector.Characteristics.UNORDERED);

}

用法示例：

List> list = IntStream.range(0,20)

.boxed().parallel()

.collect(unorderedBatches(3, Collectors.toList()));

結果：

[[2, 3, 4], [7, 8, 9], [0, 1, 5], [12, 13, 14], [17, 18, 19], [10, 11, 15], [6, 16]]

這樣的收集器是完全線程安全的，并為順序流生成有序批。

如果要對每個批次應用中間轉換，則可以使用以下版本：

public static Collector unorderedBatches(int batchSize,

Collector batchCollector,

Collector downstream) {

return unorderedBatches(batchSize,

Collectors.mapping(list -> list.stream().collect(batchCollector), downstream));

}

例如，通過這種方式，您可以即時對每個批次中的數字求和：

List list = IntStream.range(0,20)

.boxed().parallel()

.collect(unorderedBatches(3, Collectors.summingInt(Integer::intValue),

Collectors.toList()));

總結

以上是生活随笔為你收集整理的java函数式编程 map_函数式编程-对Java 8流进行分区的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。