1、不小心重用了流

我敢打賭，每人至少都會(huì)犯一次這樣的錯(cuò)誤。就像現(xiàn)有的這些"流"（比如說InputStream）,你也只能對(duì)它們消費(fèi)一次。下面的代碼是無法工作的：

IntStream stream = IntStream.of(1, 2);
stream.forEach(System.out::println);

// That was fun! Let's do it again!
stream.forEach(System.out::println);

你會(huì)碰到一個(gè)這樣的錯(cuò)誤：

java.lang.IllegalStateException: stream has already been operated upon or closed

因此使用流的時(shí)候應(yīng)當(dāng)格外注意。它只能消費(fèi)一次。

2、不小心創(chuàng)建了一個(gè)"無限"流

你可能一不留神就創(chuàng)建了一個(gè)無限流。就拿下面這個(gè)例子來說：

IntStream.iterate(0, i -> i + 1)
.forEach(System.out::println);

流的問題就在于它有可能是無限的，如果你的確是這樣設(shè)計(jì)的話。唯一的問題就是，這并不是你真正想要的。因此，你得確保每次都給流提供一個(gè)適當(dāng)?shù)拇笮∠拗疲?/p>

// That's better
IntStream.iterate(0, i -> i + 1)
.limit(10)
.forEach(System.out::println);

3、不小心創(chuàng)建了一個(gè)"隱藏的"無限流

這個(gè)話題是說不完的。你可能一不小心就真的創(chuàng)建了一個(gè)無限流。比如說下面的這個(gè)：

IntStream.iterate(0, i -> ( i + 1 ) % 2)
.distinct()
.limit(10)
.forEach(System.out::println);

這樣做的結(jié)果是：

• 我們生成了0和1的交替數(shù)列
• 然后只保留不同的數(shù)值，比如說，一個(gè)0和一個(gè)1
• 然后再將流的大小限制為10
• 最后對(duì)流進(jìn)行消費(fèi)

好吧，這個(gè)distinct()操作它并不知道iterate()所調(diào)用的這個(gè)函數(shù)生成的只有兩個(gè)不同的值。它覺得可能還會(huì)有別的值。因此它會(huì)不停地從流中消費(fèi)新的值，而這個(gè)limit(10)永遠(yuǎn)也不會(huì)被調(diào)用到。不幸的是，你的應(yīng)用程序會(huì)崩掉。

4、不小心創(chuàng)建了一個(gè)"隱藏"的并行無限流

我還是想繼續(xù)提醒下你，你可能真的一不小心就消費(fèi)了一個(gè)無限流。假設(shè)你認(rèn)為distinct()操作是會(huì)并行執(zhí)行的。那你可能會(huì)這么寫：

IntStream.iterate(0, i -> ( i + 1 ) % 2)
.parallel()
.distinct()
.limit(10)
.forEach(System.out::println);

現(xiàn)在我們可以知道的是，這段代碼會(huì)一直執(zhí)行下去。不過在前面那個(gè)例子中，你至少只消耗了機(jī)器上的一個(gè)CPU。而現(xiàn)在你可能會(huì)消耗四個(gè)，一個(gè)無限流的消費(fèi)很可能就會(huì)消耗掉你整個(gè)系統(tǒng)的資源。這可相當(dāng)不妙。這種情況下你可能得去重啟服務(wù)器了。看下我的筆記本在最終崩潰前是什么樣的:

5、操作的順序

為什么我一直在強(qiáng)調(diào)你可能一不小心就創(chuàng)建了一個(gè)無限流？很簡(jiǎn)單。因?yàn)槿绻惆焉厦娴倪@個(gè)流的limit()和distinct()操作的順序掉換一下，一切就都OK了。

IntStream.iterate(0, i -> ( i + 1 ) % 2)
.limit(10)
.distinct()
.forEach(System.out::println);

現(xiàn)在則會(huì)輸出：

0

1

為什么會(huì)這樣？因?yàn)槲覀兿葘o限流的大小限制為10個(gè)值，也就是(0 1 0 1 0 1 0 1 0 1)，然后再在這個(gè)有限流上進(jìn)行歸約，求出它所包含的不同值，（0，1）。當(dāng)然了，這個(gè)在語義上就是錯(cuò)誤的了。因?yàn)槟銓?shí)際上想要的是數(shù)據(jù)集的前10個(gè)不同值。沒有人會(huì)真的要先取10個(gè)隨機(jī)數(shù)，然后再求出它們的不同值的。如果你是來自SQL背景的話，你可能不會(huì)想到還有這個(gè)區(qū)別。就拿SQL Server 2012舉例來說，下面的兩個(gè)SQL語句是一樣的：

-- Using TOP

SELECT DISTINCT TOP 10 *

FROM i

ORDER BY ..

-- Using FETCH

SELECT *

FROM i

ORDER BY ..

OFFSET 0 ROWS

FETCH NEXT 10 ROWS ONLY

因此，作為一名SQL用戶，你可能并不會(huì)注意到流操作順序的重要性。

6、還是操作順序

既然說到了SQL，如果你用的是MySQL或者PostgreSQL，你可能會(huì)經(jīng)常用到LIMIT .. OFFSET子句。SQL里全是這種暗坑，這就是其中之一。正如SQL Server 2012中的語法所說明的那樣，OFFSET子名會(huì)優(yōu)先執(zhí)行。

如果你將MySQL/PostgreSQL方言轉(zhuǎn)化成流的話，得到的結(jié)果很可能是錯(cuò)的：

IntStream.iterate(0, i -> i + 1)
.limit(10) // LIMIT
.skip(5) // OFFSET
.forEach(System.out::println);

上面的代碼會(huì)輸出：

5

6

7

8

9

是的，它輸出9后就結(jié)束了，因?yàn)槭紫壬У氖莑imit()，這樣會(huì)輸出(0 1 2 3 4 5 6 7 8 9)。其次才是skip()，它將流縮減為(5 6 7 8 9)。而這并不是你所想要的。

警惕LIMIT .. OFFSET和OFFSET .. LIMIT的陷阱！

7、使用過濾器來遍歷文件系統(tǒng)

這個(gè)問題我們之前已經(jīng)講過了。使用過濾器來遍歷文件系統(tǒng)是個(gè)不錯(cuò)的方式：

Files.walk(Paths.get("."))
.filter(p -> !p.toFile().getName().startsWith("."))
.forEach(System.out::println);

看起來上面的這個(gè)流只是遍歷了所有的非隱藏目錄，也就是不以點(diǎn)號(hào)開始的那些目錄。不幸的是，你又犯了錯(cuò)誤五和錯(cuò)誤六了。walk()方法已經(jīng)生成一個(gè)當(dāng)前目錄下的所有子目錄的流。雖然是一個(gè)惰性流，但是也包含了所有的子路徑。現(xiàn)在的這個(gè)過濾器可以正確過濾掉所有名字以點(diǎn)號(hào)開始的那些目錄，也就是說結(jié)果流中不會(huì)包含.git或者.idea。不過路徑可能會(huì)是:..git\refs或者..idea\libraries。而這并不是你實(shí)際想要的。

你可別為了解決問題而這么寫：

Files.walk(Paths.get("."))
.filter(p -> !p.toString().contains(File.separator + "."))
.forEach(System.out::println);

雖然這么寫的結(jié)果是對(duì)的，但是它會(huì)去遍歷整個(gè)子目錄結(jié)構(gòu)樹，這會(huì)遞歸所有的隱藏目錄的子目錄。

我猜你又得求助于老的JDK1.0中所提供的File.list()了。不過好消息是， FilenameFilter和FileFilter現(xiàn)在都是函數(shù)式接口了。

8、修改流內(nèi)部的集合

當(dāng)遍歷列表的時(shí)候，你不能在迭代的過程中同時(shí)去修改這個(gè)列表。這個(gè)在Java 8之前就是這樣的，不過在Java 8的流中則更為棘手。看下下面這個(gè)0到9的列表：

// Of course, we create this list using streams:
List<Integer> list =
IntStream.range(0, 10)
.boxed()
.collect(toCollection(ArrayList::new));

現(xiàn)在，假設(shè)下我們?cè)谙M(fèi)流的時(shí)候同時(shí)去刪除元素：

list.stream()
// remove(Object), not remove(int)!
.peek(list::remove)
.forEach(System.out::println);

有趣的是，其中的一些元素中可以的刪除的。你得到的輸出將會(huì)是這樣的：

0

2

4

6

8

null

null

null

null

null

java.util.ConcurrentModificationException

如果我們捕獲異常后再查看下這個(gè)列表，會(huì)發(fā)現(xiàn)一個(gè)很有趣的事情。得到的結(jié)果是：

[1, 3, 5, 7, 9]

所有的奇數(shù)都這樣。這是一個(gè)BUG嗎？不，這更像是一個(gè)特性。如果你看一下JDK的源碼，會(huì)發(fā)現(xiàn)在ArrayList.ArraListSpliterator里面有這么一段注釋：

/* * If ArrayLists were immutable, or structurally immutable (no * adds, removes, etc), we could implement their spliterators * with Arrays.spliterator. Instead we detect as much * interference during traversal as practical without * sacrificing much performance. We rely primarily on * modCounts. These are not guaranteed to detect concurrency * violations, and are sometimes overly conservative about * within-thread interference, but detect enough problems to * be worthwhile in practice. To carry this out, we (1) lazily * initialize fence and expectedModCount until the latest * point that we need to commit to the state we are checking * against; thus improving precision. (This doesn't apply to * SubLists, that create spliterators with current non-lazy * values). (2) We perform only a single * ConcurrentModificationException check at the end of forEach * (the most performance-sensitive method). When using forEach * (as opposed to iterators), we can normally only detect * interference after actions, not before. Further * CME-triggering checks apply to all other possible * violations of assumptions for example null or too-small * elementData array given its size(), that could only have * occurred due to interference. This allows the inner loop * of forEach to run without any further checks, and * simplifies lambda-resolution. While this does entail a * number of checks, note that in the common case of * list.stream().forEach(a), no checks or other computation * occur anywhere other than inside forEach itself. The other * less-often-used methods cannot take advantage of most of * these streamlinings. */

現(xiàn)在來看下如果我們對(duì)這個(gè)流排序后會(huì)是什么結(jié)果：

list.stream()
.sorted()
.peek(list::remove)
.forEach(System.out::println);

輸出的結(jié)果看起來是我們想要的：

0

1

2

3

4

5

6

7

8

9

而流消費(fèi)完后的列表是空的：

[]

也就是說所有的元素都正確地消費(fèi)掉并刪除了。sorted()操作是一個(gè)"帶狀態(tài)的中間操作"，這意味著后續(xù)的操作不會(huì)再操作內(nèi)部的那個(gè)集合了，而是在一個(gè)內(nèi)部的狀態(tài)上進(jìn)行操作。現(xiàn)在你可以安全地從列表里刪除元素了！

不過，真的是嗎這樣？我們來試一下帶有parallel(), sorted()的刪除操作：

list.stream()
.sorted()
.parallel()
.peek(list::remove)
.forEach(System.out::println);

這個(gè)會(huì)輸出 ：

7

6

2

5

8

4

1

0

9

3

現(xiàn)在列表里包含：

[8]

唉呀。居然沒有刪完所有的元素？！誰能解決這個(gè)問題，我免費(fèi)請(qǐng)他喝酒！

這些行為看起來都是不確定的，我只能建議你在使用流的時(shí)候不要去修改它內(nèi)部的數(shù)據(jù)集合。這樣做是沒用的。

9、忘了去消費(fèi)流

你覺得下面這個(gè)流在做什么？

IntStream.range(1, 5)
.peek(System.out::println)
.peek(i -> {
if (i == 5)
throw new RuntimeException("bang");
});

看完這段代碼，你覺得應(yīng)該會(huì)輸出(1 2 3 4 5)然后拋出一個(gè)異常。不過并不是這樣。它什么也不會(huì)做。這個(gè)流并沒有被消費(fèi)掉，它只是靜靜的待在那里。

正如別的流API或者DSL那樣，你可能會(huì)忘了調(diào)用這個(gè)終止操作。當(dāng)你使用peek()的時(shí)候也是這樣的，因?yàn)閜eek有點(diǎn)類似于forEach()。

在jOOQ中也存在這樣的情況，如果你忘了去調(diào)用 execute()或者fetch():

DSL.using(configuration)
.update(TABLE)
.set(TABLE.COL1, 1)
.set(TABLE.COL2, "abc")

.where(TABLE.ID.eq(3));

杯具。忘了調(diào)用execute方法了。

10、并行流死鎖

終于快講完了～

如果你沒有正確地進(jìn)行同步的話，所有的并發(fā)系統(tǒng)都可能碰到死鎖。現(xiàn)實(shí)中的例子可能不那么明顯，不過如果你想自己創(chuàng)造一個(gè)場(chǎng)景的話倒是很容易。下面這個(gè)parallel()流肯定會(huì)造成死鎖：

Object[] locks = { new Object(), new Object() };

IntStream
.range(1, 5)
.parallel()
.peek(Unchecked.intConsumer(i -> {
synchronized (locks[i % locks.length]) {
Thread.sleep(100);

synchronized (locks[(i + 1) % locks.length]) {
Thread.sleep(50);
}
}
}))
.forEach(System.out::println);

注意這里Unchecked.intConsumer()的使用，它把IntConsumer接口轉(zhuǎn)化成了 org.jooq.lambda.fi.util.function.CheckedIntConsumer，這樣你才可以拋出已檢查異常。

好吧。這下你的機(jī)器倒霉了。這些線程會(huì)一直阻塞下去:-)。不過好消息就是，在Java里面要寫出一個(gè)這種教科書上的死鎖可不是那么容易。

想進(jìn)一步了解的話，可以看下Brian Goetz在StackOverflow上的一個(gè)回答。

結(jié)論

引入了流和函數(shù)式編程之后，我們開始會(huì)碰到許多新的難以發(fā)現(xiàn)的BUG。這些BUG很難避免，除非你見過并且還時(shí)刻保持警惕。你必須去考慮操作的順序，還得注意流是不是無限的。

流是一個(gè)非常強(qiáng)大的工具，但也是一個(gè)首先得去熟練掌握的工具。

源自//it.deepinmind.com/index.html