Java 實(shí)現(xiàn)協(xié)程的方法
協(xié)程(Coroutine)這個(gè)詞其實(shí)有很多叫法,比如有的人喜歡稱為纖程(Fiber),或者綠色線程(GreenThread)。其實(shí)究其本質(zhì),對(duì)于協(xié)程最直觀的解釋是線程的線程。雖然讀上去有點(diǎn)拗口,但本質(zhì)上就是這樣。
協(xié)程的核心在于調(diào)度那塊由他來負(fù)責(zé)解決,遇到阻塞操作,立刻放棄掉,并且記錄當(dāng)前棧上的數(shù)據(jù),阻塞完后立刻再找一個(gè)線程恢復(fù)棧并把阻塞的結(jié)果放到這個(gè)線程上去跑,這樣看上去好像跟寫同步代碼沒有任何差別,這整個(gè)流程可以稱為coroutine,而跑在由coroutine負(fù)責(zé)調(diào)度的線程稱為Fiber。
java協(xié)程的實(shí)現(xiàn)
早期,在JVM上實(shí)現(xiàn)協(xié)程一般會(huì)使用kilim,不過這個(gè)工具已經(jīng)很久不更新了,現(xiàn)在常用的工具是Quasar,而本文章會(huì)全部基于Quasar來介紹。
下面嘗試通過Quasar來實(shí)現(xiàn)類似于go語言的coroutine以及channel。
為了能有明確的對(duì)比,這里先用go語言實(shí)現(xiàn)一個(gè)對(duì)于10以內(nèi)自然數(shù)分別求平方的例子。
func counter(out chan<- int) { for x := 0; x < 10; x++ { out <- x } close(out)}func squarer(out chan<- int, in <-chan int) { for v := range in { out <- v * v } close(out)}func printer(in <-chan int) { for v := range in { fmt.Println(v) }}func main() { //定義兩個(gè)int類型的channel naturals := make(chan int) squares := make(chan int) //產(chǎn)生兩個(gè)Fiber,用go關(guān)鍵字 go counter(naturals) go squarer(squares, naturals) //獲取計(jì)算結(jié)果 printer(squares)}
上面這個(gè)例子,通過channel兩解耦兩邊的數(shù)據(jù)共享。對(duì)于這個(gè)channel,大家可以理解為Java里的SynchronousQueue。下面我直接上Quasar版JAVA代碼的,幾乎可以原封不動(dòng)的復(fù)制go語言的代碼。
public class Example { private static void printer(Channel<Integer> in) throws SuspendExecution, InterruptedException { Integer v; while ((v = in.receive()) != null) { System.out.println(v); } } public static void main(String[] args) throws ExecutionException, InterruptedException, SuspendExecution { //定義兩個(gè)Channel Channel<Integer> naturals = Channels.newChannel(-1); Channel<Integer> squares = Channels.newChannel(-1); //運(yùn)行兩個(gè)Fiber實(shí)現(xiàn). new Fiber(() -> { for (int i = 0; i < 10; i++) naturals.send(i); naturals.close(); }).start(); new Fiber(() -> { Integer v; while ((v = naturals.receive()) != null) squares.send(v * v); squares.close(); }).start(); printer(squares); }}
兩者對(duì)比,看上去Java似好像更復(fù)雜些,沒辦法這就是Java的風(fēng)格,而且這還是通過第三方的庫來實(shí)現(xiàn)的。
說到這里各位肯定對(duì)Fiber很好奇了。也許你會(huì)表示懷疑Fiber是不是如上面所描述的那樣,下面我們嘗試用Quasar建立一百萬個(gè)Fiber,看看內(nèi)存占用多少,我先嘗試了創(chuàng)建百萬個(gè)Thread。
for (int i = 0; i < 1_000_000; i++) { new Thread(() -> { try { Thread.sleep(10000); } catch (InterruptedException e) { e.printStackTrace(); } }).start();}
很不幸,直接報(bào)Exception in thread 'main' java.lang.OutOfMemoryError: unable to create new native thread,這是情理之中的。下面是通過Quasar建立百萬個(gè)Fiber。
public static void main(String[] args) throws ExecutionException, InterruptedException, SuspendExecution { int FiberNumber = 1_000_000; CountDownLatch latch = new CountDownLatch(1); AtomicInteger counter = new AtomicInteger(0); for (int i = 0; i < FiberNumber; i++) { new Fiber(() -> { counter.incrementAndGet(); if (counter.get() == FiberNumber) { System.out.println('done'); } Strand.sleep(1000000); }).start(); } latch.await();}
我這里加了latch,阻止程序跑完就關(guān)閉,Strand.sleep其實(shí)跟Thread.sleep一樣,只是這里針對(duì)的是Fiber。
最終控制臺(tái)是可以輸出done的,說明程序已經(jīng)創(chuàng)建了百萬個(gè)Fiber,設(shè)置Sleep是為了讓Fiber一直運(yùn)行,從而方便計(jì)算內(nèi)存占用。官方宣稱一個(gè)空閑的Fiber大約占用400Byte,那這里應(yīng)該是占用400MB堆內(nèi)存,但是這里通過jmap -heap pid顯示大約占用了1000MB,也就是說一個(gè)Fiber占用1KB。
Quasar是怎么實(shí)現(xiàn)Fiber的
其實(shí)Quasar實(shí)現(xiàn)的coroutine的方式與Go語言很像,只不過前者是使用框架來實(shí)現(xiàn),而go語言則是語言內(nèi)置的功能。
不過如果你熟悉了Go語言的調(diào)度機(jī)制的話,那么對(duì)于Quasar的調(diào)度機(jī)制就會(huì)好理解很多了,因?yàn)閮烧哂泻芏嘞嗨浦帯?/p>
Quasar里的Fiber其實(shí)是一個(gè)continuation,他可以被Quasar定義的scheduler調(diào)度,一個(gè)continuation記錄著運(yùn)行實(shí)例的狀態(tài),而且會(huì)被隨時(shí)中斷,并且也會(huì)隨后在他被中斷的地方恢復(fù)。
Quasar其實(shí)是通過修改bytecode來達(dá)到這個(gè)目的,所以運(yùn)行Quasar程序的時(shí)候,你需要先通過java-agent在運(yùn)行時(shí)修改你的代碼,當(dāng)然也可以在編譯期間這么干。go語言的內(nèi)置了自己的調(diào)度器,而Quasar則是默認(rèn)使用ForkJoinPool這個(gè)具有work-stealing功能的線程池來當(dāng)調(diào)度器。work-stealing非常重要,因?yàn)槟悴磺宄膫€(gè)Fiber會(huì)先執(zhí)行完,而work-stealing可以動(dòng)態(tài)的從其他的等等隊(duì)列偷一個(gè)context過來,這樣可以最大化使用CPU資源。
那這里你會(huì)問了,Quasar怎么知道修改哪些字節(jié)碼呢,其實(shí)也很簡(jiǎn)單,Quasar會(huì)通過java-agent在運(yùn)行時(shí)掃描哪些方法是可以中斷的,同時(shí)會(huì)在方法被調(diào)用前和調(diào)度后的方法內(nèi)插入一些continuation邏輯,如果你在方法上定義了@Suspendable注解,那Quasar會(huì)對(duì)調(diào)用該注解的方法做類似下面的事情。
這里假設(shè)你在方法f上定義了@Suspendable,同時(shí)去調(diào)用了有同樣注解的方法g,那么所有調(diào)用f的方法會(huì)插入一些字節(jié)碼,這些字節(jié)碼的邏輯就是記錄當(dāng)前Fiber棧上的狀態(tài),以便在未來可以動(dòng)態(tài)的恢復(fù)。(Fiber類似線程也有自己的棧)。在suspendable方法鏈內(nèi)Fiber的父類會(huì)調(diào)用Fiber.park,這樣會(huì)拋出SuspendExecution異常,從而來停止線程的運(yùn)行,好讓Quasar的調(diào)度器執(zhí)行調(diào)度。這里的SuspendExecution會(huì)被Fiber自己捕獲,業(yè)務(wù)層面上不應(yīng)該捕獲到。如果Fiber被喚醒了(調(diào)度器層面會(huì)去調(diào)用Fiber.unpark),那么f會(huì)在被中斷的地方重新被調(diào)用(這里Fiber會(huì)知道自己在哪里被中斷),同時(shí)會(huì)把g的調(diào)用結(jié)果(g會(huì)return結(jié)果)插入到f的恢復(fù)點(diǎn),這樣看上去就好像g的return是f的local variables了,從而避免了callback嵌套。
上面說了一大堆,其實(shí)簡(jiǎn)單點(diǎn)來講就是,想辦法讓運(yùn)行中的線程棧停下來,然后讓Quasar的調(diào)度器介入。
JVM線程中斷的條件有兩個(gè):
1、拋異常
2、return。
而在Quasar中,一般就是通過拋異常的方式來達(dá)到的,所以你會(huì)看到上面的代碼會(huì)拋出SuspendExecution。但是如果你真捕獲到這個(gè)異常,那就說明有問題了,所以一般會(huì)這么寫。
@Suspendablepublic int f() { try { // do some stuff return g() * 2; } catch(SuspendExecution s) { //這里不應(yīng)該捕獲到異常. throw new AssertionError(s); }}
以上就是Java 實(shí)現(xiàn)協(xié)程的方法的詳細(xì)內(nèi)容,更多關(guān)于Java 實(shí)現(xiàn)協(xié)程的資料請(qǐng)關(guān)注好吧啦網(wǎng)其它相關(guān)文章!
相關(guān)文章:
1. 低版本IE正常運(yùn)行HTML5+CSS3網(wǎng)站的3種解決方案2. 利用CSS制作3D動(dòng)畫3. 使用Spry輕松將XML數(shù)據(jù)顯示到HTML頁的方法4. 測(cè)試模式 - XSL教程 - 55. html5手機(jī)觸屏touch事件介紹6. 用xslt+css讓RSS顯示的跟網(wǎng)頁一樣漂亮7. 《CSS3實(shí)戰(zhàn)》筆記--漸變?cè)O(shè)計(jì)(一)8. JSP的Cookie在登錄中的使用9. xpath簡(jiǎn)介_動(dòng)力節(jié)點(diǎn)Java學(xué)院整理10. 讀大數(shù)據(jù)量的XML文件的讀取問題
