為什么需要 Stream

Stream 作為 Java 8 的一大亮點(diǎn)，它與 java.io 包里的 InputStream 和 OutputStream 是完全不同的概念。它也不同于 StAX 對(duì) XML 解析的 Stream，也不是 Amazon Kinesis 對(duì)大數(shù)據(jù)實(shí)時(shí)處理的 Stream。Java 8 中的 Stream 是對(duì)集合（Collection）對(duì)象功能的增強(qiáng)，它專注于對(duì)集合對(duì)象進(jìn)行各種非常便利、高效的聚合操作（aggregate operation），或者大批量數(shù)據(jù)操作 (bulk data operation)。Stream API 借助于同樣新出現(xiàn)的 Lambda 表達(dá)式，極大的提高編程效率和程序可讀性。同時(shí)它提供串行和并行兩種模式進(jìn)行匯聚操作，并發(fā)模式能夠充分利用多核處理器的優(yōu)勢(shì)，使用 fork/join 并行方式來(lái)拆分任務(wù)和加速處理過(guò)程。通常編寫并行代碼很難而且容易出錯(cuò), 但使用 Stream API 無(wú)需編寫一行多線程的代碼，就可以很方便地寫出高性能的并發(fā)程序。所以說(shuō)，Java 8 中首次出現(xiàn)的 java.util.stream 是一個(gè)函數(shù)式語(yǔ)言+多核時(shí)代綜合影響的產(chǎn)物。

什么是流

Stream不是集合元素，它不是數(shù)據(jù)結(jié)構(gòu)并不保存數(shù)據(jù)，它是有關(guān)算法和計(jì)算的，它更像一個(gè)高級(jí)版本的 Iterator。原始版本的 Iterator，用戶只能顯式地一個(gè)一個(gè)遍歷元素并對(duì)其執(zhí)行某些操作；高級(jí)版本的 Stream，用戶只要給出需要對(duì)其包含的元素執(zhí)行什么操作，比如 “過(guò)濾掉長(zhǎng)度大于 10 的字符串”、“獲取每個(gè)字符串的首字母”等，Stream 會(huì)隱式地在內(nèi)部進(jìn)行遍歷，做出相應(yīng)的數(shù)據(jù)轉(zhuǎn)換。Stream 就如同一個(gè)迭代器（Iterator），單向，不可往復(fù)，數(shù)據(jù)只能遍歷一次，遍歷過(guò)一次后即用盡了，就好比流水從面前流過(guò)，一去不復(fù)返。而和迭代器又不同的是，Stream 可以并行化操作，迭代器只能命令式地、串行化操作。顧名思義，當(dāng)使用串行方式去遍歷時(shí)，每個(gè) item 讀完后再讀下一個(gè) item。而使用并行去遍歷時(shí)，數(shù)據(jù)會(huì)被分成多個(gè)段，其中每一個(gè)都在不同的線程中處理，然后將結(jié)果一起輸出。Stream 的并行操作依賴于 Java7 中引入的 Fork/Join 框架（JSR166y）來(lái)拆分任務(wù)和加速處理過(guò)程。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 的另外一大特點(diǎn)是，數(shù)據(jù)源本身可以是無(wú)限的。

1、一般進(jìn)行批處理時(shí)會(huì)將數(shù)據(jù)加入到一個(gè)臨時(shí)的集合中，當(dāng)數(shù)據(jù)量達(dá)到一定大小后進(jìn)行下一步操作，數(shù)據(jù)量不足時(shí)需要進(jìn)行額外的判斷；

2、若使用Java8的Stream流中的 skip + limit 則可以讓我們對(duì)集合方便快捷的操作，其中：
（1）skip(x)：返回丟棄流中的前x個(gè)元素后剩下元素組成的新流；若原流中包含的元素個(gè)數(shù)小于x，則返回空流。
（2）limit(x): 對(duì)一個(gè)Stream流進(jìn)行截?cái)嗖僮?，獲取其前x個(gè)元素；若原流中包含的元素個(gè)數(shù)小于x，那就獲取其所有的元素；

3、實(shí)例代碼如下：

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);
        }
}

輸出結(jié)果：

[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、通過(guò)結(jié)果也可以看出 skip 配合 limit 操作使用時(shí), limit 是最多取出限制的大小，不用再判斷最后一批數(shù)據(jù)量大小不夠限定的大小時(shí)的處理。

改進(jìn)版

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;
}

到此這篇關(guān)于使用Java8 Stream流的skip + limit實(shí)現(xiàn)批處理的文章就介紹到這了,更多相關(guān)Java8 Stream流批處理內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論