1、不小心重用了流

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

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

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

你會碰到一個這樣的錯誤：

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

因此使用流的時候應當格外注意。它只能消費一次。

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

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

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

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

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

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

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

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

這樣做的結果是：

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

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

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

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

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

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

5、操作的順序

為什么我一直在強調你可能一不小心就創(chuàng)建了一個無限流？很簡單。因為如果你把上面的這個流的limit()和distinct()操作的順序掉換一下，一切就都OK了。

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

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

0

1

為什么會這樣？因為我們先將無限流的大小限制為10個值，也就是(0 1 0 1 0 1 0 1 0 1)，然后再在這個有限流上進行歸約，求出它所包含的不同值，（0，1）。當然了，這個在語義上就是錯誤的了。因為你實際上想要的是數(shù)據(jù)集的前10個不同值。沒有人會真的要先取10個隨機數(shù)，然后再求出它們的不同值的。如果你是來自SQL背景的話，你可能不會想到還有這個區(qū)別。就拿SQL Server 2012舉例來說，下面的兩個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用戶，你可能并不會注意到流操作順序的重要性。

6、還是操作順序

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

如果你將MySQL/PostgreSQL方言轉化成流的話，得到的結果很可能是錯的：

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

上面的代碼會輸出：

5

6

7

8

9

是的，它輸出9后就結束了，因為首先生效的是limit()，這樣會輸出(0 1 2 3 4 5 6 7 8 9)。其次才是skip()，它將流縮減為(5 6 7 8 9)。而這并不是你所想要的。

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

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

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

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

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

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

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

雖然這么寫的結果是對的，但是它會去遍歷整個子目錄結構樹，這會遞歸所有的隱藏目錄的子目錄。

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

8、修改流內部的集合

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

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

現(xiàn)在，假設下我們在消費流的時候同時去刪除元素：

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

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

0

2

4

6

8

null

null

null

null

null

java.util.ConcurrentModificationException

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

[1, 3, 5, 7, 9]

所有的奇數(shù)都這樣。這是一個BUG嗎？不，這更像是一個特性。如果你看一下JDK的源碼，會發(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)在來看下如果我們對這個流排序后會是什么結果：

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

輸出的結果看起來是我們想要的：

0

1

2

3

4

5

6

7

8

9

而流消費完后的列表是空的：

[]

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

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

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

這個會輸出 ：

7

6

2

5

8

4

1

0

9

3

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

[8]

唉呀。居然沒有刪完所有的元素？！誰能解決這個問題，我免費請他喝酒！

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

9、忘了去消費流

你覺得下面這個流在做什么？

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

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

正如別的流API或者DSL那樣，你可能會忘了調用這個終止操作。當你使用peek()的時候也是這樣的，因為peek有點類似于forEach()。

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

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

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

杯具。忘了調用execute方法了。

10、并行流死鎖

終于快講完了～

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

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接口轉化成了 org.jooq.lambda.fi.util.function.CheckedIntConsumer，這樣你才可以拋出已檢查異常。

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

想進一步了解的話，可以看下Brian Goetz在StackOverflow上的一個回答。

結論

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

流是一個非常強大的工具，但也是一個首先得去熟練掌握的工具。

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