日本综合一区二区|亚洲中文天堂综合|日韩欧美自拍一区|男女精品天堂一区|欧美自拍第6页亚洲成人精品一区|亚洲黄色天堂一区二区成人|超碰91偷拍第一页|日韩av夜夜嗨中文字幕|久久蜜综合视频官网|精美人妻一区二区三区

RELATEED CONSULTING
相關(guān)咨詢
選擇下列產(chǎn)品馬上在線溝通
服務(wù)時(shí)間：8:30-17:00
你可能遇到了下面的問題
關(guān)閉右側(cè)工具欄

新聞中心

這里有您想知道的互聯(lián)網(wǎng)營銷解決方案
Java響應(yīng)式編程實(shí)踐與原理解析

背景

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

創(chuàng)新互聯(lián)建站是一家專注于成都網(wǎng)站設(shè)計(jì)、網(wǎng)站建設(shè)與策劃設(shè)計(jì),東勝網(wǎng)站建設(shè)哪家好?創(chuàng)新互聯(lián)建站做網(wǎng)站,專注于網(wǎng)站建設(shè)十載,網(wǎng)設(shè)計(jì)領(lǐng)域的專業(yè)建站公司;建站業(yè)務(wù)涵蓋:東勝等地區(qū)。東勝做網(wǎng)站價(jià)格咨詢:028-86922220

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

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

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

特點(diǎn)

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

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

實(shí)現(xiàn)

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

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

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

示例

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

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

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

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

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

Mono mono1 = Mono.just("one");

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

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

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

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

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

Flux 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的底層實(shí)現(xiàn)邏輯和編程思想，我們下面會給大家詳細(xì)的演示下flux.create方法的執(zhí)行。尤其是前面提到的直到訂閱才會發(fā)生任何事情，這句話的真實(shí)含義。

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

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

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

  • Flux.create方法接受一個(gè)函數(shù)式接口Consumer作為輸入?yún)?shù)，在我們這個(gè)例子中就是。
(t) -> {
 t.next("create");
 t.next("create1");
}，
public static  Flux create(Consumer> emitter) {
return create(emitter, OverflowStrategy.BUFFER);
}

  • 我們一路追蹤下去，發(fā)現(xiàn)它把我們的函數(shù)式接口賦值給了Fluxcreate對象的一個(gè)屬性source,然后就返回了。并沒有執(zhí)行這個(gè)函數(shù)式接口的邏輯。
FluxCreate(Consumer> source, OverflowStrategy backpressure, FluxCreate.CreateMode createMode) {
 this.source = (Consumer)Objects.requireNonNull(source, "source");
 this.backpressure = (OverflowStrategy)Objects.requireNonNull(backpressure, "backpressure");
 this.createMode = createMode;
}

  • 那么什么時(shí)候執(zhí)行我們的代碼邏輯呢，接著向下看。subscribe方法也是接收了一個(gè)函數(shù)式接口。
(st->{
 System.out.println(st);
}）
public final Disposable subscribe(Consumer 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í)行代碼，而這個(gè)里面的每次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


文章標(biāo)題：Java響應(yīng)式編程實(shí)踐與原理解析
網(wǎng)站URL：http://www.dlmjj.cn/article/cosojsc.html