自定義 ForkJoinPool 提高並行流 ParallelStream 執行速度

時間 2019-11-07

標籤自定義 forkjoinpool 提高並行 parallelstream 執行速度欄目 Mac OS 简体版

原文原文鏈接

簡介

在 java8 中添加了流Stream，可讓你以一種聲明的方式處理數據。使用起來很是簡單優雅。ParallelStream 則是一個並行執行的流，採用 ForkJoinPool 並行執行任務，提升執行速度。java

下面咱們看看2個簡單的示例：git

示例1 （list）

Arrays.asList(1,2,3,4,5,6)
	.parallelStream()
	.forEach((value) -> {
		String name = Thread.currentThread().getName();
		System.out.println("示例1 Thread:" + name + " value:" + value);
	});
複製代碼

示例2 （array）

Stream.of(1,2,3,4,5,6)
	.parallel()
	.forEach((value) -> {
		String name = Thread.currentThread().getName();
		System.out.println("示例2 Thread:" + name + " value:" + value);
	});
複製代碼

問題引出

筆者最近在作一些爬蟲相關的業務，其核心工具已開源 mica-http：gitee.com/596392912/m… ，通過2個版本的迭代已經發展成了一個強大非帳號爬蟲利器，趕忙來試試吧。數據庫

咱們採集了大量的代理 ip 用來供爬蟲使用，其中有個定時任務每 5 分鐘去檢測代理是否失效，代理 ip 檢測比較費時，咱們給每一個檢測的請求設定了 2s 的超時，這樣單線程的話 1000 個 ip 就得消耗半個多小時，固然筆者在校驗的時候採用的 parallel Stream 簡化開發。bash

而後發現效果並不明顯，代理 ip 數量上來以後 5 分鐘徹底檢測不完，致使任務堆積。明明用了併發流爲何沒有明顯的提升執行速度呢？多線程

下面咱們來看看剛剛的「示例」打印出的信息：架構

示例1 Thread:main value:4
示例1 Thread:ForkJoinPool.commonPool-worker-2 value:1
示例1 Thread:main value:6
示例1 Thread:ForkJoinPool.commonPool-worker-2 value:5
示例1 Thread:main value:3
示例1 Thread:ForkJoinPool.commonPool-worker-1 value:2
示例2 Thread:main value:4
示例2 Thread:ForkJoinPool.commonPool-worker-3 value:3
示例2 Thread:ForkJoinPool.commonPool-worker-2 value:5
示例2 Thread:ForkJoinPool.commonPool-worker-4 value:1
示例2 Thread:ForkJoinPool.commonPool-worker-5 value:2
示例2 Thread:ForkJoinPool.commonPool-worker-1 value:6
複製代碼

咱們能夠看到 Parallel Stream，默認採用的是一個 ForkJoinPool.commonPool 的線程池，這樣咱們就算使用了 Parallel Stream，整個 jvm 共用一個 common pool 線程池，一不當心就職務堆積了，在校驗代理 ip 的時候咱們還有采集代理等其餘的任務中也大量使用了併發流，這樣也就印證了爲何會任務堆積了。併發

解決問題

使用自定義 ForkJoinPool 執行速度。示例代碼以下：jvm

// 示例：自定義線程池
ForkJoinPool forkJoinPool = new ForkJoinPool(8);

// 這裏是從數據庫裏查出來的一批代理 ip
List<ProxyList> records = new ArrayList<>();

// 找出失效的代理 ip
List<String> needDeleteList = forkJoinPool.submit(() -> records.parallelStream()
	.map(ProxyList::getIpPort)
	.filter(IProxyListTask::isFailed)
	.collect(Collectors.toList())
).join();

// 刪除失效的代理
複製代碼

整個代碼依然比較優雅，在使用自定義的 ForkJoin 線程池以後，執行速度有了明顯的提高。之前 5 分鐘執行不完的任務如今 2 分鐘以內就能所有執行完畢。微服務