java函数式编程 map_函数式编程-对Java 8流进行分区

短命女 2022-11-06 13:56 206阅读 0赞

将任意源流划分为固定大小的批次是不可能的，因为这会加重并行处理。 并行处理时，您可能不知道拆分后第一个子任务中有多少个元素，因此您无法为下一个子任务创建分区，直到完全处理第一个子任务。

但是，可以从随机访问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()));

发表评论取消回复

表情：

评论列表（有 0 条评论，206人围观）

还没有评论，来说两句吧...

相关阅读

相关 Java 8新特性：流式编程与函数式接口

Java 8引入了一系列新特性，其中一个是流式编程（Stream Processing），另一个是函数式接口（Functional Interface）。 1. 流式编程：

朴灿烈づ我的快乐病毒、/ 2024年12月08日 14:06/ 0 赞/ 33 阅读

相关 java8函数式编程

[1、Java8 中 List 转 Map(Collectors.toMap) 使用技巧][1_Java8 _ List _ Map_Collectors.toMap_]

╰半橙微兮°/ 2023年10月03日 13:54/ 0 赞/ 36 阅读

相关 Java8 函数式编程

文章目录 Java 函数式编程 1. Lambda 表达式 1.1 标准格式 1.2 使用前提

红太狼/ 2023年09月26日 10:35/ 0 赞/ 162 阅读

相关 java函数式编程 map_函数式编程-对Java 8流进行分区

将任意源流划分为固定大小的批次是不可能的，因为这会加重并行处理。并行处理时，您可能不知道拆分后第一个子任务中有多少个元素，因此您无法为下一个子任务创建分区，直到完全处理第一个

短命女/ 2022年11月06日 13:56/ 0 赞/ 207 阅读

相关 Java8 函数式编程

Java8 函数式编程 1. Java8 接口变化（default）先看下Java8 中的迭代器接口中的默认方法 public interface It

比眉伴天荒/ 2022年09月11日 04:16/ 0 赞/ 207 阅读

相关 java8函数式编程(2)

/ @author zhangdi @description 常用的流操作 / public class LambdaChapter3_3 {

朴灿烈づ我的快乐病毒、/ 2022年05月22日 10:43/ 0 赞/ 275 阅读

相关 java8函数式编程（三）

并行化流操作如果已经有一个Stream对象，调用parallel方法即拥有并行操作的能力 boolean b = Stream.of("z", "s", "c

末蓝、/ 2022年05月08日 05:38/ 0 赞/ 270 阅读

相关 java8函数式编程（四）

相关链接： [java8函数式编程（一）][java8] [java8函数式编程（二）][java8 1] [java8函数式编程（三）][java8 2] 其

末蓝、/ 2022年04月11日 11:56/ 0 赞/ 274 阅读

相关 Java8 函数式式编程

文章目录 1. 引言 2. 第一个Lambda表达式 2.1 如何辨别Lambda表达式 2.2 闭包 2.3 函数接

秒速五厘米/ 2022年04月10日 13:06/ 0 赞/ 356 阅读

相关 JAVA8函数式编程

[2019独角兽企业重金招聘Python工程师标准>>> ][2019_Python_] ![hot3.png][] 提示： JAVA8由->表示分割，左半部分表示

曾经终败给现在/ 2021年09月20日 02:54/ 0 赞/ 469 阅读