簡介
上篇文章我們簡單的介紹了Reactor的發展史和基本的Flux和Mono的使用，本文将會進一步挖掘Reactor的進階用法，一起來看看吧。
自定義Subscriber
之前的文章我們提到了4個Flux的subscribe的方法：
Disposable subscribe();
Disposable subscribe(Consumer<? super T> consumer);
Disposable subscribe(Consumer<? super T> consumer,
Consumer<? super Throwable> errorConsumer);
Disposable subscribe(Consumer<? super T> consumer,
Consumer<? super Throwable> errorConsumer,
Runnable completeConsumer);
Disposable subscribe(Consumer<? super T> consumer,
Consumer<? super Throwable> errorConsumer,
Runnable completeConsumer,
Consumer<? super Subscription> subscriptionConsumer); 這四個方法，需要我們使用lambda表達式來自定義consumer，errorConsumer,completeSonsumer和subscriptionConsumer這四個Consumer。
寫起來比較複雜，看起來也不太友善，我們考慮一下，這四個Consumer是不是和Subscriber接口中定義的4個方法是一一對應的呢？
public static interface Subscriber<T> {
public void onSubscribe(Subscription subscription);
public void onNext(T item);
public void onError(Throwable throwable);
public void onComplete();
} 對的，是以我們有一個更加簡單點的subscribe方法：
public final void subscribe(Subscriber<? super T> actual) 這個subscribe方法直接接收一個Subscriber類。進而實作了所有的功能。
自己寫Subscriber太麻煩了，Reactor為我們提供了一個BaseSubscriber的類，它實作了Subscriber中的所有功能，還附帶了一些其他的方法。
我們看下BaseSubscriber的定義：
public abstract class BaseSubscriber<T> implements CoreSubscriber<T>, Subscription,
Disposable 注意，BaseSubscriber是單次使用的，這就意味着，如果它首先subscription到Publisher1，然後subscription到Publisher2，那麼将會取消對第一個Publisher的訂閱。
因為BaseSubscriber是一個抽象類，是以我們需要繼承它，并且重寫我們需要自己實作的方法。
下面看一個自定義的Subscriber：
public class CustSubscriber<T> extends BaseSubscriber<T> {
public void hookOnSubscribe(Subscription subscription) {
System.out.println("Subscribed");
request(1);
}
public void hookOnNext(T value) {
System.out.println(value);
request(1);
}
} BaseSubscriber中有很多以hook開頭的方法，這些方法都是我們可以重寫的，而Subscriber原生定義的on開頭的方法，在BaseSubscriber中都是final的，都是不能重寫的。
我們看一個定義：
@Override
public final void onSubscribe(Subscription s) {
if (Operators.setOnce(S, this, s)) {
try {
hookOnSubscribe(s);
}
catch (Throwable throwable) {
onError(Operators.onOperatorError(s, throwable, currentContext()));
}
}
} 可以看到，它内部實際上調用了hook的方法。
上面的CustSubscriber中，我們重寫了兩個方法，一個是hookOnSubscribe，在建立訂閱的時候調用，一個是hookOnNext，在收到onNext信号的時候調用。
在這些方法中，給了我們足夠的自定義空間，上面的例子中我們調用了request(1)，表示再請求一個元素。
其他的hook方法還有： hookOnComplete, hookOnError, hookOnCancel 和 hookFinally。
Backpressure處理
我們之前講過了，reactive stream的最大特征就是可以處理Backpressure。
什麼是Backpressure呢？就是當consumer處理過不來的時候，可以通知producer來減少生産速度。
我們看下BaseSubscriber中預設的hookOnSubscribe實作：
protected void hookOnSubscribe(Subscription subscription){
subscription.request(Long.MAX_VALUE);
} 可以看到預設是request無限數目的值。 也就是說預設情況下沒有Backpressure。
通過重寫hookOnSubscribe方法，我們可以自定義處理速度。
除了request之外，我們還可以在publisher中限制subscriber的速度。
public final Flux<T> limitRate(int prefetchRate) {
return onAssembly(this.publishOn(Schedulers.immediate(), prefetchRate));
} 在Flux中，我們有一個limitRate方法，可以設定publisher的速度。
比如subscriber request(100),然後我們設定limitRate(10),那麼最多producer一次隻會産生10個元素。
建立Flux
接下來，我們要講解一下怎麼建立Flux，通常來講有4種方法來建立Flux。
使用generate
第一種方法就是最簡單的同步建立的generate.
先看一個例子：
public void useGenerate(){
Flux<String> flux = Flux.generate(
() -> 0,
(state, sink) -> {
sink.next("3 x " + state + " = " + 3*state);
if (state == 10) sink.complete();
return state + 1;
});
flux.subscribe(System.out::println);
} 輸出結果：
3 x 0 = 0
3 x 1 = 3
3 x 2 = 6
3 x 3 = 9
3 x 4 = 12
3 x 5 = 15
3 x 6 = 18
3 x 7 = 21
3 x 8 = 24
3 x 9 = 27
3 x 10 = 30 上面的例子中，我們使用generate方法來同步的生成元素。
generate接收兩個參數：
public static <T, S> Flux<T> generate(Callable<S> stateSupplier, BiFunction<S, SynchronousSink<T>, S> generator) 第一個參數是stateSupplier，用來指定初始化的狀态。
第二個參數是一個generator，用來消費SynchronousSink，并生成新的狀态。
上面的例子中，我們每次将state+1，一直加到10。
然後使用subscribe來将所有的生成元素輸出。
使用create
Flux也提供了一個create方法來建立Flux，create可以是同步也可以是異步的，并且支援多線程操作。
因為create沒有初始的state狀态，是以可以用在多線程中。
create的一個非常有用的地方就是可以将第三方的異步API和Flux關聯起來，舉個例子，我們有一個自定義的EventProcessor，當處理相應的事件的時候，會去調用注冊到Processor中的listener的一些方法。
interface MyEventListener<T> {
void onDataChunk(List<T> chunk);
void processComplete();
} 我們怎麼把這個Listener的響應行為和Flux關聯起來呢？
public void useCreate(){
EventProcessor myEventProcessor = new EventProcessor();
Flux<String> bridge = Flux.create(sink -> {
myEventProcessor.register(
new MyEventListener<String>() {
public void onDataChunk(List<String> chunk) {
for(String s : chunk) {
sink.next(s);
}
}
public void processComplete() {
sink.complete();
}
});
});
} 使用create就夠了，create接收一個consumer參數：
public static <T> Flux<T> create(Consumer<? super FluxSink<T>> emitter) 這個consumer的本質是去消費FluxSink對象。
上面的例子在MyEventListener的事件中對FluxSink對象進行消費。
使用push
push和create一樣，也支援異步操作，但是同時隻能有一個線程來調用next, complete 或者 error方法，是以它是單線程的。
使用Handle
Handle和上面的三個方法不同，它是一個執行個體方法。
它和generate很類似，也是消費SynchronousSink對象。
Flux<R> handle(BiConsumer<T, SynchronousSink<R>>); 不同的是它的參數是一個BiConsumer，是沒有傳回值的。
看一個使用的例子：
public void useHandle(){
Flux<String> alphabet = Flux.just(-1, 30, 13, 9, 20)
.handle((i, sink) -> {
String letter = alphabet(i);
if (letter != null)
sink.next(letter);
});
alphabet.subscribe(System.out::println);
}
public String alphabet(int letterNumber) {
if (letterNumber < 1 || letterNumber > 26) {
return null;
}
int letterIndexAscii = 'A' + letterNumber - 1;
return "" + (char) letterIndexAscii;
} 本文的例子
learn-reactive本文作者：flydean程式那些事
本文連結：
http://www.flydean.com/reactor-core-in-depth/本文來源：flydean的部落格
歡迎關注我的公衆号:「程式那些事」最通俗的解讀，最深刻的幹貨，最簡潔的教程，衆多你不知道的小技巧等你來發現！
![手機軟體抓包工具及其使用方法[圖]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)