您好,登錄后才能下訂單哦!
這篇文章主要介紹“響應(yīng)式編程簡介之如何掌握Reactor”,在日常操作中,相信很多人在響應(yīng)式編程簡介之如何掌握Reactor問題上存在疑惑,小編查閱了各式資料,整理出簡單好用的操作方法,希望對大家解答”響應(yīng)式編程簡介之如何掌握Reactor”的疑惑有所幫助!接下來,請跟著小編一起來學(xué)習(xí)吧!
Reactor是基于JVM的非阻塞API,他直接跟JDK8中的API相結(jié)合,比如:CompletableFuture,Stream和Duration等。
它提供了兩個非常有用的異步序列API:Flux和Mono,并且實現(xiàn)了Reactive Streams的標(biāo)準(zhǔn)。
并且還可以和reactor-netty相結(jié)合,作為一些異步框架的底層服務(wù),比如我們非常熟悉的Spring MVC 5中引入的WebFlux。
我們知道WebFlux的底層使用的是reactor-netty,而reactor-netty又引用了Reactor。所以,如果你在POM中引入了webFlux依賴:
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-webflux</artifactId> </dependency>
那么項目將會自動引入Reactor。
如果你用的不是Spring webflux,沒關(guān)系,你可以直接添加下面的依賴來使用Reactor:
<dependency> <groupId>io.projectreactor</groupId> <artifactId>reactor-core</artifactId> </dependency>
最最開始的時候微軟為.NET平臺創(chuàng)建了Reactive Extensions (Rx) library。接著RxJava實現(xiàn)了JVM平臺的Reactive。
然后Reactive Streams標(biāo)準(zhǔn)出現(xiàn)了,它定義了Java平臺必須滿足的的一些規(guī)范。并且已經(jīng)集成到JDK9中的java.util.concurrent類中。
在Flow中定義了實現(xiàn)Reactive Streams的四個非常重要的組件,分別是Publisher,Subscriber,Subscription和Processor。
一般來說reactive在面向?qū)ο蟮木幊陶Z言中是以觀察者模式的擴(kuò)展來使用的。
我們來具體看一下這個觀察者模式的實現(xiàn),以Publisher和Subscriber為例:
public static interface Publisher<T> { public void subscribe(Subscriber<? super T> subscriber); }
public static interface Subscriber<T> { public void onSubscribe(Subscription subscription); public void onNext(T item); public void onError(Throwable throwable); public void onComplete(); }
上面定義了兩個接口,Publisher和Subscriber,Publisher的作用就是subscribe到subscriber。
而subscriber定義了4個on方法,用來觸發(fā)特定的事件。
那么Publisher中的subscribe是怎么觸發(fā)Subscriber的onSubscribe事件呢?
很簡單,我們看一個具體的實現(xiàn):
public void subscribe(Flow.Subscriber<? super T> subscriber) { Subscription sub; if (throwable != null) { assert iterable == null : "non-null iterable: " + iterable; sub = new Subscription(subscriber, null, throwable); } else { assert throwable == null : "non-null exception: " + throwable; sub = new Subscription(subscriber, iterable.iterator(), null); } subscriber.onSubscribe(sub); if (throwable != null) { sub.pullScheduler.runOrSchedule(); } }
上面的例子是PullPublisher的subscribe實現(xiàn)。我們可以看到,在這個subscribe中觸發(fā)了subscriber.onSubscribe方法。而這就是觀察者模式的秘密。
或者說,當(dāng)Publisher調(diào)用subscribe的時候,是主動push subscriber的onSubscribe方法。
熟悉Iterable-Iterator模式的朋友應(yīng)該都知道,Iterator模式,其實是一個主動的pull模式,因為需要不斷的去調(diào)用next()方法。所以它的控制權(quán)是在調(diào)用方。
在現(xiàn)代應(yīng)用程序中,隨著用戶量的增多,程序員需要考慮怎么才能提升系統(tǒng)的處理能力。
傳統(tǒng)的block IO的方式,因為需要占用大量的資源,所以是不適合這樣的場景的。我們需要的是NO-block IO。
JDK中提供了兩種異步編程的模型:
第一種是Callbacks,異步方法可以通過傳入一個Callback參數(shù)的形式來在Callback中執(zhí)行異步任務(wù)。比較典型的像是java Swing中的EventListener。
第二中就是使用Future了。我們使用Callable來提交一個任務(wù),然后通過Future來拿到它的運行結(jié)果。
這兩種異步編程會有什么問題呢?
callback的問題就在于回調(diào)地獄。熟悉JS的朋友應(yīng)該很理解這個回調(diào)地獄的概念。
簡單點講,回調(diào)地獄就是在callback中又使用了callback,從而造成了這種callback的層級調(diào)用關(guān)系。
而Future主要是對一個異步執(zhí)行的結(jié)果進(jìn)行獲取,它的 get()實際上是一個block操作。并且不支持異常處理,也不支持延遲計算。
當(dāng)有多個Future的組合應(yīng)該怎么處理呢?JDK8 實際上引入了一個CompletableFuture類,這個類是Future也是一個CompletionStage,CompletableFuture支持then的級聯(lián)操作。不過CompletableFuture提供的方法不是那么的豐富,可能滿足不了我的需求。
于是我們的Reactor來了。
Reactor提供了兩個非常有用的操作,他們是 Flux 和 Mono。 其中Flux 代表的是 0 to N 個響應(yīng)式序列,而Mono代表的是0或者1個響應(yīng)式序列。
我們看一個Flux是怎么transfer items的:
先看下Flux的定義:
public abstract class Flux<T> implements Publisher<T>
可以看到Flux其實就是一個Publisher,用來產(chǎn)生異步序列。
Flux提供了非常多的有用的方法,來處理這些序列,并且提供了completion和error的信號通知。
相應(yīng)的會去調(diào)用Subscriber的onNext, onComplete, 和 onError 方法。
我們看下Mono是怎么transfer items的:
看下Mono的定義:
public abstract class Mono<T> implements Publisher<T>
Mono和Flux一樣,也是一個Publisher,用來產(chǎn)生異步序列。
Mono因為只有0或者1個序列,所以只會觸發(fā)Subscriber的onComplete和onError方法,沒有onNext。
另一方面,Mono其實可以看做Flux的子集,只包含F(xiàn)lux的部分功能。
Mono和Flux是可以互相轉(zhuǎn)換的,比如Mono#concatWith(Publisher)返回一個Flux,而 Mono#then(Mono)返回一個Mono.
我們看下Flux創(chuàng)建的例子:
Flux<String> seq1 = Flux.just("foo", "bar", "foobar"); List<String> iterable = Arrays.asList("foo", "bar", "foobar"); Flux<String> seq2 = Flux.fromIterable(iterable); Flux<Integer> numbersFromFiveToSeven = Flux.range(5, 3);
可以看到Flux提供了很多種創(chuàng)建的方式,我們可以自由選擇。
再看看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);
subscribe可以一個參數(shù)都沒有,也可以多達(dá)4個參數(shù)。
看下沒有參數(shù)的情況:
Flux<Integer> numbersFromFiveToSeven = Flux.range(5, 3); numbersFromFiveToSeven.subscribe();
注意,沒有參數(shù)并不表示Flux的對象不被消費,只是不可見而已。
看下帶參數(shù)的情況:consumer用來處理on each事件,errorConsumer用來處理on error事件,completeConsumer用來處理on complete事件,subscriptionConsumer用來處理on subscribe事件。
前面的3個參數(shù)很好理解,我們來舉個例子:
Flux<Integer> ints3 = Flux.range(1, 4); ints3.subscribe(System.out::println, error -> System.err.println("Error " + error), () -> System.out.println("Done"), sub -> sub.request(2));
我們構(gòu)建了從1到4的四個整數(shù)的Flux,on each就是打印出來,如果中間有錯誤的話,就輸出Error,全部完成就輸出Done。
那么最后一個subscriptionConsumer是做什么用的呢?
subscriptionConsumer accept的是一個Subscription對象,我們看下Subscription的定義:
public interface Subscription { public void request(long n); public void cancel(); }
Subscription 定義了兩個方法,用來做初始化用的,我們可以調(diào)用request(n)來決定這次subscribe獲取元素的最大數(shù)目。
比如上面我們的例子中,雖然構(gòu)建了4個整數(shù),但是最終輸出的只有2個。
上面所有的subscribe方法,都會返回一個Disposable對象,我們可以通過Disposable對象的dispose()方法,來取消這個subscribe。
Disposable只定義了兩個方法:
public interface Disposable { void dispose(); default boolean isDisposed() { return false; }
dispose的原理是向Flux 或者 Mono發(fā)出一個停止產(chǎn)生新對象的信號,但是并不能保證對象產(chǎn)生馬上停止。
有了Disposable,當(dāng)然要介紹它的工具類Disposables。
Disposables.swap() 可以創(chuàng)建一個Disposable,用來替換或者取消一個現(xiàn)有的Disposable。
Disposables.composite(…)可以將多個Disposable合并起來,在后面統(tǒng)一做處理。
到此,關(guān)于“響應(yīng)式編程簡介之如何掌握Reactor”的學(xué)習(xí)就結(jié)束了,希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學(xué)習(xí),快去試試吧!若想繼續(xù)學(xué)習(xí)更多相關(guān)知識,請繼續(xù)關(guān)注億速云網(wǎng)站,小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進(jìn)行舉報,并提供相關(guān)證據(jù),一經(jīng)查實,將立刻刪除涉嫌侵權(quán)內(nèi)容。