1小時(shí)讓你掌握響應(yīng)式編程，并入門(mén)Reactor
我看同步阻塞
“你知道什么是同步阻塞嗎”，當(dāng)然知道了。“那你怎么看它呢”，這個(gè)。。。
在同步阻塞的世界里，代碼執(zhí)行到哪里，數(shù)據(jù)就跟到哪里。如果數(shù)據(jù)很慢跟不上來(lái)，代碼就停在那里等待數(shù)據(jù)的到來(lái)，然后再帶著數(shù)據(jù)一起往下執(zhí)行。
可以說(shuō)是，代碼執(zhí)行和數(shù)據(jù)是結(jié)伴而行，不離不棄。執(zhí)子之手與子偕老。讓人老感動(dòng)了。
如果還不太理解的話，可以認(rèn)為代碼執(zhí)行其實(shí)就是一些行為動(dòng)作，這些行為動(dòng)作的目的就是為了獲取/操作數(shù)據(jù)。
例如加法，這里的行為動(dòng)作就是執(zhí)行相加，數(shù)據(jù)就是加數(shù)和被加數(shù)。操作結(jié)果就是得到了另一個(gè)數(shù)據(jù)，即兩個(gè)數(shù)的和。
只是在這個(gè)加法里，數(shù)據(jù)跑的特別快，(CPU的寄存器，能不快嗎)，我們幾乎覺(jué)察不到執(zhí)行動(dòng)作在等數(shù)據(jù)的過(guò)程。怎么辦呢，那就看一個(gè)能把它們拉開(kāi)的例子。
那自然非數(shù)據(jù)庫(kù)查詢莫屬了，既有網(wǎng)絡(luò)I/O，又有磁盤(pán)I/O，肯定會(huì)慢一些。
假設(shè)我的業(yè)務(wù)是這樣的，代碼先去數(shù)據(jù)庫(kù)查詢一個(gè)用戶，接著修改用戶的密碼，然后再更新回?cái)?shù)據(jù)庫(kù)，最后代碼返回成功。
如果網(wǎng)速和數(shù)據(jù)庫(kù)都很慢的話，可能是這樣的。代碼執(zhí)行一個(gè)查詢數(shù)據(jù)庫(kù)動(dòng)作，然后等啊等啊等，等的花都謝了，終于數(shù)據(jù)庫(kù)把用戶返回過(guò)來(lái)了，接著，代碼飛快的修改了密碼，并執(zhí)行一個(gè)更新數(shù)據(jù)庫(kù)的動(dòng)作，然后又是等啊等啊等，等的花又開(kāi)了，數(shù)據(jù)庫(kù)終于回話了，更新成功。然后代碼返回成功，全部執(zhí)行完了。
所以同步阻塞代碼的最大特點(diǎn)就是，帶著數(shù)據(jù)上路，數(shù)據(jù)不到位就阻塞住。
最后來(lái)個(gè)小小的升華：
- 所謂同步就是快的等慢的，然后一起往前走，表示的是目的。
- 所謂阻塞就是想辦法讓快的停滯不前，等待慢的到來(lái)，表示的是手段。
一言以蔽之，同步是目的，阻塞是手段。
我看異步非阻塞
“你知道什么是異步非阻塞嗎”，當(dāng)然知道了，不過(guò)我不知道該怎么看它。“哦，恭喜你都會(huì)搶答了。。。”。
我們生活在異步的世界，卻是最不懂異步的人。
你去飯店吃飯，服務(wù)員把你的菜單寫(xiě)好，交給廚房后就去服務(wù)別人了。
廚房把飯做好后，通過(guò)按鈴?fù)ㄖ?wù)員，服務(wù)員再把飯送到你的位置上。
服務(wù)員是主(或I/O)線程，把任務(wù)交給廚房這個(gè)工作線程去執(zhí)行，廚房接到任務(wù)的同時(shí)還要記住送來(lái)該任務(wù)的服務(wù)員，然后廚房去執(zhí)行任務(wù)，服務(wù)員也去忙別的了。
廚房執(zhí)行完任務(wù)后，對(duì)當(dāng)時(shí)的那個(gè)服務(wù)員進(jìn)行通知，服務(wù)員接到通知后，再去執(zhí)行接下來(lái)的內(nèi)容，如把飯送到客人餐桌。
這是一個(gè)非常常見(jiàn)的異步場(chǎng)景，由于其中一方不愿意等待(或時(shí)時(shí)刻刻關(guān)注)另一方，但又不知道對(duì)方什么時(shí)候能做完，所以只能寄希望于對(duì)方做完的時(shí)候告訴自己一聲，然后自己再進(jìn)行后續(xù)的工作。
這就是我們常說(shuō)的異步回調(diào)(或通知)。
早上項(xiàng)目經(jīng)理開(kāi)完會(huì)，給大家分好任務(wù)，并把測(cè)試用例代碼也給了大家，說(shuō)誰(shuí)做完了跑一邊測(cè)試用例，通了就可以了。然后就散會(huì)，各自忙去了。
下午5點(diǎn)你做完了，開(kāi)始跑測(cè)試用例，很幸運(yùn)，一次性全部通過(guò)。你的任務(wù)就算完成了，接下來(lái)就可以干自己想干的事情，比如看“編程新說(shuō)”公眾號(hào)。
項(xiàng)目經(jīng)理是主(或I/O)線程，把任務(wù)交給各個(gè)開(kāi)發(fā)人員這些工作線程，并給每個(gè)人一段邏輯代碼，告訴他們?cè)谧约旱娜蝿?wù)完成后再執(zhí)行這一段邏輯代碼。
開(kāi)發(fā)人員完成任務(wù)后，接著執(zhí)行邏輯代碼，執(zhí)行完邏輯代碼后，就算已經(jīng)結(jié)束了。不再需要告知項(xiàng)目經(jīng)理一聲。
這也是一個(gè)常見(jiàn)的異步場(chǎng)景，一方給另一方安排好任務(wù)后，再給它一段邏輯代碼，接著彼此就分道揚(yáng)鑣。之后的日子里，你走你的陽(yáng)關(guān)道，我過(guò)我的獨(dú)木橋，井水不犯河水，老死不相往來(lái)。
這段邏輯代碼通常是由一個(gè)Runnable接口傳入，且是在任務(wù)完成時(shí)執(zhí)行，就暫且稱它為的“完成執(zhí)行”吧。
所以異步非阻塞代碼的最大特點(diǎn)就是，我給你分配任務(wù)，你完事給我回復(fù)，咱倆互相不耽誤。
最后來(lái)個(gè)小小的升華：
- 所謂異步就是你走你的，我走我的，大家各自往前走，表示的是一種事實(shí)形態(tài)。
- 所謂非阻塞就是快的快走，慢的慢走，一刻都不為你停留，表示的是一種直觀現(xiàn)象。
一言以蔽之，異步是形態(tài)，非阻塞是現(xiàn)象。
異步非阻塞它本身并沒(méi)有什么明顯的可圈可點(diǎn)的特征，注意我說(shuō)的是它“本身”。因?yàn)槲覀冋麄€(gè)世界都是按照異步非阻塞模式在運(yùn)行。
上廁所的時(shí)候玩手機(jī)，等車的時(shí)候玩手機(jī)，上班的時(shí)候玩手機(jī)，等飯的時(shí)候玩手機(jī)，回家以后玩手機(jī)，睡覺(jué)做夢(mèng)玩手機(jī)。第二天還是這樣的。哈哈。一個(gè)人就沒(méi)有被阻塞住的時(shí)候。
不可否認(rèn)，我們生活的社會(huì)又很復(fù)雜，主要是因?yàn)槿撕腿酥g的溝通、交流和協(xié)調(diào)有時(shí)并非一件容易之事。
同理，異步非阻塞“本身”并不難，難就難在怎么實(shí)現(xiàn)它。畢竟讓一群聽(tīng)不懂人話的二貨線程們互相溝通協(xié)調(diào)更非一件易事。
我看響應(yīng)式
所謂響應(yīng)式就是外界發(fā)生了變化，你要做出反應(yīng)。所以響應(yīng)式編程就是圍繞著變化來(lái)構(gòu)建的。
如何收集到原始變化，如何把這個(gè)變化告知相關(guān)處理者，處理者如何做出反應(yīng)，做出反應(yīng)的過(guò)程其實(shí)就是引發(fā)了新的變化，這個(gè)新的變化又該如何被收集，又該如何告知下一個(gè)處理者，如此往復(fù)，直至全部結(jié)束。
可以說(shuō)整個(gè)自然界都是響應(yīng)式的，因?yàn)樗鼈兌紩?huì)對(duì)外界的變化或自身的變化產(chǎn)生反應(yīng)。
先說(shuō)人類，冷的時(shí)候加衣，餓的時(shí)候吃飯，病的時(shí)候去醫(yī)院?？吹骄G色放松，看到藍(lán)色鎮(zhèn)定，看到紅色易激動(dòng)。
再說(shuō)動(dòng)植物，向日葵圍繞太陽(yáng)轉(zhuǎn)叫趨光性，植物的根系朝水多的地方生長(zhǎng)叫趨水性，鴿子可以磁場(chǎng)辨別方向，鯨魚(yú)、海歸都可以利用磁場(chǎng)記住自己走過(guò)的路。
所以響應(yīng)式“本身”是一個(gè)很簡(jiǎn)單的模型，你給我一個(gè)變化，我做出一個(gè)反應(yīng)。
動(dòng)植物都有一套完善的感覺(jué)器官，能夠感受到外界變化。同時(shí)他們又有超高的智商或完善的一套生物系統(tǒng)能夠?qū)@種變化作出反應(yīng)。這是數(shù)萬(wàn)年甚至數(shù)千萬(wàn)年進(jìn)化的結(jié)果，是基因決定的，所以看起來(lái)很自然。
再來(lái)看看編程界的響應(yīng)式，也是這兩個(gè)問(wèn)題，一是如何知道外界的變化，二是如何對(duì)這種變化作出反應(yīng)。
代碼可是沒(méi)有生命的，那就只能簡(jiǎn)單粗暴了。如何知道變化，那就讓別人告訴你唄。如何做出反應(yīng)，那就執(zhí)行一段邏輯代碼唄。
別人告訴你就等于異步回調(diào)/通知，執(zhí)行的這段邏輯代碼，可以是外界傳入的，也可以是自己本身的一個(gè)方法。
現(xiàn)在明白了吧，異步非阻塞就是響應(yīng)式。
最后來(lái)個(gè)小小升華：
所謂響應(yīng)式就是一個(gè)概念，或是一種編程模式，它并不是一個(gè)知識(shí)，也不是一個(gè)技術(shù)。但它需要用到一個(gè)技術(shù)，那就是實(shí)現(xiàn)異步非阻塞的技術(shù)。
我看Reactor
在傳統(tǒng)的編碼中，會(huì)將邏輯處理代碼寫(xiě)成方法，需要的數(shù)據(jù)由方法參數(shù)傳入，處理過(guò)的數(shù)據(jù)由方法的返回值返回。
執(zhí)行時(shí)以main方法為入口點(diǎn)啟動(dòng)，按照一定的順序執(zhí)行這些方法，數(shù)據(jù)依次流入流出每個(gè)方法，當(dāng)所有的方法執(zhí)行完時(shí)，數(shù)據(jù)也處理完了，就結(jié)束了。
整個(gè)過(guò)程是以邏輯代碼的執(zhí)行為主線，數(shù)據(jù)只是一個(gè)必須的參與者而已，因?yàn)榇a要處理數(shù)據(jù)，如果數(shù)據(jù)不到位，代碼就停下來(lái)不執(zhí)行，等待數(shù)據(jù)的到來(lái)。
這就是典型的同步阻塞式的執(zhí)行過(guò)程，非常簡(jiǎn)單，易于理解，而且代碼也很好寫(xiě)。
到目前為止，我們提到的都是響應(yīng)式的理論，那應(yīng)該怎樣去實(shí)現(xiàn)它呢，一時(shí)間還真沒(méi)有頭緒。
響應(yīng)式是異步非阻塞，和同步阻塞應(yīng)該是相對(duì)的。那我們不妨就拿響應(yīng)式往同步阻塞上套一下，看看能得到什么有價(jià)值的發(fā)現(xiàn)。
響應(yīng)式關(guān)注兩點(diǎn)，變化和反應(yīng)，而且是變化在前，反應(yīng)在后。同步阻塞也關(guān)注兩點(diǎn)，執(zhí)行邏輯和數(shù)據(jù)，而且是執(zhí)行邏輯在前，數(shù)據(jù)在后。
那就開(kāi)始建立對(duì)應(yīng)關(guān)系。因?yàn)?ldquo;反應(yīng)”是一系列行為動(dòng)作，所以應(yīng)該和“執(zhí)行邏輯”對(duì)應(yīng)。那“變化”只能和“數(shù)據(jù)”對(duì)應(yīng)，其實(shí)這是對(duì)的，“數(shù)據(jù)”由不可用到可用，本身就是發(fā)生了一個(gè)“變化”。
這個(gè)對(duì)應(yīng)關(guān)系建立的很完美，但是邏輯順序卻完全沖突。響應(yīng)式是由變化主導(dǎo)反應(yīng)，這很好理解，我都沒(méi)有變化，你無(wú)須做出反應(yīng)。同步阻塞是由執(zhí)行邏輯主導(dǎo)數(shù)據(jù)，這也很好理解，我代碼都沒(méi)執(zhí)行呢，根本不需要數(shù)據(jù)。
可見(jiàn)，它們的對(duì)應(yīng)關(guān)系非常完美，但主導(dǎo)順序完全相反，這就是一個(gè)非常非常有價(jià)值的發(fā)現(xiàn)。
因?yàn)槲覀冎恍璋淹阶枞惯^(guò)來(lái)，就是實(shí)現(xiàn)響應(yīng)式的大致方向。這樣的推理貌似是對(duì)的，但實(shí)際當(dāng)中是這樣的嗎?嗯，是這樣的。
現(xiàn)在請(qǐng)大家和我一起扭轉(zhuǎn)思維。原來(lái)以邏輯代碼執(zhí)行作為主線，數(shù)據(jù)作為參與者?，F(xiàn)在以數(shù)據(jù)作為主線，邏輯代碼執(zhí)行作為參與者。說(shuō)的再白一些，原來(lái)是數(shù)據(jù)傳遞到邏輯代碼里，現(xiàn)在是邏輯代碼傳遞到數(shù)據(jù)里。
有人也許會(huì)問(wèn)，邏輯代碼怎么傳遞?哈哈，Lambda表達(dá)式呀，函數(shù)式編程呀。
想象一下，有一個(gè)長(zhǎng)長(zhǎng)的管子，里面的水一直在流。
如果你想讓水變成橙色的，只需在管子上開(kāi)個(gè)口，加裝一個(gè)可以持續(xù)投放橙色染料的裝置，結(jié)果流經(jīng)它的水都變成橙色的了。
如果你想讓橙色的水變甜的話，只需在后面的管子上開(kāi)個(gè)口，加裝一個(gè)可以持續(xù)投放白糖的裝置，結(jié)果流經(jīng)它的水都變成甜的了。
同理，可以在后面繼續(xù)加裝投放檸檬酸的裝置，讓水變酸，在后面繼續(xù)加裝壓入二氧化碳的裝置，讓水帶氣泡。
最后發(fā)現(xiàn)，自來(lái)水經(jīng)過(guò)多道工序處理后變成了芬達(dá)。
如果把水流看作是數(shù)據(jù)流，把投放裝置看作是邏輯代碼，就變成了，數(shù)據(jù)先流入第一個(gè)邏輯代碼，處理后再流入第二個(gè)邏輯代碼，依次流下去直至結(jié)束。
這就是以數(shù)據(jù)作為主線，邏輯代碼只是參與者，同時(shí)它也是Reactor實(shí)現(xiàn)響應(yīng)式編程的原理，Spring官方使用的響應(yīng)式類庫(kù)就是Reactor。
其中，“以數(shù)據(jù)為主線”和“在變化時(shí)通知處理者”這兩個(gè)功能Reactor庫(kù)都已經(jīng)實(shí)現(xiàn)了，我們需要做的就是“對(duì)變化做出反應(yīng)”，即插入邏輯代碼。
Reactor入門(mén)
在Reactor中，有兩個(gè)非常重要的類，就是Mono和Flux，它們都是數(shù)據(jù)源，在它們內(nèi)部都已經(jīng)實(shí)現(xiàn)了“以數(shù)據(jù)為主線”和“在變化時(shí)通知處理者”這兩個(gè)功能，而且還提供了方法讓我們來(lái)插入邏輯代碼用于“對(duì)變化做出反應(yīng)”。
Mono表示0個(gè)或1個(gè)數(shù)據(jù)，F(xiàn)lux表示0到多個(gè)數(shù)據(jù)。先從簡(jiǎn)單的Mono開(kāi)始。
設(shè)計(jì)一個(gè)簡(jiǎn)單的示例，首先創(chuàng)建一個(gè)數(shù)據(jù)源，只包含一個(gè)數(shù)據(jù)10，第一個(gè)處理就是加1，第二個(gè)處理就是奇偶性過(guò)濾，第三個(gè)處理就是把這個(gè)數(shù)據(jù)消費(fèi)掉，然后就結(jié)束了。
為了清楚地看出來(lái)主線程執(zhí)行的是哪些代碼，工作線程執(zhí)行的是哪些代碼，特意打印了很多信息。
- public static void main(String[] args) {
- displayCurrTime(1);
- displayCurrThreadId(1);
- //創(chuàng)建一個(gè)數(shù)據(jù)源
- Mono.just(10)
- //延遲5秒再發(fā)射數(shù)據(jù)
- .delayElement(Duration.ofSeconds(5))
- //在數(shù)據(jù)上執(zhí)行一個(gè)轉(zhuǎn)換
- .map(n -> {
- displayCurrTime(2);
- displayCurrThreadId(2);
- displayValue(n);
- delaySeconds(2);
- return n + 1;
- })
- //在數(shù)據(jù)上執(zhí)行一個(gè)過(guò)濾
- .filter(n -> {
- displayCurrTime(3);
- displayCurrThreadId(3);
- displayValue(n);
- delaySeconds(3);
- return n % 2 == 0;
- })
- //如果數(shù)據(jù)沒(méi)了就用默認(rèn)值
- .defaultIfEmpty(9)
- //訂閱一個(gè)消費(fèi)者把數(shù)據(jù)消費(fèi)了
- .subscribe(n -> {
- displayCurrTime(4);
- displayCurrThreadId(4);
- displayValue(n);
- delaySeconds(2);
- System.out.println(n + " consumed, worker Thread over, exit.");
- });
- displayCurrTime(5);
- displayCurrThreadId(5);
- pause();
- }
- //顯示當(dāng)前時(shí)間
- static void displayCurrTime(int point) {
- System.out.println(point + " : " + LocalTime.now());
- }
- //顯示當(dāng)前線程Id
- static void displayCurrThreadId(int point) {
- System.out.println(point + " : " + Thread.currentThread().getId());
- }
- //顯示當(dāng)前的數(shù)值
- static void displayValue(int n) {
- System.out.println("input : " + n);
- }
- //延遲若干秒
- static void delaySeconds(int seconds) {
- try {
- TimeUnit.SECONDS.sleep(seconds);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- }
- //主線程暫停
- static void pause() {
- try {
- System.out.println("main Thread over, paused.");
- System.in.read();
- } catch (IOException e) {
- e.printStackTrace();
- }
- }
以下是輸出結(jié)果：
- 1 : 15:00:39.809
- 1 : 1
- 5 : 15:00:40.158
- 5 : 1
- main Thread over, paused.
- 2 : 15:00:45.158
- 2 : 9
- input : 10
- 3 : 15:00:47.160
- 3 : 9
- input : 11
- 4 : 15:00:50.162
- 4 : 9
- input : 9
- 9 consumed, worker Thread over, exit.
可以看到不到1秒鐘時(shí)間主線程就執(zhí)行完了。然后5秒后數(shù)據(jù)從數(shù)據(jù)源發(fā)射出來(lái)進(jìn)入第一步處理，2秒后進(jìn)入第二步處理，3秒后進(jìn)入第三步處理，數(shù)據(jù)被消費(fèi)掉，就結(jié)束了。其中主線程Id是1，工作線程Id是9。
這段代碼其實(shí)是建立了一個(gè)數(shù)據(jù)通道，在通道的指定位置上插入處理邏輯，等待數(shù)據(jù)到來(lái)。
主線程執(zhí)行的是建立通道的代碼，主線程很快執(zhí)行完，通道就建好了。此時(shí)只是一個(gè)空的通道，根本就沒(méi)有數(shù)據(jù)。
在數(shù)據(jù)到來(lái)時(shí)，由工作線程執(zhí)行每個(gè)節(jié)點(diǎn)的邏輯代碼來(lái)處理數(shù)據(jù)，然后把數(shù)據(jù)傳入下一個(gè)節(jié)點(diǎn)，如此反復(fù)直至結(jié)束。
所以，在寫(xiě)響應(yīng)式代碼的時(shí)候，心里一定要默念著，我所做的事情就是建立一條數(shù)據(jù)通道，在通道上指定的位置插入適合的邏輯處理代碼。同時(shí)還要切記，主線程執(zhí)行完時(shí)，只是建立了通道，并沒(méi)有數(shù)據(jù)。
如果本文內(nèi)容你沒(méi)有看懂，那就多看幾遍，保證能懂。如果你都看懂了，那恭喜你已經(jīng)入門(mén)響應(yīng)式編程了。
