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

短命女 2022-11-06 13:56 150阅读 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 条评论，150人围观）

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

相关阅读

相关 Java8 函数式编程

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

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

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

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

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

相关 Java8 函数式编程

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

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

相关 java8函数式编程（一）

1.常见函数接口 ![这里写图片描述][70] 例如： public static void main(String[] args) {

逃离我推掉我的手/ 2022年05月31日 13:21/ 0 赞/ 227 阅读

相关 java8函数式编程(2)

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

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

相关 java8函数式编程（三）

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

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

相关 java8函数式编程（四）

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

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

相关 Java8 函数式式编程

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

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

相关 java8函数式编程实例

什么是函数式编程函数式编程是java8的一大特色，也就是将函数作为一个参数传递给指定方法。别人传的要么是基本数据类型，要么就是地址引用，我们要穿一个“动作”。 S

迈不过友情╰/ 2021年12月15日 11:33/ 0 赞/ 346 阅读

相关 JAVA8函数式编程

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

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