一塊兒來學Java8（七）——Stream（下）

在一塊兒來學Java8（七）——Stream（中）咱們學習了Stream.collect的用法，今天咱們來學習下Stream.reduce的用法。

reduce操做能夠理解成對Stream中元素累計處理，它有三個重載方法。

• 重載1：Optional<T> reduce(BinaryOperator<T> accumulator);
• 重載2：T reduce(T identity, BinaryOperator<T> accumulator);
• 重載3：<U> U reduce(U identity, BiFunction<U, ? super T, U> accumulator, BinaryOperator<U> combiner);

reduce(accumulator)

先來看下重載1方法，這個方法須要咱們傳入一個參數，參數名字定義爲收集器，顧名思義是須要咱們對元素進行收集。

下面經過一個數值累加的例子來講明reduce的基本用法。

Optional<Integer> opt = Stream.of(1, 2, 3, 4)
		.reduce((n1, n2) -> {
			int ret = n1 + n2;
			System.out.println(n1 + "(n1) + " + n2 + "(n2) = " + ret);
			return ret;
		});
int sum = opt.orElse(0);
System.out.println("sum:" + sum);

打印：

1(n1) + 2(n2) = 3
3(n1) + 3(n2) = 6
6(n1) + 4(n2) = 10
sum:10

這個例子中對Stream例子中的三個數字進行相加，獲得總和。最後返回一個Optional<Integer>對象是由於考慮到Stream中沒有元素的狀況，所以返回結果是未知的，應該由開發者來肯定返回值。

在Lambda表達式中提供了兩個參數n1,n2。從打印結果中能夠看出，n1,n2最開始分別是Stream中第一，第二兩個元素，把這兩個數進行相加後返回，而後帶着這個結果再次進入到Lambda表達式中，n1是前一次相加後的值，n2是下一個元素值。

這段代碼效果等同於：

int[] arr = { 1, 2, 3, 4};
int sum = 0;
for (int i : arr) {
	sum += i;
}

reduce(identity, accumulator, combiner)

再來看下重載3，這個方法有三個參數，每一個參數說明以下：

• identity：給定一個初始值
• accumulator：基於初始值，對元素進行收集概括
• combiner：對每一個accumulator返回的結果進行合併，此參數只有在並行模式中生效。

若是Stream對象是串行的，那麼只有accumulator生效，combiner是不生效的。

使用Stream.parallel()的方法開啓並行模式，使用Stream.sequential()開啓串行模式，默認開啓的是串行模式。

並行模式能夠簡單理解爲在多線程中執行，每一個線程中單獨執行它的任務。串行則是在單一線程中順序執行。

下面來看下重載3的例子：

int sum = Stream.of(1, 2, 3, 4)
		.reduce(0, (n1, n2) -> {
			int ret = n1 + n2;
			System.out.println(n1 + "(n1) + " + n2 + "(n2) = " + ret);
			return ret;
		}, (s1, s2) -> {
			int ret = s1 + s2;
			System.out.println(s1 + "(s1) + " + s2 + "(s2) = " + ret);
			return ret;
		});
System.out.println("sum:" + sum);

打印：

0(n1) + 1(n2) = 1
1(n1) + 2(n2) = 3
3(n1) + 3(n2) = 6
6(n1) + 4(n2) = 10
sum:10

能夠看到，在串行模式下並沒運行combiner參數，只運行了accumulator參數，從給定的初始值0開始累加。

這裏已經指定了初始值(identity)，所以返回類型就是初始值的類型。

咱們把例子改爲並行模式，而後看下執行結果。

int sum = Stream.of(1, 2, 3, 4)
		.parallel() // 並行模式
		.reduce(0, (n1, n2) -> {
			int ret = n1 + n2;
			System.out.println(n1 + "(n1) + " + n2 + "(n2) = " + ret);
			return ret;
		}, (s1, s2) -> {
			int ret = s1 + s2;
			System.out.println(s1 + "(s1) + " + s2 + "(s2) = " + ret);
			return ret;
		});
System.out.println("sum:" + sum);

打印：

0(n1) + 3(n2) = 3
0(n1) + 1(n2) = 1
0(n1) + 2(n2) = 2
1(s1) + 2(s2) = 3
0(n1) + 4(n2) = 4
3(s1) + 4(s2) = 7
3(s1) + 7(s2) = 10
sum:10

從打印的結果中咱們能夠看到幾個現象：

1. combiner參數被執行了
2. 打印的內容是無序的，說明它們在多線程環境下執行的
3. n1參數始終是0

由於是並行模式，前2個現象很好理解，那爲何n1參數始終是0？

由於開了並行模式後，運行reduce方法的底層是使用了ForkJoinPool(分支/合併框架)。

分支/合併框架的原理是將一個大任務拆分紅多個子任務，這些子任務並行處理本身的事情，而後框架將這些子任務的結果合併起來，生成一個最終結果。

每一個子任務之間是沒有關聯的，它們的執行狀態都是同樣的，所以每一個子任務給到的初始值（identity）都是同樣的，在本例中是0

同時須要一個合併方法用來合併每一個子任務的處理結果，而後最終返回，使用數學表達式即爲：

(0+1) + (0+2) + (0+3) + (0+4) = 10

再來看下重載3這個方法簽名，每一個參數的分工都明確了。

reduce(identity, accumulator, combiner)

• identity：初始值
• accumulator：每一個子任務執行的操做
• combiner：合併每一個子任務的結果

注意事項

查看reduce方法文檔，發現有下面一段話：

Performs a reduction on the elements of this stream, 
using the provided identity value and an associative accumulation function, 
and returns the reduced value. This is equivalent to: 

T result = identity;
for (T element : this stream)
 result = accumulator.apply(result, element)
return result;

The identity value must be an identity for the accumulator function. 
This means that for all t, accumulator.apply(identity, t) is equal to t. 
The accumulator function must be an associative function.

其中有一句重要的話：This means that for all t, accumulator.apply(identity, t) is equal to t.

簡單來講，必需要知足下面這個公式：

accumulator.apply(identity, t) == t

若是不知足的話，在並行模式下執行accumulator會有問題。

咱們把上一個例子中的初始值改爲1，而後看看執行結果

int sum = Stream.of(1, 2, 3, 4)
		.parallel()
		// 這裏改爲了1
		.reduce(1, (n1, n2) -> {
			int ret = n1 + n2;
			System.out.println(n1 + "(n1) + " + n2 + "(n2) = " + ret);
			return ret;
		}, (s1, s2) -> {
			int ret = s1 + s2;
			System.out.println(s1 + "(s1) + " + s2 + "(s2) = " + ret);
			return ret;
		});
System.out.println("sum:" + sum);

打印：

1(n1) + 3(n2) = 4
1(n1) + 4(n2) = 5
4(s1) + 5(s2) = 9
1(n1) + 2(n2) = 3
1(n1) + 1(n2) = 2
2(s1) + 3(s2) = 5
5(s1) + 9(s2) = 14
sum:14

理想中的結果應該是11纔對，即1 + 1 + 2 + 3 + 4。能夠看到在並行模式下對identity的值是有要求的。 必須知足公式：accumulator.apply(identity, t) == t。

這裏accumulator.apply(identity, t) == t即爲：accumulator.apply(1, 1) == 1，使用數學表達式表示：

1（identity） + 1 == 1

顯然這個等式是不成立的，把identity改爲0則公式成立：0 + 1 == 1

緊接着，對於combiner參數，須要知足另外一個公式：

combiner.apply(u, accumulator.apply(identity, t)) == accumulator.apply(u, t)

• t：表示第一個參數
• u：表示第二個參數

在這個例子中，咱們取第一次執行combiner狀況: 4(s1) + 5(s2) = 9，套用公式即爲：

combiner.apply(5, accumulator.apply(1, 4)) == accumulator.apply(5, 4)

在這裏u=5，identity=1，t=4

轉換成數學表達式爲：5 + (1 + 4) == 5 + 4

顯然這個等式是不成立的，把identity改爲0，等式就成立了：5 + (0 + 4) == 5 + 4

總結一下

使用reduce(identity, accumulator, combiner)方法時，必須同時知足下面兩個公式：

• 公式1，針對accumulator：accumulator.apply(identity, t) == t
• 公式2，針對combiner：combiner.apply(u, accumulator.apply(identity, t)) == accumulator.apply(u, t)

reduce(identity, accumulator)

這個方法實際上是reduce(identity, accumulator, combiner)的一種特殊形式，只不過是把combiner部分用accumulator來代替了，即

reduce(identity, accumulator)等同於reduce(identity, accumulator, accumulator)

所以reduce(identity, accumulator)的使用方式和注意事項是跟reduce(identity, accumulator, combiner)同樣的，這裏再也不贅述。

小節

本篇主要講解了Stream.reduce的使用方法及注意事項，在並行模式下，reduce是使用分支/合併框架實現的，在下一篇文章中咱們開始學習分支/合併框架。

按期分享技術乾貨，一塊兒學習，一塊兒進步！微信公衆號：猿敲月下碼