摘要：本文介紹和點(diǎn)評(píng)上的等并發(fā)編程模型。異步更適合并發(fā)編程。同步使線程阻塞，導(dǎo)致等待。基本模型這是最簡(jiǎn)單的模型，創(chuàng)建線程來執(zhí)行一個(gè)任務(wù)，完畢后銷毀線程。響應(yīng)式編程是一種面向數(shù)據(jù)流和變化傳播的編程模式。起源于電信領(lǐng)域的的編程模型。

本文介紹和點(diǎn)評(píng)JVM上的Thread, Thread Pool, Future, Rx, async-await, Fiber, Actor等并發(fā)編程模型。本人經(jīng)驗(yàn)有限，難免粗陋，還請(qǐng)高手多多指教。

我們知道程序分為同步風(fēng)格和異步風(fēng)格。

可以寫成同步風(fēng)格用多個(gè)線程來并發(fā)執(zhí)行。

也可以寫成異步風(fēng)格以支持更為靈活的調(diào)度。

異步更適合并發(fā)編程。

異步的目的：充分利用計(jì)算資源。

同步使線程阻塞，導(dǎo)致等待。

異步是非阻塞的，無需等待。

如果發(fā)生了不必要的等待，就會(huì)浪費(fèi)資源，使程序變慢。

比如這樣的程序：

val res1 = get("http://server1")
val res2 = get("http://server2")
compute(res1, res2)

按照同步編程風(fēng)格，一定要先拿到res1，才能開始拿res2。

按照異步編程風(fēng)格，res1和res2互不依賴，發(fā)起對(duì)res1的獲取后，不必等待結(jié)果，而是馬上發(fā)起對(duì)res2的獲取，到了compute的時(shí)候，才需要阻塞等待兩個(gè)數(shù)據(jù)。

這是一種“順序解耦”。有時(shí)候我們并不要求某些操作按順序執(zhí)行！那么為什么要強(qiáng)制其順序呢？異步風(fēng)格讓我們能放棄強(qiáng)制，解放資源，減少不必要的等待。

如果異步操作能并行，程序性能就提升了，如果不能并行，程序性能就沒有提升。在當(dāng)今的硬件條件下，一般都能并行，所以異步成為了趨勢(shì)。

怎么個(gè)并行法？這要從計(jì)算機(jī)架構(gòu)說起了。讓我們把任何有處理能力的硬件看做一個(gè)處理單元——CPU顯然是主要的處理單元，I/O設(shè)備也是處理單元，比如說網(wǎng)卡、內(nèi)存控制器、硬盤控制器。CPU可以向一或多個(gè)I/O設(shè)備發(fā)出請(qǐng)求，當(dāng)設(shè)備在準(zhǔn)備數(shù)據(jù)時(shí)，CPU可以做其他事情(設(shè)備就緒后會(huì)用中斷通知CPU)，這時(shí)就有n個(gè)硬件在并行了！況且CPU本就是多核的，能做并行計(jì)算。除此之外，在分布式系統(tǒng)中，能同時(shí)調(diào)動(dòng)多臺(tái)計(jì)算機(jī)配合完成任務(wù)，也是并行。

因此，讓CPU等待、每次只請(qǐng)求一個(gè)I/O設(shè)備、不利用多核、不利用其他空閑的計(jì)算機(jī)，都是比較浪費(fèi)的。

下面我們來分析常見的并發(fā)編程模型。

這是最簡(jiǎn)單的模型，創(chuàng)建線程來執(zhí)行一個(gè)任務(wù)，完畢后銷毀線程。當(dāng)任務(wù)數(shù)量大時(shí)，會(huì)創(chuàng)建大量的線程。

大家都知道大量的線程會(huì)降低性能，但是你真的清楚性能開銷在哪里嗎？我試列舉一下：

創(chuàng)建線程

創(chuàng)建一個(gè)線程是比較耗時(shí)間的。需要請(qǐng)求操作系統(tǒng)、分配棧空間、初始化等工作。

上下文切換

大家都知道的，操作系統(tǒng)基本概念，不再贅述。值得注意的是，WAITING狀態(tài)的線程(多見于I/O等待)幾乎不會(huì)被調(diào)度，因此并不導(dǎo)致過多的上下文切換。

CPU cache miss

大量線程頻繁切換，勢(shì)必要訪問不同的數(shù)據(jù)，打亂了空間局部性，導(dǎo)致CPU cache miss增加，需要經(jīng)常訪問更慢的內(nèi)存，會(huì)明顯影響CPU密集型程序的性能，這點(diǎn)大家恐怕沒想到吧。

內(nèi)存占用

線程會(huì)增加內(nèi)存占用，線程的棧空間通常占1MB，1000個(gè)就是1GB。而且在棧上引用了很多對(duì)象，暫時(shí)不能回收，你說有多少個(gè)GB？

資源占用

一些有限的資源，如鎖、數(shù)據(jù)庫(kù)連接、文件句柄等，當(dāng)線程被掛起或阻塞，就暫時(shí)無人可用了，浪費(fèi)！還有死鎖風(fēng)險(xiǎn)！

那么分配多少線程好呢？

對(duì)于I/O密集型程序：一個(gè)經(jīng)驗(yàn)數(shù)值是兩倍于數(shù)據(jù)庫(kù)連接數(shù)，例如你有30個(gè)數(shù)據(jù)庫(kù)連接，就開60個(gè)線程；我還有個(gè)經(jīng)驗(yàn)數(shù)值是500以下，超過500就慢一些，如果調(diào)用棧特別深，這個(gè)數(shù)值還要下調(diào)。

對(duì)于CPU密集型程序：我的經(jīng)驗(yàn)數(shù)值是略多于CPU核心數(shù) (理論上是等于，但你難免有幾個(gè)阻塞操作)。除了核心數(shù)，還要考慮CPU cache的大小，最好實(shí)際測(cè)試一下。舉個(gè)例子，某司內(nèi)部的自動(dòng)重構(gòu)程序在Intel i7 3630QM CPU上測(cè)試，3~4個(gè)線程效果最好。

傳統(tǒng)的網(wǎng)絡(luò)程序是每個(gè)會(huì)話占用一個(gè)連接、一個(gè)線程。I/O多路復(fù)用（I/O multiplexing：多個(gè)會(huì)話共用一個(gè)連接）是應(yīng)C10K問題而生的，C10K就是1萬個(gè)連接。1萬個(gè)連接是很耗系統(tǒng)資源的，何況還有1萬個(gè)線程。從上文的分析可知，C1K的時(shí)候就可以開始運(yùn)用I/O多路復(fù)用了。

預(yù)留一些可反復(fù)使用的線程在一個(gè)池里，反復(fù)地接受任務(wù)。線程數(shù)量可能是固定的，也可能是一定范圍內(nèi)變動(dòng)的，依所選擇的線程池的實(shí)現(xiàn)而定。

這個(gè)模型是極其常用的，例如Tomcat就是用線程池來處理請(qǐng)求的。

注意——盡量不要阻塞任務(wù)線程；若實(shí)在無法避免，多開一些線程——每阻塞一個(gè)線程，線程池就少一個(gè)可用的線程。

Java典型的線程池有Executors.newFixedThreadPool Executors.newFixedThreadPool Executors.newFixedThreadPool Executors.newScheduledThreadPool等等，也可以直接new ThreadPoolExecutor(可指定線程數(shù)的上限和下限)。

Scala沒有增加新的線程池種類，但有個(gè)blocking方法能告訴線程池某個(gè)調(diào)用會(huì)阻塞，需要臨時(shí)增加1個(gè)線程。

Future是一個(gè)未來將會(huì)有值的對(duì)象，相當(dāng)于一個(gè)占位符（提貨憑證！）。

將任務(wù)投入線程池執(zhí)行時(shí)，可為任務(wù)綁定一個(gè)Future，憑此Future即可在未來取得任務(wù)執(zhí)行結(jié)果。未來是什么時(shí)候呢？要通過檢查Future內(nèi)部的狀態(tài)來獲知——任務(wù)完成時(shí)會(huì)修改這個(gè)狀態(tài)，將執(zhí)行結(jié)果存進(jìn)去。

最初的代碼示例可改寫為：

// 兩個(gè)future是并行的
val f1 = Future { get("http://server1") }
val f2 = Future { get("http://server2") }
compute(f1.get(), f2.get())

Rx (Reactive Extensions)是響應(yīng)式編程的一種具體形式。響應(yīng)式編程是一種面向數(shù)據(jù)流和變化傳播的編程模式。

我們知道Java 8提供了Stream類型，代表一個(gè)有限或無限的數(shù)據(jù)流，可應(yīng)用map, filter, collect等操作。Rx類似于Stream，也是有限或無限的數(shù)據(jù)流，只不過數(shù)據(jù)操作可以委托給線程池異步執(zhí)行。（Rx也像是生產(chǎn)者/消費(fèi)者模型的延伸，增加了分發(fā)和轉(zhuǎn)換的能力。對(duì)數(shù)據(jù)流進(jìn)行連接組合，這邊生產(chǎn)，那邊分發(fā)和轉(zhuǎn)換，源源不斷交給消費(fèi)者。）

以RxJava為例：

Flowable.just("file.txt")
.map(name -> Files.readLines(name))
.subscribe(lines -> System.out.println(lines.size()), Throwable::printStackTrace);

以Reactor為例：

Flux.fromIterable(getSomeLongList())
    .mergeWith(Flux.interval(100))
    .doOnNext(serviceA::someObserver)
    .map(d -> d * 2)
    .take(3)
    .onErrorResumeWith(errorHandler::fallback)
    .doAfterTerminate(serviceM::incrementTerminate)
    .subscribe(System.out::println);

由代碼可見，對(duì)數(shù)據(jù)流的操作很像是對(duì)集合的函數(shù)式操作，subscribe就是異步的forEach，doOnNext就是有返回值的異步的forEach。

主流實(shí)現(xiàn)有RxJava、Reactor、Akka Streams，API各有不同。但是它們都在靠攏Reactive Streams規(guī)范，想必會(huì)變得越來越相似。

async-await是一種特殊語法，能自動(dòng)把同步風(fēng)格代碼轉(zhuǎn)換成異步風(fēng)格代碼。正確運(yùn)用，就能使代碼在阻塞時(shí)自動(dòng)讓出控制權(quán)。

C#內(nèi)置的async-await是最完整的實(shí)現(xiàn)。Scala通過Async庫(kù)提供這個(gè)語法，代碼大概是這樣：

val future = async {
  println("Begin blocking")
  await {
    async {Thread.sleep(1000)}
  }
  println("End blocking")    
}

代碼會(huì)被自動(dòng)轉(zhuǎn)換成多種future的組合形式。無需特意處理，能并行的部分都會(huì)自動(dòng)并行。

Fiber是協(xié)程的仿制品。一般多線程是搶占式調(diào)度，你一個(gè)任務(wù)跑得好好的突然把你暫停；協(xié)程是協(xié)作式的，你一個(gè)任務(wù)阻塞或完成時(shí)要主動(dòng)讓出控制權(quán)，讓調(diào)度器換入另一個(gè)任務(wù)。

async-await自動(dòng)把代碼轉(zhuǎn)換成可自動(dòng)讓出控制權(quán)的形式，已經(jīng)有協(xié)程的雛形了。Fiber更加智能，連async-await語法都不用了，只要把代碼寫在Fiber里面，就像寫在Thread里面一樣，自動(dòng)異步化了。

async-await只能暫存當(dāng)前作用域（轉(zhuǎn)換成閉包），F(xiàn)iber則能暫存整個(gè)執(zhí)行棧（每個(gè)作用域只是一個(gè)棧幀）。當(dāng)然了，運(yùn)用嵌套的async-await也能暫存整個(gè)執(zhí)行棧，我更贊同如此，因?yàn)槟芨玫乜刂苾?nèi)存占用。

JVM上主流的實(shí)現(xiàn)是Quasar，通過java-agent改寫字節(jié)碼來實(shí)現(xiàn)，在需要讓出控制權(quán)時(shí)拋出異常打斷控制流(不必?fù)?dān)心異常方面的性能開銷)，保存執(zhí)行棧，然后換入另一個(gè)任務(wù)。

Java示例：

new Fiber() {
  @Override
  protected V run() throws SuspendExecution, InterruptedException {
    // your code
  }
}.start();

Kotlin示例：

fiber @Suspendable {
  // your code
}

代碼中調(diào)用的任何會(huì)阻塞的方法都要標(biāo)記@Suspendable，讓Quasar知道調(diào)這個(gè)方法時(shí)要暫停當(dāng)前Fiber并執(zhí)行另一個(gè)Fiber，同時(shí)用另外的線程池執(zhí)行會(huì)阻塞的方法。

起源于電信領(lǐng)域的Erlang的編程模型。actor是任務(wù)處理單元：每個(gè)actor只處理一個(gè)任務(wù)，每個(gè)任務(wù)同時(shí)只有一個(gè)actor處理(如果有大任務(wù)，就要分解成小任務(wù))，actor之間用消息來通信。

在Erlang中，每個(gè)actor是一個(gè)輕量級(jí)進(jìn)程，有獨(dú)立的內(nèi)存空間(所以通信只能靠消息)，因此有獨(dú)立的垃圾回收，不會(huì)stop the world。

actor可以發(fā)了消息就不管了(tell)，這是典型的異步；也可以發(fā)了消息等回應(yīng)(ask)，返回值是一個(gè)Future，實(shí)際上是創(chuàng)建了一個(gè)新的actor在悄悄等待回應(yīng)，仍然是異步。

actor可以透明地分布在不同機(jī)器上，消息可以發(fā)給本機(jī)的actor，也可以發(fā)給遠(yuǎn)程的actor。

JVM上唯一成熟的實(shí)現(xiàn)是Akka，JVM不能給每個(gè)actor獨(dú)立的內(nèi)存，垃圾回收仍可能stop the world。

actor顯然是一個(gè)對(duì)象，擁有狀態(tài)和行為。
actor也可被視為一個(gè)閉包，擁有函數(shù)和上下文（整個(gè)對(duì)象的狀態(tài)都是上下文）。
actor每次能接收并處理一個(gè)消息，處理中可以發(fā)送消息給自己或另一個(gè)actor，然后掛起或結(jié)束。
為什么要發(fā)送消息給自己呢？因?yàn)檎谔幚硐r(shí)是不能掛起的，只能在“一個(gè)消息之后，下一個(gè)消息之前”的間隙中掛起。
假設(shè)你收到一個(gè)A消息，執(zhí)行前半段業(yè)務(wù)邏輯，要做一次I/O再執(zhí)行后半段業(yè)務(wù)邏輯。做I/O時(shí)應(yīng)當(dāng)結(jié)束當(dāng)前處理，當(dāng)IO完成時(shí)給自己發(fā)一個(gè)B消息，下次再讓你在處理B消息時(shí)完成剩余業(yè)務(wù)邏輯。前后邏輯要分開寫，共享變量要聲明為actor的對(duì)象字段。
偽代碼如下：

class MyActor extends BasicActor {
  var halfDoneResult: XXX = None

  def receive(): Receive = {
    case A => {
      halfDoneResult = 前半段邏輯()
      doIO(halfDoneResult).onComplete {
        self ! B()
      }
    }
    case B => 后半段邏輯(halfDoneResult)
  }
}

當(dāng)actor的狀態(tài)要徹底改變時(shí)，可以用become操作徹底改變actor的行為。從面向?qū)ο缶幊痰脑O(shè)計(jì)模式來看，這是state pattern，從函數(shù)式編程來看，這是把一個(gè)函數(shù)變換成另一個(gè)函數(shù)。

由此可見，actor模型就是把函數(shù)表示成了更容易控制的對(duì)象，以便于滿足一些并發(fā)或分布式方面的架構(gòu)約束。

這段邏輯假如改寫成async-await或fiber，偽代碼如下所示，簡(jiǎn)單多了：

def logicInAsync() = async {
  val halfDoneResult = 前半段邏輯()
  await { doIO(halfDoneResult) }
  后半段邏輯(halfDoneResult)
}

def logicInFiber() = fiber {
  val halfDoneResult = 前半段邏輯()
  doIO(halfDoneResult)
  后半段邏輯(halfDoneResult)
}

可以看出，相比于async-await或Fiber，actor就是一種狀態(tài)機(jī)，是較為底層、不易用的編程模型。但是actor附帶了成熟的分布式能力。

我感覺actor很像異步版的EJB。EJB中有stateless session bean和stateful session bean，actor也可按stateless和stateful來分類。

PayPal的支付系統(tǒng)就是基于Akka的，還為此編寫并開源了一個(gè)Squbs框架。業(yè)務(wù)邏輯仍是用actor實(shí)現(xiàn)，Squbs只增加了集成和運(yùn)維方面的支持（這個(gè)也重要）。然而我對(duì)此技術(shù)路線（業(yè)務(wù)邏輯基于actor）持審慎態(tài)度，接下來就分類說明我的意見：

我認(rèn)為，此架構(gòu)只需要三種通信模型：消息隊(duì)列、同步RPC、異步RPC。

消息隊(duì)列：異步的，只管發(fā)送消息，不等待返回結(jié)果(即使需要知道結(jié)果，讓consumer向sender回發(fā)一個(gè)消息即可，會(huì)異步觸發(fā)sender這邊的回調(diào))。消息可能觸發(fā)遠(yuǎn)端的一個(gè)任務(wù)，也可能觸發(fā)更多消息的發(fā)出，也可能什么都不觸發(fā)。

同步RPC：同步的，向遠(yuǎn)程結(jié)點(diǎn)發(fā)送消息，保持當(dāng)前的執(zhí)行棧，同步等待回復(fù)。執(zhí)行棧一直占著線程。簡(jiǎn)單易懂而廣泛流行的模型。

異步RPC：異步的，向遠(yuǎn)程結(jié)點(diǎn)發(fā)送消息，保持當(dāng)前的執(zhí)行棧，異步等待回復(fù)。執(zhí)行棧可暫時(shí)被換出線程，收到回復(fù)時(shí)再切回。

消息隊(duì)列、同步RPC都不需要Akka出場(chǎng)，自有各種MQ、RPC框架來解決。至于異步RPC，GRPC是一個(gè)跨語言的RPC框架，也可建造一個(gè)基于WebSocket協(xié)議的RPC框架。如果無需跨語言，也可讓Akka出場(chǎng)，但不是直接基于Akka編程——而是在Akka之上構(gòu)建一個(gè)RPC層。如果功力較高，可直接基于Netty構(gòu)建RPC層。

actor進(jìn)行“請(qǐng)求-響應(yīng)”往返通信時(shí)，在收到響應(yīng)之前，請(qǐng)求端的actor要掛起、暫存在內(nèi)存中。協(xié)程進(jìn)行這種通信時(shí)，則是請(qǐng)求端的執(zhí)行棧要掛起、暫存在內(nèi)存中。

這是actor的龍興之地， 也是最合適的用武之地。

以即時(shí)聊天(IM)為例，用actor怎么實(shí)現(xiàn)呢？

如果每個(gè)actor對(duì)應(yīng)一個(gè)人，1萬人只需要1萬個(gè)actor，1萬個(gè)連接。用戶A對(duì)用戶B說話時(shí)，actor A收到消息，轉(zhuǎn)發(fā)給actor B，由actor B發(fā)送給用戶B，反之亦然。

如果每個(gè)actor對(duì)應(yīng)一個(gè)會(huì)話，最多需要1億(1萬×1萬)個(gè)actor，連接數(shù)不到1億(同一臺(tái)服務(wù)器與某個(gè)用戶的連接可供相關(guān)會(huì)話共用)，但也過多了。

因此選擇第一種實(shí)現(xiàn)：每個(gè)actor對(duì)應(yīng)一個(gè)人，actor要記得它對(duì)應(yīng)哪個(gè)人、消息往來情況如何，這就是“狀態(tài)”！如果10萬用戶在線，就要10萬連接(這與IO多路復(fù)用無關(guān)，對(duì)吧？)，單機(jī)顯然hold不住，需要多機(jī)。如果用actor A和actor B不在同一臺(tái)機(jī)器，就要遠(yuǎn)程通信了。對(duì)基于Akka的程序來說，本地通信或遠(yuǎn)程通信是透明的，贊！

其實(shí)不用actor也能實(shí)現(xiàn)，一切狀態(tài)和關(guān)系都能用數(shù)據(jù)結(jié)構(gòu)來表達(dá)，只不過actor可能更方便一些。

總而言之，Akka模仿Erlang，精心設(shè)計(jì)了業(yè)務(wù)無關(guān)的actor的概念，然而越是精心設(shè)計(jì)的業(yè)務(wù)無關(guān)的概念越有可能不符合多變的業(yè)務(wù)需求:)。如果問我用不用actor，我只能說，看情況吧。也希望有哪位英雄能介紹一兩個(gè)非actor不可的場(chǎng)景。

現(xiàn)在，假設(shè)有一個(gè)微服務(wù)架構(gòu)，在眾多服務(wù)中有A、B、C三個(gè)服務(wù)，調(diào)用順序是A->B->C。RPC只能以A->B->C的方向請(qǐng)求，再以C->B->A的方向響應(yīng)；actor則能讓C直接發(fā)送響應(yīng)給A。但如果C要直接回復(fù)A，就要與A建立連接，使網(wǎng)絡(luò)拓?fù)浜鸵蕾嚬芾矶甲儚?fù)雜了——如非必要，勿增復(fù)雜。

為了避免，利用MQ來發(fā)送響應(yīng)？MQ就像一個(gè)聊天服務(wù)，讓分布各處的服務(wù)能彼此聊天。IM、actor、MQ，一切都聯(lián)系起來了，有沒有感受到妙不可言的意境？

但是壓力集中到了MQ的broker，網(wǎng)絡(luò)也多了一跳(publisher->broker->consumer)，對(duì)性能有所影響。

本文介紹、點(diǎn)評(píng)了JVM上多種常見的并發(fā)模型，并試圖建立模型之間的聯(lián)系，最后以分布式架構(gòu)為例加以分析。

那么應(yīng)用程序要怎么寫呢？看文檔吧，各種庫(kù)或框架都希望有人來用，滿足它們吧！