讓我們看一下如何通過這個方法來實現 zipWithIndex吧：

實現zipWithIndex (indexed)

``` Source.from(Arrays.asList("A", "B", "C", "D"))

.statefulMap(
    () -> 0L,
    (index, element) -> Pair.create(index + 1, Pair.create(element, index)),
    indexOnComplete -> Optional.empty())


.runForeach(System.out::println, system);

// prints // Pair(A,0) // Pair(B,1) // Pair(C,2) // Pair(D,3) ````

也可以實現 zipWithNext、zipWithPreviousAndNext我們再看看如何實現較為複雜的 bufferUntilChanged吧。

實現 bufferUntilChanged

``` Source.from(Arrays.asList("A", "B", "B", "C", "C", "C", "D"))

.statefulMap(
    () -> (List<String>) new LinkedList<String>(),
    (buffer, element) -> {
      if (buffer.size() > 0 && (!buffer.get(0).equals(element))) {
        return Pair.create(
            new LinkedList<>(Collections.singletonList(element)),
            Collections.unmodifiableList(buffer));
      } else {
        buffer.add(element);
        return Pair.create(buffer, Collections.<String>emptyList());
      }
    },
    Optional::ofNullable)
.filterNot(List::isEmpty)

.runForeach(System.out::println, system);

// prints // [A] // [B, B] // [C, C, C] // [D] ````

舉一反三，如何實現 distinctUntilChanged呢 ?

實現 distinctUntilChanged

``` Source.from(Arrays.asList("A", "B", "B", "C", "C", "C", "D"))

.statefulMap(
    Optional::<String>empty,
    (lastElement, element) -> {
      if (lastElement.isPresent() && lastElement.get().equals(element)) {
        return Pair.create(lastElement, Optional.<String>empty());
      } else {
        return Pair.create(Optional.of(element), Optional.of(element));
      }
    },
    listOnComplete -> Optional.empty())
.via(Flow.flattenOptional())

.runForeach(System.out::println, system);

// prints // A // B // C // D ``` 如果要實現聚合buffer`呢？

實現 buffer

``` Source.fromJavaStream(() -> IntStream.rangeClosed(1, 10))

.statefulMap(        () -> new ArrayList<Integer>(3),        (list, element) -> {          list.add(element);          if (list.size() == 3) {            return Pair.create(new ArrayList<Integer>(3), Collections.unmodifiableList(list));          } else {            return Pair.create(list, Collections.<Integer>emptyList());          }        },        listOnComplete -> Optional.ofNullable(listOnComplete))    .filterNot(List::isEmpty)

.runForeach(System.out::println, system);// printsList(1, 2, 3)List(4, 5, 6)List(7, 8, 9)List(10)

````

更復雜的例子：處理資源

在前面看了如何實現 zipWithIndex 、bufferUntilChanged 之後，讓我們進一步看看如何優雅和安全地處理資源。在任何的程式語言和框架中，資源的處理都是非常基礎但是又很棘手的事項。在 Java 7 中首次引入了 try-with-resources 語法，對資源處理進行了一定程度的簡化，而在反應式流中，我們又應該如何的操作呢？這裡我們可以分為兩種情況：

1. 針對流中的每個元素都建立一個新的資源，使用這個資源，關閉這個資源。
2. 針對整個流建立一個資源，並在處理流中的每個元素時使用這個資源，並在流的生命週期結束後，關閉這個資源。

因為資源通常開銷較大且需要妥善管理，所以在開發過程中，我們更容易遇到的是 第2種情況，即資源的建立和銷燬和流的生命週期進行了繫結。反應式流中的資源管理，還有更多的細節需要考慮：

1. 資源的初始化和關閉需要支援併發安全；反應式流可以被多次物化，被多個下游訂閱者訂閱和處理，並且以任意的順序進行取消訂閱，需要在各種情況下（上游完成、下游取消、處理異常等）等情況下妥善的建立和銷燬資源。
2. 在流生命週期的各個階段安全地建立和銷燬資源；比如：即使在建立資源或者銷燬資源的時觸發了異常，也不會對同一個資源關閉多次。
3. 支援非同步從而提高資源使用的效率。
4. 感知流的生命週期，支援在關閉資源時提供可選的值給到下游以標識流的結束，比如處理檔案時，使用一個特殊的識別符號標識檔案的結尾。

綜合上面的這些訴求，對應的程式碼就會變得很複雜，大家可以給自己一點時間思考一下：如果是自己獨立實現類似的操作需要做出那些努力呢？而在現實的開發過程中，我們遇到的述求很多時候並非一起提出，而時隨著迭代接踵而至，那麼如果當初的程式碼編寫的不是很易於擴充套件，擁有良好的測試，則可能按下葫蘆浮起瓢。比如在 reactor-core中就有如下的using 操作符：

``` public static Mono using( Callable<? extends D> resourceSupplier, Function<? super D, ? extends Mono<? extends T>> sourceSupplier, Consumer<? super D> resourceCleanup) {...} ````

resourceSupplier 針對每個訂閱者，建立一個資源

sourceSupplier 結合建立的資源，產生對應的元素

resourceCleanup 取消訂閱或者流完成時，清理對應的資源

在 reactor-core 中，對應的底層實現為 MonoUsing 共 360 行程式碼，而要實現我們想要的邏輯，我們還需要和另一個流進行合併，即這裡的 using 類似於 unfoldResource 。那麼有沒有可能使用更加簡單的方案來進行實現呢？答案是肯定的，和前面的幾個操作符一樣，我們可以使用 statefulMap 來實現mapWithResource ，思維過程如下：

1. using / mapWithResource 的生命週期管理 和 statefulMap 的 create 和  onComplete 方法對應，針對資源，onComplete 方法可以被命名為更加貼切的 release / close / cleanUp。

2. 在流中使用的資源，我們可以認為是一個狀態，只不過這個狀態在流的整個生命週期中不再變化，一直是 create 方法中返回的 Resource 。

3. 在關閉資源時，我們可以通過返回一個 Optional<Out> 來返回一個可選的值。

4. 對併發資源的非同步處理，則可以通過返回一個 CompletionStage<Out> 而非 Out 來實現，在 using 方法中，我們返回的是一個 Mono<T>

經過上面的思維過程，我們不難得出這個流上的方法的宣告可以為：

``` public mapWithResource( Supplier<? extends R> create, BiFunction<? super R, ? super In, ? extends Out> function, Function<? super R, ? extends Optional<? extends Out> close) {...} ````

resourceSupplier 針對每個訂閱者/每次物化，建立一個資源

function 使用create 中建立的資源處理流中的每個元素

close 在流關閉的同時關閉資源，並再向下遊提供一個可選的值

具體的的實現這裡留空，感興趣的小夥伴可以結合前面的例子進行實現。下面我們看一下如何使用這個 mapWithResource 方法，從而加深大家的理解。

使用mapWithResource

假設我們有一組 SQL 需要進行處理，我們需要從資料庫中的多個表中查詢對應的結果，並將最終結果進行合併和輸出到控制檯。在mapWithResource 的幫助下，我們可以極大的簡化我們的程式碼：

``` Source.from( Arrays.asList( "SELECT * FROM shop ORDER BY article-0000 order by gmtModified desc limit 100;", "SELECT * FROM shop ORDER BY article-0001 order by gmtModified desc limit 100;")) .mapWithResource( () -> dbDriver.create(url, userName, password), (connection, query) -> db.doQuery(connection, query).toList(), connection -> { connection.close(); return Optional.empty(); }) .mapConcat(elems -> elems) .runForeach(System.out::println, system); ````

在上面的例子中：我們有一組預先定義好的 SQL，分別從多個表中讀取最新的 100 條資料，通過使用mapWithResource ，我們優雅地為每個流建立了 db 相關的連線，並進行對應的查詢操作，併合並查詢結果，在流處理完成後，關閉對應的資源。上面的程式碼通過複用我們前面編寫的 mapWithResource 將複雜資源和生命週期管理進行了簡化，作為對比，大家可以思考一下如果我們不使用已有抽象所需要付出的努力。

總結

在上面的例子中，我們通過 statefulMap以及和其他的操作符相互組合，實現了很多和狀態、生命週期相關的操作符，而程式碼量則大大減少。基於一個經過考驗的操作符來編寫自定義操作符，也能進一步降低出錯的概率，以及程式碼審查的難度。而相關的操作符都是通過一個底層的 statefulMap 來實現。對映到我們的工作中則是儘可能地抽象、提煉，對系統的核心模型、核心功能進行打磨，從而每個應用都有一個精巧的核心，並和其他的應用構成豐富的生態。而非上來就 複製、貼上，重複造輪子；避免最終陷入複製、貼上的泥潭中。雖然有時我們可能沒有足夠的時間來進一步抽象，而是業務先行。但是我依然建議，在後續的實踐中，進行不斷回顧和提煉，在保障系統穩定可靠、在有測試手段保障的情況下，進行逐步的重構，使得系統更加容易理解、維護和穩固。

筆者相信：磨刀不誤砍柴工，在設計、方案review、測試和不斷重構、精煉的過程中所花費的時間，一定會在將來多倍的回報。