使用Java8 Stream流的skip + limit实现批处理的方法

为什么需要 Stream

Stream 作为 Java 8 的一大亮点，它与 java.io 包里的 InputStream 和 OutputStream 是完全不同的概念。它也不同于 StAX 对 XML 解析的 Stream，也不是 Amazon Kinesis 对大数据实时处理的 Stream。Java 8 中的 Stream 是对集合（Collection）对象功能的增强，它专注于对集合对象进行各种非常便利、高效的聚合操作（aggregate operation），或者大批量数据操作 (bulk data operation)。Stream API 借助于同样新出现的 Lambda 表达式，极大的提高编程效率和程序可读性。同时它提供串行和并行两种模式进行汇聚操作，并发模式能够充分利用多核处理器的优势，使用 fork/join 并行方式来拆分任务和加速处理过程。通常编写并行代码很难而且容易出错, 但使用 Stream API 无需编写一行多线程的代码，就可以很方便地写出高性能的并发程序。所以说，Java 8 中首次出现的 java.util.stream 是一个函数式语言+多核时代综合影响的产物。

什么是流

Stream不是集合元素，它不是数据结构并不保存数据，它是有关算法和计算的，它更像一个高级版本的 Iterator。原始版本的 Iterator，用户只能显式地一个一个遍历元素并对其执行某些操作；高级版本的 Stream，用户只要给出需要对其包含的元素执行什么操作，比如 “过滤掉长度大于 10 的字符串”、“获取每个字符串的首字母”等，Stream 会隐式地在内部进行遍历，做出相应的数据转换。Stream 就如同一个迭代器（Iterator），单向，不可往复，数据只能遍历一次，遍历过一次后即用尽了，就好比流水从面前流过，一去不复返。而和迭代器又不同的是，Stream 可以并行化操作，迭代器只能命令式地、串行化操作。顾名思义，当使用串行方式去遍历时，每个 item 读完后再读下一个 item。而使用并行去遍历时，数据会被分成多个段，其中每一个都在不同的线程中处理，然后将结果一起输出。Stream 的并行操作依赖于 Java7 中引入的 Fork/Join 框架（JSR166y）来拆分任务和加速处理过程。Java 的并行 API 演变历程基本如下：

• 1.0-1.4 中的 java.lang.Thread
• 5.0 中的 java.util.concurrent
• 6.0 中的 Phasers 等
• 7.0 中的 Fork/Join 框架
• 8.0 中的 Lambda

Stream 的另外一大特点是，数据源本身可以是无限的。

1、一般进行批处理时会将数据加入到一个临时的集合中，当数据量达到一定大小后进行下一步操作，数据量不足时需要进行额外的判断；

2、若使用Java8的Stream流中的 skip + limit 则可以让我们对集合方便快捷的操作，其中：
（1）skip(x)：返回丢弃流中的前x个元素后剩下元素组成的新流；若原流中包含的元素个数小于x，则返回空流。
（2）limit(x): 对一个Stream流进行截断操作，获取其前x个元素；若原流中包含的元素个数小于x，那就获取其所有的元素；

3、实例代码如下：

public static void main(String[] args) {
        List<Integer> list = new ArrayList<>();
        for (int i = 10; i < 36; i++) {
            list.add(i);
        }
        int limit = 10;
        for (int offset = 0; offset < list.size(); offset += limit) {
            List<Integer> subList = list.stream()
                                        .skip(offset)
                                        .limit(limit)
                                        .collect(Collectors.toList());
            System.out.println(subList);
        }
}

输出结果：

[10, 11, 12, 13, 14, 15, 16, 17, 18, 19]
[20, 21, 22, 23, 24, 25, 26, 27, 28, 29]
[30, 31, 32, 33, 34, 35]

3、通过结果也可以看出 skip 配合 limit 操作使用时, limit 是最多取出限制的大小，不用再判断最后一批数据量大小不够限定的大小时的处理。

改进版

int j = 0, size = list.size(), batchSize = 100;
while (j < size) {
    batchList = list.stream().skip(j).limit(Math.min(j + batchSize, size) - j).collect(Collectors.toList());
    j += batchSize;
}

到此这篇关于使用Java8 Stream流的skip + limit实现批处理的文章就介绍到这了,更多相关Java8 Stream流批处理内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

最新评论