Java 并行操作之流的使用
在實(shí)際開發(fā)過程中，我們使用最多的程序結(jié)構(gòu)是順序結(jié)構(gòu)，也就是序列 sequence，通常還會與循環(huán)結(jié)構(gòu)的代碼相混合 Loops。
順序結(jié)構(gòu)只能依次執(zhí)行任務(wù)，消耗的時(shí)間是各個(gè)順序任務(wù)的總和。當(dāng)對接口的請求時(shí)間有要求，需要對任務(wù)的處理時(shí)間進(jìn)行壓縮，轉(zhuǎn)化到代碼上就是對代碼結(jié)構(gòu)進(jìn)行優(yōu)化，我們希望將序列任務(wù)轉(zhuǎn)化為并行任務(wù) parallelism。并行指的是同時(shí)發(fā)生，在計(jì)算機(jī)上為每個(gè)任務(wù)分配 單獨(dú)的 CPU 核心，同時(shí)進(jìn)行任務(wù)處理完畢后進(jìn)行匯總。如果你了解 Hadoop ，這種方式與 MapReduce 如出一轍。
我們可以調(diào)用 parallelStream 方法來把集合轉(zhuǎn)換為并行流。并行流就是一個(gè)把內(nèi)容分成多個(gè)數(shù)據(jù)塊，并用不同的線程分別處理每個(gè)數(shù)據(jù)塊的流。這樣就可以把工作負(fù)荷分配給多核處理器的所有內(nèi)核，讓它們都忙起來。
什么是流
流是 JavaAPI 的新成員，它允許你以聲明性方式處理數(shù)據(jù)集合（通過查詢語句來表達(dá)，而不是臨時(shí)編寫一個(gè)實(shí)現(xiàn)）。就現(xiàn)在來說，你可以把它們看成遍歷數(shù)據(jù)集的高級迭代器。
Stream<T>
支持順序 sequential 和并行 parallel 聚合操作的元素序列。下面的示例演示了使用 Stream和 IntStream 的聚合操作。
int sum = widgets.stream()
.filter(w -> w.getColor() == RED)
.mapToInt(w -> w.getWeight())
.sum();在這個(gè)例子中，widgets是一個(gè) Collection<Widget>。我們通過以下方法創(chuàng)建一個(gè)小部件對象流 Collection.stream() 。該方法返回以此集合為源的順序流。如上方法過濾顏色為紅色的 Widgets 并將 Collection<widget> 轉(zhuǎn)化為Collection<int> 的流，并對所有int流求和。
流在與集合主要不同：
沒有存儲空間。流不是存儲元素的數(shù)據(jù)結(jié)構(gòu)。
本質(zhì)上是功能性的。對流的操作會產(chǎn)生結(jié)果，但不會修改其源。例如，上面的例子不會修改 Collection<Widget>。
消耗品。在流的生存期內(nèi)，流中元素只能訪問一次。
流操作分為中間（intermediate ）操作和終端（terminal ）操作，并合并以形成流管道（stream pipelines）。流管道由源（例如 Collection，數(shù)組，生成器函數(shù)或I / O通道）組成；隨后是零個(gè)或多個(gè)中間操作，例如 Stream.filter 或 Stream.map；以及諸如Stream.forEach 或終端操作 Stream.reduce。
終端操作（例如Stream.forEach或 IntStream.sum）可能會遍歷流執(zhí)行操作。執(zhí)行終端操作后，流管道被視為已消耗，無法再使用；如果需要再次遍歷同一數(shù)據(jù)源，則必須返回到數(shù)據(jù)源以獲取新的流。
流的各種操作
我們使用一個(gè) Menu 菜單列表的例子來看一下流的各種操作
List < Dish > menu = Arrays.asList( new Dish("pork", false, 800, Dish.Type.MEAT),
new Dish("beef", false, 700, Dish.Type.MEAT),
new Dish("chicken", false, 400, Dish.Type.MEAT),
new Dish("french fries", true, 530, Dish.Type.OTHER),
new Dish("rice", true, 350, Dish.Type.OTHER),
new Dish("season fruit", true, 120, Dish.Type.OTHER),
new Dish("pizza", true, 550, Dish.Type.OTHER),
new Dish("prawns", false, 300, Dish.Type.FISH),
new Dish("salmon", false, 450, Dish.Type.FISH) );
public class Dish {
private final String name;
private final boolean vegetarian;
private final int calories;
private final Type type;
public boolean isVegetarian() { return vegetarian; }
public enum Type { MEAT, FISH, OTHER }
//...省略 Getter Setter
}
如上的集合可以使用如下流操作
import static java.util.Comparator.comparing;
import static java.util.stream.Collectors.toList;
List <String> lowCaloricDishesName = menu.stream()
.filter(d -> d.getCalories() < 400) // 選出400卡路里以下的菜肴
.sorted(comparing(Dish::getCalories)) // 按照卡路里排序
.map(Dish::getName) // 提取菜肴的名稱
.limit(3) // 只選擇頭三個(gè)
.collect(toList()); // 將所有名稱保存到 List 中
為 了 利 用 多 核 架 構(gòu) 并 行 執(zhí) 行 這 段 代 碼， 你 只 需 要 把 stream() 換 成 parallelStream() ：
List < String > lowCaloricDishesName = menu.parallelStream()
.filter(d -> d.getCalories() < 400)
.sorted(comparing( Dishes::getCalories))
.map(Dish::getName)
.limit(3)
.collect(toList());
數(shù)據(jù)處理操作——流的數(shù)據(jù)處理功能支持類似于數(shù)據(jù)庫的操作，以及函數(shù)式編程語言中的常用操作，如 filter 、map 、reduce 、find 、match 、sort 等。流操作可以順序執(zhí)行，也可并行執(zhí)行。
如上操作中
filter —— 接受 Lambda， 從流中排除某些信息
map —— 接受 Lambada，將元素轉(zhuǎn)化為其他形式或提取信息。
limit —— 截?cái)嗔?，使元素不超過指定數(shù)量。
collect ——將流轉(zhuǎn)化為其他形式。
你可以看到兩類操作：filter、map 和 limit 可以連成一條流水線；collect 觸發(fā)流水線執(zhí)行并關(guān)閉它?？梢赃B接起來的流操作稱為中間操作，關(guān)閉流的操作稱為終端操作。
流的三要素
一個(gè)數(shù)據(jù)源（如集合）來執(zhí)行一個(gè)查詢；
一個(gè)中間操作鏈，形成一條流的流水線；
一個(gè)終端操作，執(zhí)行流水線，并能生成結(jié)果。
中間操作 ：filter、map、limit、sorted、distinct
終端操作：forEach、count、collect
流的中間操作
使用謂詞篩選
filter() 操作會接受一個(gè)謂詞（一個(gè)返回boolean的函數(shù)）作為參數(shù)，并返回一個(gè)包括所有符合謂詞的元素的流。filter(d -> d.getCalories() < 400) 這個(gè)代碼中定義了一個(gè)匿名內(nèi)部類，這個(gè)用 Lambda 表達(dá)式定義的方法返回 d.getCalories()<400 這個(gè)表達(dá)式的值，隨后 filter 會根據(jù)過濾條件進(jìn)行流過濾。
也可以使用 filter(Dish::isVegetarian) ，有雙冒號的用法，就是把方法當(dāng)做參數(shù)傳到stream內(nèi)部，使 stream 的每個(gè)元素都傳入到該方法里面執(zhí)行一下。Dish::isVegetarian 返回每個(gè)元素的 true 或 false 值。
映射
map 方法會接受一個(gè)函數(shù)作為參數(shù)。這個(gè)函數(shù)會被應(yīng)用到每個(gè)元素上，并將其映射成一個(gè)新的元素。
List < String > dishNames = menu.stream()
.map(Dish::getName)
.collect(toList());
map 方法返回的流類型是 Stream<String>。
匹配
anyMatch 方法返回 Boolean 值，用來查找流中是否包含某個(gè)值，是一個(gè)終端操作。
if( menu.stream().anyMatch(Dish::isVegetarian)){
System.out.println(" The menu is vegetarian friendly!!");
}
AllMatch 方法也返回 Boolean 值，用來判斷是否流中元素是否全部匹配。
NoneMatch 方法確保流中沒有任何元素與謂詞相匹配。
查找
可以使用流的 find 方法查找流中元素，可以使用順序查找 findFirst() 或任意查找 findAny() 。查找方法是終端操作。
List <Integer> someNumbers = Arrays.asList(1, 2, 3, 4, 5);
Optional <Integer> firstSquareDivisibleByThree = someNumbers.stream()
.map( x -> x * x)
.filter( x -> x % 3 = = 0)
.findFirst(); // 9
規(guī)約操作
規(guī)約操作指將流規(guī)約成一個(gè)值。也可稱為折疊 fold。
元素求和
int sum = 0;
for (int x : numbers) { sum + = x; }
// 使用流的寫法
int sum = numbers.stream().reduce(0, (a, b) -> a + b);
reduce 接受兩個(gè)參數(shù)：
一個(gè)初始值，這里是 0；
一個(gè) BinaryOperator 來將兩個(gè)元素結(jié)合起來產(chǎn)生一個(gè)新值，這里我們用的是 lambda (a,b)->a+b。
如上的例子中，0 作為 a 的第一個(gè)參數(shù)，從流中獲取 numbers 列表中的一個(gè)數(shù)值如 3 作為 b ，兩者之和 3 作為 a 的第二個(gè)參數(shù)與 numbers 列表獲取的第二個(gè)數(shù)值 b =4 相加，直到所有 numbers 列表中數(shù)據(jù)都參與計(jì)算返回最終的 a + b 的 sum 值。
最大值與最小值
Optional <Integer> max = numbers.stream().reduce(Integer::max);
Optional <Integer> min = numbers.stream().reduce(Integer::min);
// 也可以使用 Lambda (X,Y)->X<Y?X:Y數(shù)值流
int calories = menu.stream() .map(Dish:: getCalories) .reduce( 0, Integer:: sum);
int calories = menu.stream() // 返回一個(gè) Stream <Dish>.mapToInt(Dish::getCalories) //返回一個(gè) IntStream.sum();
IntStream intStream = menu.stream(). mapToInt( Dish:: getCalories); // 將 Stream 轉(zhuǎn)換為數(shù)值流
Stream <Integer > stream = intStream.boxed(); // 將數(shù)值流轉(zhuǎn)換為 Stream
并行流
并行計(jì)算數(shù)值和
public static long parallelSum( long n) {
return Stream.iterate(1L, i -> i + 1)
.limit(n)
.parallel() //將流轉(zhuǎn)換為并行流
.reduce(0L, Long:: sum);
}
并行流默認(rèn)的線程數(shù)量是你的處理器數(shù)量，要保證在內(nèi)核中并行執(zhí)行工作的時(shí)間比在內(nèi)核之間傳輸數(shù)據(jù)的時(shí)間長。使用并行流要避免共享對象可變狀態(tài)，并且要避免自動拆裝箱。對于較小的數(shù)據(jù)一般不使用并行流，并行處理少數(shù)幾個(gè)元素可能耗費(fèi)時(shí)間增多，因?yàn)橐苿佣鄠€(gè)核的數(shù)據(jù)代價(jià)也很大。
參考資料
《 Java 8實(shí)戰(zhàn) 》 . 人民郵電出版社.
JDK 11 官方文檔