Reactor 介紹

        Reactor 是一個基礎庫，可用它構建時效性流式資料應用，或者有低延遲和容錯性要求的微/納/皮級服務。

        簡單說，Reactor 是一個輕量級 JVM 基礎庫，幫助你的服務或應用高效，非同步地傳遞訊息。

"高效"是指什麼?

• 訊息從A傳遞到B時，產生很少的記憶體垃圾，甚至不產生。
• 解決消費者處理訊息的效率低於生產者時帶來的溢位問題。
• 儘可能提供非阻塞非同步流。

        非同步程式設計很難，而像 JVM 這類提供眾多可選引數的平臺則尤其困難。 Reactor 旨在幫助大多數用例真正非阻塞地執行。我們提供的 API 比 JDK 的 java.util.concurrent

• 阻塞等待：如 Future.get()

• 不安全的資料訪問：如 ReentrantLock.lock()

• 異常冒泡：如 try…​catch…​finally

• 同步阻塞：如 synchronized{ }

• Wrapper分配(GC 壓力)：如 new Wrapper<T>(event)

  當訊息傳遞效率成為系統性能瓶頸的時候(10k msg/s，100k msg/s，1M...)，非阻塞機制就顯得尤為重要。 雖然這個有理論支援 (參見 Amdahl’s Law

  )，但讀起來太無聊了。我們舉例說明，比如你用了個 Executor 方法：

  private ExecutorService  threadPool = Executors.newFixedThreadPool(8);
  
  final List<T> batches = new ArrayList<T>();
  
  Callable<T> t = new Callable<T>() { // *1
  
          public T run() {
                  synchronized(batches) { // *2
                          T result = callDatabase(msg); // *3
                          batches.add(result);
                          return result;
                  }
          }
  };
  
  Future<T> f = threadPool.submit(t); // *4
  T result = f.get() // *5
   
  

• 1.Callable 分配 -- 可能導致 GC 壓力。
• 2.同步過程強制每個執行緒執行停-檢查操作。
• 3.訊息的消費可能比生產慢。
• 4.使用執行緒池(ThreadPool)將任務傳遞給目標執行緒 -- 通過 FutureTask 方式肯定會產生 GC 壓力。
• 5.阻塞直至 callDatabase() 回撥。

在這個簡單的例子中，很容易指出為什麼擴容是很有限的：

• 分配物件可能產生GC壓力，特別是當任務執行時間過長。

  • 每次 GC 暫停都會影響全域性效能。

• 預設，佇列是無界的，任務會因為資料庫呼叫而堆積。

  • 積壓雖然不會直接導致記憶體洩漏，但會帶來嚴重副作用：GC 暫停時要掃描更多的物件；有丟失重要資料位的風險；等等 …

  • 典型鏈式佇列節點分配時會產生大量記憶體壓力。

• 阻塞回調容易產生惡性迴圈。

  • 阻塞回調會降低訊息生產者的效率。在實踐中，任務提交後需要等待結果返回，此時流式過程幾乎演變為同步的了。

  • 會話過程丟擲的任何帶資料儲存的異常都會以不受控的方式被傳遞給生產者，否定了任何通常線上程邊界附近可用的容錯性。

• 要實現完全非阻塞是很難辦到的，尤其是在有著類似微服務架構這樣時髦綽號的分散式系統的世界裡。因此 Reactor 做了部分妥協，嘗試利用最優的可用模式，使開發者覺得他們是在寫非同步納米服務，而不是什麼數學論文。

  沒有什麼傳播得比光快(除了緋聞和網紅貓的視訊)，正如到了某個階段，延遲是每一個系統到都要面對的實實在在的問題。為此：

  Reactor 提供的框架可以幫助減輕應用中由延遲產生的副作用，只需要增加一點點開銷：

  • 使用了一些聰明的結構，通過啟動預分配策略解決執行時分配問題；

  • 通過確定資訊傳遞主結構的邊界，避免任務的無限堆疊；

  • 採用主流的響應與事件驅動構架模式，提供包含反饋在內的非阻塞端對端流；

  • 引入新的 Reactive Streams標準，拒絕超過當前容量請求，從而保證限制結構的有效性；

  • 在IPC上也使用了類似理念，提供對流控制友好的非阻塞 IO 驅動；

  • 開放了幫助開發者們以零副作用方式組織他們程式碼的函式介面，藉助這些函式來處理容錯性和執行緒安全。

Reactor版本與依賴

        <!-- reactor -->
        <!-- https://mvnrepository.com/artifact/io.projectreactor/reactor-core -->
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-core</artifactId>
            <version>2.0.8.RELEASE</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/io.projectreactor/reactor-stream -->
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-stream</artifactId>
            <version>2.0.8.RELEASE</version>
        </dependency>

個人程式碼

package com.brillilab.test;

import org.junit.Test;
import reactor.Environment;
import reactor.core.Dispatcher;
import reactor.core.DispatcherSupplier;
import reactor.core.processor.RingBufferProcessor;
import reactor.core.support.Assert;
import reactor.rx.Stream;
import reactor.rx.Streams;

import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

public class ReactorTest {

    /**
     * Function 包 的使用
     */
    @Test
    public void functionTest(){

        Consumer<String> consumer = value ->{
            System.out.println(value);
        };

        Supplier<Integer> supplier = () -> 123;

        BiConsumer<Consumer<String> , String> biConsumer = (callback ,value) -> {
            for (int i=0; i<10; i++){
                callback.accept(value);
            }
        };

        Function<Integer,String> transformation = integer -> ""+integer;

        biConsumer.accept(
                consumer,
                transformation.apply(
                        supplier.get()
                )
        );
    }

    /**
     *  Reactor 庫的Environment 和 Dispatcher
     */
    @Test
    public void EnvironmentAndDispatcher (){

        Environment.initialize();

        DispatcherSupplier supplier=Environment.newCachedDispatchers(2);

        Dispatcher d1=supplier.get();
        Dispatcher d2=supplier.get();
        Dispatcher d3=supplier.get();
        Dispatcher d4=supplier.get();

        Assert.isTrue(d1 == d3 && d2 == d4);
        supplier.shutdown();

        // 建立並註冊帶3個排程者的新池
        DispatcherSupplier supplier1 = Environment.newCachedDispatchers(3,"myPool");
        DispatcherSupplier supplier2 = Environment.cachedDispatchers("myPool");

        Assert.isTrue( supplier1 == supplier2 );
        supplier1.shutdown();
    }

    /**
     * Reactor 庫 Steams 使用 進行流式程式設計
     */
    @Test
    public void ReactorStreamTest(){

        Environment.initialize();

        Stream<String> st=Streams.just("Hello","World","!");

        st.dispatchOn(Environment.cachedDispatcher())
                .map(String::toUpperCase)
                .consume(s -> System.out.printf("%s greeting = %s%n", Thread.currentThread(), s));

    }

    /**
     * Reactor 流式程式設計之流合併 wrap concat 方法的使用
     */
    @Test
    public void ReactorStreamTest2(){

        RingBufferProcessor<String> processor=RingBufferProcessor.create();

        Stream<String> st1=Streams.just("hello ");
        Stream<String> st2=Streams.just("world ");

        //一個針對傳入 Publisher.subscribe(Subscriber<T>) 引數的下發 Stream。
        // 只支援正確使用 Reactive Stream 協議的格式正確的 Publisher：
        //onSubscribe > onNext* > (onError | onComplete)
        Stream<String> st3=Streams.wrap(processor);

        //如果一個 Publisher<T> 已經發送了，在處理下一個等待處理的 Publisher<T> 之前要等待這個 onComplete()。
        //其名稱就暗示它對於串聯不同的資料來源並保持順序正確，這些方面的作用。
        Streams.concat(st1,st2,st3)
                .reduce((prev,next) -> prev + next)
                .consume(s -> System.out.printf("%s greeting = %s%n", Thread.currentThread(), s));

        processor.onNext("!");
        processor.onNext("!!!");
        processor.onComplete();
    }
}

================================================================================================