背景

在傳統(tǒng)的命令式編程模式下，程序都是按照人工編寫的指令一條條順序的同步執(zhí)行，也就是說，只有當前指令運行完畢，下一條指令才開始執(zhí)行。那么傳統(tǒng)的命令式編程有有些線程處理模型呢?

首先是同步阻塞式，在這種模型下，只有阻塞操作完成后，程序才能夠繼續(xù)執(zhí)行。而且阻塞會浪費資源，比如等待網(wǎng)絡(luò)連接(數(shù)據(jù)庫請求，其他服務(wù)請求)，就會導致執(zhí)行線程處于空閑狀態(tài)。

第二種就是異步阻塞式，在這種方式下一般會通過線程池，創(chuàng)建很多線程，然后針對請求，分配空閑的線程來處理。每個處理線程當遇到阻塞操作時，還是會中斷等待操作完成，不過相對于同步阻塞的模式，減少了任務(wù)的響應(yīng)時間。通過增加并行度，提升了資源利用率。

第三種是異步非阻塞，通過回調(diào)方法來摒棄阻塞操作帶來的資源浪費。不過回調(diào)函數(shù)會層層嵌套，導致回調(diào)噩夢(callback hell)，讓可讀性變得很差。

為了利用第三種模型的優(yōu)勢，同時又讓代碼維護性更高，spring社區(qū)推出了spring flux響應(yīng)式非阻塞編程。它默認的實現(xiàn)叫projectreactor。projectreactor是JVM的完全非阻塞響應(yīng)式編程基礎(chǔ)，具有高效的需求管理(以管理“背壓”的形式).它提供了可組合的異步序列API Flux(用于[0…N]元素)和 Mono(用于[0 | 1]元素)，廣泛地實現(xiàn)了Reactive Extensions規(guī)范。

特點

響應(yīng)式編程的特點包括以下幾點。待會會通過例子給大家詳細展示下。

• 可組合性&可讀性
• 直到訂閱才會發(fā)生任何事情
• 采用背壓或通過其他方式消費者向生產(chǎn)者發(fā)出排放率過高的信號的能力
• 具體豐富的數(shù)據(jù)流運算符
• 高水平但高價值的抽象是并發(fā)性不可知的

實現(xiàn)

projectreactor引入了可組合的反應(yīng)類型，它們實現(xiàn)Publisher同時也提供了豐富的操作符，尤其是 Flux 和Mono 。

Flux 表示一個0..N項的反應(yīng)序列，可以有 完成信號、錯誤信息來結(jié)束整個流程。所以傳輸?shù)臄?shù)據(jù)為一個普通值、一個完成信號、一個錯誤信號。對應(yīng)的方法為onNext()、onComplete()、onError()。

而一個Mono對象表示一個單值或空的(0..1)結(jié)果，可以認為是一種特殊的 Flux，最多可以發(fā)出一個普通值，同樣包含onComplete()、onError()。

示例

• 靜態(tài)數(shù)據(jù)創(chuàng)建

直接調(diào)用just()方法進行創(chuàng)建，也可以通過一個Stream或者一個Iterable對象(比如List)。還有通過Flux靜態(tài)方法來生成，range方法(這個方法生成的是一個 Integer 序列，第一個參數(shù)表示起始數(shù)字，第二個參數(shù)表示，生成的個數(shù)，這里生成的數(shù)據(jù)就為1、2、3)，empty() 方法就是生成一個空的序列。

Flux<String> flux1 = Flux.just("one", "two", "three");
Flux<String> flux2 = Flux.fromStream(Stream.of("one", "two", "three"));
List<String> iterable = Arrays.asList("one", "two", "three");
Flux<String> flux3 = Flux.fromIterable(iterable);
Flux<Integer> flux4 = Flux.range(1, 3);
//或者通過 #empty() 生成空數(shù)據(jù)
Flux<String> fluxEmpty = Flux.empty()

Mono 也有類似的創(chuàng)建方法，只是對于的 just() 方法是對應(yīng)只是一個參數(shù)。而 justOrEmpty() 方法會對空值進行校驗，選擇調(diào)用 just() 或者 empty()。

//Mono 也是類型
Mono<String> monoEmpty = Mono.empty();

Mono<String> mono1 = Mono.just("one");

//justOrEmpty 可以保證傳入?yún)?shù)為空時也不會報錯
Mono<String> mono2 = Mono.justOrEmpty(null);

• 動態(tài)數(shù)據(jù)創(chuàng)建

動態(tài)數(shù)據(jù)創(chuàng)建方法主要有g(shù)enerate與create兩種方法。

對于generate 方法，在Flux中有3個重載方法，不管是哪個方法都是會包含一個循環(huán)構(gòu)造函數(shù)。在每個循環(huán)中，sink.next()方法最多被調(diào)用一次。比如在 flux_generate1 這個實例對應(yīng)的方法。循環(huán)生成1~10的序列，當atomicInteger大于10的時候就調(diào)用complete()方法，發(fā)出信息通知訂閱者。flux_generate2 實例對應(yīng)的方法則將atomicInteger作為一個對象，在方法中進行傳遞，并且在最后打印在控制臺上。

// generate 生成，調(diào)用 next 即生成數(shù)據(jù)，complete 則是完成了整個流程
// 一個循環(huán)中只允許調(diào)用 next 方式一次
AtomicInteger atomicInteger = new AtomicInteger();
Flux<Integer> flux_generate1 = Flux.generate(sink -> {
if(atomicInteger.incrementAndGet() > 10){
sink.complete();
}
sink.next(atomicInteger.get());
});

Flux<Integer> flux_generate2 = Flux.generate(() -> 0, (integer, sink) -> {
if (++integer > 10) {
sink.complete();
}
sink.next(integer);
return integer;
}, integer -> {
System.out.println("last integer value is " + integer);
});

執(zhí)行過程解析

為了更好的理解flux的底層實現(xiàn)邏輯和編程思想，我們下面會給大家詳細的演示下flux.create方法的執(zhí)行。尤其是前面提到的直到訂閱才會發(fā)生任何事情，這句話的真實含義。

flux.create((t) -> {
 t.next("create");
 t.next("create1");

}).subscribe(st->{
 System.out.println(st);
});

上面是我們要執(zhí)行的一段代碼。通過debug我們可以看到如下的執(zhí)行過程。

• Flux.create方法接受一個函數(shù)式接口Consumer作為輸入?yún)?shù)，在我們這個例子中就是。

(t) -> {
 t.next("create");
 t.next("create1");
}，
public static <T> Flux<T> create(Consumer<? super FluxSink<T>> emitter) {
return create(emitter, OverflowStrategy.BUFFER);
}

• 我們一路追蹤下去，發(fā)現(xiàn)它把我們的函數(shù)式接口賦值給了Fluxcreate對象的一個屬性source,然后就返回了。并沒有執(zhí)行這個函數(shù)式接口的邏輯。

FluxCreate(Consumer<? super FluxSink<T>> source, OverflowStrategy backpressure, FluxCreate.CreateMode createMode) {
 this.source = (Consumer)Objects.requireNonNull(source, "source");
 this.backpressure = (OverflowStrategy)Objects.requireNonNull(backpressure, "backpressure");
 this.createMode = createMode;
}

• 那么什么時候執(zhí)行我們的代碼邏輯呢，接著向下看。subscribe方法也是接收了一個函數(shù)式接口。

(st->{
 System.out.println(st);
}）
public final Disposable subscribe(Consumer<? super T> consumer) {
Objects.requireNonNull(consumer, "consumer");
return this.subscribe(consumer, (Consumer)null, (Runnable)null);
}

• 下面我們看看調(diào)用subscribe后發(fā)生了什么。

調(diào)用subscribe函數(shù)之1

調(diào)用subscribe函數(shù)之2

調(diào)用subscribe函數(shù)之3

沒錯就是通過subscribe出發(fā)了Flux.create里面的執(zhí)行代碼，而這個里面的每次next調(diào)用，又觸發(fā)了后面的subscriber的執(zhí)行，最終將結(jié)果打印出來。

Connected to the target VM, address: '127.0.0.1:53984', transport: 'socket'
create
create1
Disconnected from the target VM, address: '127.0.0.1:53984', transport: 'socket'
Process finished with exit code 0