Java. 的 Stream API，有經驗的 Java 工程師應該都不陌生了。它為數據處理邏輯提供了一組串化的操作序列接口，為流式的業務提供了一個整齊一致的接口。但是要指出的是，在常見的編程語言中，Java的標準庫幾乎是這方面最弱的。而Akka Stream 是一組非常強大的工具集，它允許我們用非常直觀的方式定義數據流，不僅僅是線性流，還可以是有向圖。例如官方給出的這個例子：

val

它直接對應的這樣一個邏輯：

圖片來自 Akka Stream 官方文檔

我相信很多同行即使沒有scala的經驗，對照這個圖也能很容易的理解代碼。但是這里面有一個問題，同樣功能的代碼在 Java 中是這樣的：

final

相比之下 Java 版本冗長的實在過分了。

這還是一個非常簡單的圖。做量化數據訂閱的時候，我還寫過一些更復雜的圖，我需要把服務器訂閱的數據解壓，然后分離心跳數據和業務數據，響應后再從出口發送回去。這個過程用 Java 寫實在是太不友善了。

但是好在我們有 Clojure ，這幾天我想了一下，為 Akka Stream 做了一些 Clojure 的封裝。

Scala 的 graph dsl 之所以簡潔有力， `~>` 操作符功不可沒。Java 因為沒有操作符重載，只能用方法鏈代替，加上沒有隱式類型，造成了大量形式上的冗余。

Clojure 的 s 表達式，可以用幾乎任意字符來定義方法，加上宏，它可以把 Java 版的 graph dsl 封裝到非常漂亮的程度。我們先看一下對應前面那個例子的 Clojure 版本代碼：

(

這一段代碼可以說干凈漂亮很多，特別是定義邊時，可以用 `|=>` 一次串起多個節點。

實際上，`|=>` 是一個“平凡”的 Clojure 函數：

(

它看起來比較笨拙，但是把這些“笨拙”的代碼封裝成簡潔漂亮的形式，不正是高級語言的價值所在嗎？

這個函數中，其實就是對給定的節點序列，逐個判斷類型，調用對應的方法，把它們連城一個合法的 Graph DSL 線。這其中多少參考了 Akka Stream 的實現代碼。

對于剛開始學習 Clojure 的同行，可以在這里看到 loop recur 實現遞歸邏輯的 clojue 風格代碼。說起來沒有智能的尾遞歸一直算是 Clojue 的一個短板，目前標準的尾遞歸定式就是用 recur 。

我這里沒有用 scala 里的 `~>` 運算符，是因為 `~` 在 Clojure 里用于它的宏模板。如果我們在代碼中加入這個字符，我不確認寫在宏里是否會遇到麻煩，所以刻意回避了它，同理也沒有使用 `#` 和 `@` 。想來想去，`|=>` 雖然麻煩一點，但是 S 表達式的形式特點使得它處理多個節點的串接時特別方便，多寫一個字符也就可以接受了。

在這個代碼中，展示的是 from 到 to 的連接邏輯，其中 via 操作的代碼在另一個函數中：

(

對于靜態語言，這種分派反而簡單很多，不過把它封裝起來后，我們就可以找回 Lisp 語族簡潔的體驗。我針對官方 [Working with Graphs] 提到的一些例子，構造了對應的 Clojure 實現，作為 akka-stream-clojure 庫的測試代碼，例如：

(

這個測試展示的是整數序列數據經過 filter 分離成奇偶兩個數據流，在通過 zip 合成一個 pair 的邏輯，官方的 Scala 版本是：

val

Java 版本我們就不放在這里了，實在是比較冗長。當然有一部分原因是 Clojure 自帶了可以無限求值的惰性序列，這些Clojure工具庫極大的提升了它的表達能力。

有 Akka 經驗的朋友可能會知道，除了從 from 到 to 的 `~>` ，Akka Stream 還允許我們從 to 節點反向連接到 from ，這個運算符是 `<~` 。

對應的，我也實現了 `<=|` 操作符：

(

但是暫時我還沒有為它寫測試。

需要注意的一個細節就是，Graph DSL 的節點（Xxx）和它的圖形節點（XxxShape）還是有區別的，一般來說我們需要把構造出來的邏輯節點 add 到構造器（builder）中，從 add 調用的返回值得到對應的圖形節點。我提供了一些方法，封裝了幾個常用的節點構造邏輯，可以一步得到對應的圖形節點，這個將來應該還會繼續追加。這個約束也是我們一定要在 `|=>` 和其它工具方法中傳入 builder 對象的原因。在 Scala 中，因為有強大（同時也是難以駕馭的）的 implicit 功能，我們可以進一步簡化形式，把 builder “藏起來”。但是我想了很久，還是決定在 Clojure 里不做太復雜的“黑魔法”，樸素的把 builder 放在明處。考慮到 S 表達式和其它抽象帶來的代碼簡化，多寫一個 builder 還是可以接受的。雖然用宏可以模擬此類功能，但是需要做大量的工作，并且降低了代碼的擴展能力，在工程上，還是要考慮代價和收益的平衡。

暫時這個庫還只是針對 graph dsl，但是將來可能會嘗試加入更豐富的功能，這樣我們可以通過 clojure 代碼，把 java 項目變得更簡潔優雅，容易維護。但是接下來，可能我會先針對 akka-http 的 java api 做一些嘗試。

相關代碼在 MarchLiu/akka-stream-clojure

總結

以上是生活随笔為你收集整理的java宏定义_现代化的 Java （二十六）—— Akka Stream Graph的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。