進入 Reactive Streams 時代

Reactor 2.0 的開發始於 2014 年底，與 Reactive Streams 大致同時。我們熱衷於加入這項工作，並儘早採用 背壓協定來減輕我們主要的消息傳遞限制：有界容量。我們在 Reactor 2.0 中交付了首次嘗試，以使用 基於 RingBuffer 的排程器實現 Reactive Streams，並衍生出越來越流行的反應式模式：Reactive Extensions。

同時，Reactive Streams 開始獲得關注，並且整個函式庫生態系統都在討論這種轉變。常見的擔憂是什麼？實作 Reactive Streams 語義並非易事。我們觀察到越來越需要一個反應式基礎來解決消息傳遞並實作常見的串流運算子。因此，我們為 Reactor Core 建立了一個專案空間，並開始與 Spring Framework 團隊進行重點合作。

從 2.5 開始，Reactor 現在 組織成多個專案，維護分支（例如 2.0.x）保持不變。這反映在發布管理中，例如 Reactor Core 2.5 M1 是唯一可用的里程碑，其他專案將使用其獨有的版本控制來跟進。

為了支援這個新的專案模型，我們在 http://projectreactor.io 上部署了一個新的、希望更受歡迎的網站。

對 Reactive Streams 的全新協作方式

這種新的組織結構降低了參與專案活動的門檻。該專案受益於 Spring API 設計協作，以及來自 Sébastien Deleuze 和 Brian Clozel 的直接貢獻。Reactor 也歡迎新的外部貢獻者和審閱者提供協助。

• Alex Petrov 重新發明了流行的 Reactor 事件路由功能。
• Anatoly Kadyshev 致力於為 Reactor IO 打造極其高效的 Aeron Reactive Streams 橋接器。
• Ben Hale 及其團隊正在開發新的 Reactive Cloud Foundry Client API。Ben 不僅提早採用了 Reactor 2.5 Core 和 Stream，還不斷與我們一起迭代他的真實用例。
• Damien Vitrac 貢獻了新的 專案網站風格，並準備 Reactor Console 使用者體驗。
• Dávid Karnok，研究員、主要的活躍 RxJava 提交者，以及優秀的 Advanced RxJava 部落格的作者，對我們的內部運營模式產生了重大影響。我們對效率和反應式模式的共同熱情促使我們創建了一個研究空間，Reactive Streams Commons。

與 Reactor 2.5 相關的依賴關係和協作

Reactive Streams Commons

Reactive Streams Commons 儲存庫是一項開放的研究工作，專注於 Reactive Extensions 的效率，以及 Reactive Streams 規範。它完全由 Reactor Core 和 Stream 內聯，後者作為該工作所關注的許多變革的合約閘道。

因此，「RSC」是一個自由形式的專案，類似於 JCTools 對並發佇列的處理方式。其最大的進展之一是「融合」協定，用於減少反應式處理鏈中大多數同步和某些非同步階段的開銷。最後，這項工作有助於修復一百多個串流錯誤，並且我們的測試流程現在涉及 RSC 單元/整合測試和 JMH 基準測試，以及 Reactor 自己的整合測試和 基準測試。

Reactor Core 2.5.0.M1

今天的 Reactor 部落格系列文章以一個令人愉快的事件開始：Reactor Core 2.5.0.M1 的發布！在其新範圍以及與 Reactive Streams Commons 的密切聯繫下，Reactor Core 提供了足夠的 Rx 覆蓋範圍，可以構建反應式應用程式或函式庫，例如 Spring Reactive Web 支援。對於迫不及待的讀者，請查看已 在 github 上提供的快速入門。

快速瀏覽分散-收集情境

Mono.from(userRequestPublisher)
    .then(userRepository::findUserProfile, 
          userRepository::findUserPaymentMethod)
    .log("user.requests")
    .or(Mono.delay(5)
            .then(n -> Mono.error(new TimeoutException()))
    .mergeWith(userRepository::findSimilarUserDetails)
    .map(userDetailsTuple -> userDetailsTuple.t1.username)
    .publishOn(SchedulerGroup.io())
    .subscribe(responseSubscriber);

詳細資訊

• Flux，一個包含 0 到 N 個資料訊號的 Publisher，具有簡化的 Rx 範圍。運算子包括 create()、interval()、merge()、zip()、concat()、switchOnError() 和 switchOnEmpty()

• Mono，一個包含 0 或 1 個資料訊號的 Publisher，具有簡化的 Rx 衍生範圍，適用於強型別化此特定容量性質。運算子包括 delay()、then()、any()、and()、or()、otherwise()、otherwiseIfEmpty()、where() 和一個阻塞的 get()。

• 基於純 Java 介面（Runnable、Callable）的新簡化排程合約。

-- 具有 SchedulerGroup、TopicProcessor 和 WorkQueueProcessor。 -- 取代了以前的 Enviroment/Dispatcher 配對，同時滿足了相同的需求，並且很快將提供一個簡單的遷移路徑。不再有靜態狀態保存調度器的引用。 -- 連結的運算子：publishOn() 和 dispatchOn()

• 具有 TestSubscriber 的 Publisher 來源的測試支援。
• 將 Callable、Runnable、Iterable、Java 8 CompletableFuture、Java 9 Flow.Publisher、RxJava 1 Observable 和 Single 轉換為可立即用於 Reactive Streams 的 Flux 和 Mono，無需額外的橋接依賴關係。
• 完全改進並整合的 Javadoc，包括略微調整的彈珠圖。
• 一個微型工具組，包含實用程式和基本 Subscriber，可隨意重複使用以實作您自己的 Reactive 元件。

-- 一個具有成本效益的計時器 API 和實作（雜湊輪計時器）。 -- 新的 Fusion API，可實際合併來自反應式鏈的 2 個或多個階段。 -- 一個調整後的 QueueSupplier，它將為正確的容量提供正確的佇列。

• 基於 狀態 和 流程表示的新自省 API。

-- 使用回退到 java.util.logging 或 SLF4J（如果可用）的 Publisher 記錄。可以直接在具有 log() 運算子的 Flux 和 Mono 上使用。 -- 與包括 Reactive Streams 在內的任何其他合約正交，一切都可以是 Backpressurable、Completable 或 Receiver，後者生成到泛型 Object（可能是 Subscriber），這反過來又允許我們追蹤流程的完整圖形，並使用狀態指示器來擴充它：

接下來是什麼？

我們希望收集您的寶貴意見，您可以前往相關的問題追蹤儲存庫提交問題，或加入我們新建立的 Gitter 頻道。請持續關注 Reactor Stream 2.5.0.M1 的下一篇介紹，它是完整的基於 Reactive Streams 實現的 Rx。