多核處理器越來越普及。有沒有一種簡單的辦法，能夠讓我們寫的軟體釋放多核的威力？是有的。隨著Golang, Erlang, Scala等為併發設計的程式語言的興起，新的併發模式逐漸清晰。正如程序式程式設計和麵向物件一樣，一個好的程式設計模式有一個極其簡潔的核心，還有在此之上豐富的外延。可以解決現實世界中各種各樣的問題。本文以GO語言為例，解釋其中核心、外延。

併發模式之核心

這種併發模式的核心只需要 協程 和 通道 就夠了。協程負責執行程式碼，通道負責在協程之間傳遞事件。

不久前，併發程式設計是個非常困難的事。要想編寫一個良好的併發程式，我們不得不瞭解執行緒，鎖，semaphore，barrier甚至CPU更新快取記憶體的方式，而且他們個個都有怪脾氣，處處是陷阱。筆者除非萬不得以，決不會自己操作這些底層併發元素。一個簡潔的併發模式不需要這些複雜的底層元素，協程和通道就夠了。

協程是輕量級的執行緒。在程序式程式設計中，當呼叫一個過程的時候，需要等待其執行完才返回。而呼叫一個協程的時候，不需要等待其執行完，會立即返回。協程十分輕量，Go語言可以在一個程序中執行有數以十萬計的協程，依舊保持高效能。而對於普通的平臺，一個程序有數千個執行緒，其CPU會忙於上下文切換，效能急劇下降。隨意建立執行緒可不是一個好主意，但是我們可以大量使用的協程。

通道是協程之間的資料傳輸通道。通道可以在眾多的協程之間傳遞資料，具體可以值也可以是個引用。通道有兩種使用方式。

• 協程可以試圖向通道放入資料，如果通道滿了，會掛起協程，直到通道可以為他放入資料為止。
• 協程可以試圖向通道索取資料，如果通道沒有資料，會掛起協程，直到通道返回資料為止。

如此，通道就可以在傳遞資料的同時，控制協程的執行。有點像事件驅動，也有點像阻塞佇列。

這兩個概念非常的簡單，各個語言平臺都會有相應的實現。在Java和C上也各有庫可以實現兩者。

只要有協程和通道，就可以優雅的解決併發的問題。不必使用其他和併發有關的概念。那如何用這兩把利刃解決各式各樣的實際問題呢？

併發模式之外延

協程相較於執行緒，可以大量建立。開啟這扇門，我們拓展出新的用法，可以做生成器，可以讓函式返回“服務”，可以讓迴圈併發執行，還能共享變數。但是出現新的用法的同時，也帶來了新的棘手問題，協程也會洩漏，不恰當的使用會影響效能。下面會逐一介紹各種用法和問題。演示的程式碼用GO語言寫成，因為其簡潔明瞭，而且支援全部功能。

生成器

有的時候，我們需要有一個函式能不斷生成資料。比方說這個函式可以讀檔案，讀網路，生成自增長序列，生成隨機數。這些行為的特點就是，函式的已知一些變數，如檔案路徑。然後不斷呼叫，返回新的資料。

下面生成隨機數為例， 以讓我們做一個會併發執行的隨機數生成器。

非併發的做法是這樣的：

func rand_generator_1() int {
    return rand.Int()
}

上面是一個函式，返回一個int。假如rand.Int()這個函式呼叫需要很長時間等待，那該函式的呼叫者也會因此而掛起。所以我們可以建立一個協程，專門執行rand.Int()。

// 函式 rand_generator_2，返回 通道(Channel)
func rand_generator_2() chan int {
    // 建立通道
    out := make(chan int)
    // 建立協程
    go func() {
        for {
            //向通道內寫入資料，如果無人讀取會等待
            out <- rand.Int()
        }
    }()
    return out
}

func main() {
    // 生成隨機數作為一個服務
    rand_service_handler := rand_generator_2()
    // 從服務中讀取隨機數並列印
    fmt.Printf("%d\n", <-rand_service_handler)
}

上面的這段函式就可以併發執行了rand.Int()。有一點值得注意到函式的返回可以理解為一個“服務”。但我們需要獲取隨機資料 時候，可以隨時向這個服務取用，他已經為我們準備好了相應的資料，無需等待，隨要隨到。如果我們呼叫這個服務不是很頻繁，一個協程足夠滿足我們的需求了。但如果我們需要大量訪問，怎麼辦？我們可以用下面介紹的多路複用技術，啟動若干生成器，再將其整合成一個大的服務。

呼叫生成器，可以返回一個“服務”。可以用在持續獲取資料的場合。用途很廣泛，讀取資料，生成ID，甚至定時器。這是一種非常簡潔的思路，將程式併發化。

多路複用

多路複用是讓一次處理多個佇列的技術。Apache使用處理每個連線都需要一個程序，所以其併發效能不是很好。而Nighx使用多路複用的技術，讓一個程序處理多個連線，所以併發效能比較好。同樣，在協程的場合，多路複用也是需要的，但又有所不同。多路複用可以將若干個相似的小服務整合成一個大服務。

那麼讓我們用多路複用技術做一個更高併發的隨機數生成器吧。

// 函式 rand_generator_3 ，返回通道(Channel)
func rand_generator_3() chan int {
    // 建立兩個隨機數生成器服務
    rand_generator_1 := rand_generator_2()
    rand_generator_2 := rand_generator_2()

    //建立通道
    out := make(chan int)

    //建立協程
    go func() {
        for {
            //讀取生成器1中的資料，整合
            out <- <-rand_generator_1
        }
    }()
    go func() {
        for {
            //讀取生成器2中的資料，整合
            out <- <-rand_generator_2
        }
    }()
    return out
}

上面是使用了多路複用技術的高併發版的隨機數生成器。通過整合兩個隨機數生成器，這個版本的能力是剛才的兩倍。雖然協程可以大量建立，但是眾多協程還是會爭搶輸出的通道。Go語言提供了Select關鍵字來解決，各家也有各家竅門。加大輸出通道的緩衝大小是個通用的解決方法。

多路複用技術可以用來整合多個通道。提升效能和操作的便捷。配合其他的模式使用有很大的威力。

Furture技術

Furture是一個很有用的技術，我們常常使用Furture來操作執行緒。我們可以在使用執行緒的時候，可以建立一個執行緒，返回Furture，之後可以通過它等待結果。 但是在協程環境下的Furtue可以更加徹底，輸入引數同樣可以是Furture的。

呼叫一個函式的時候，往往是引數已經準備好了。呼叫協程的時候也同樣如此。但是如果我們將傳入的引數設為通道，這樣我們就可以在不準備好引數的情況下呼叫函式。這樣的設計可以提供很大的自由度和併發度。函式呼叫和函式引數準備這兩個過程可以完全解耦。下面舉一個用該技術訪問資料庫的例子。

//一個查詢結構體
type query struct {
    //引數Channel
    sql chan string
    //結果Channel
    result chan string
}

//執行Query
func execQuery(q query) {
    //啟動協程
    go func() {
        //獲取輸入
        sql := <-q.sql
        //訪問資料庫，輸出結果通道
        q.result <- "get " + sql
    }()

}

func main() {
    //初始化Query
    q :=
        query{make(chan string, 1), make(chan string, 1)}
    //執行Query，注意執行的時候無需準備引數
    execQuery(q)

    //準備引數
    q.sql <- "select * from table"
    //獲取結果
    fmt.Println(<-q.result)
}

上面利用Furture技術，不單讓結果在Furture獲得，引數也是在Furture獲取。準備好引數後，自動執行。Furture和生成器的區別在於，Furture返回一個結果，而生成器可以重複呼叫。還有一個值得注意的地方，就是將引數Channel和結果Channel定義在一個結構體裡面作為引數，而不是返回結果Channel。這樣做可以增加聚合度，好處就是可以和多路複用技術結合起來使用。

Furture技術可以和各個其他技術組合起來用。可以通過多路複用技術，監聽多個結果Channel，當有結果後，自動返回。也可以和生成器組合使用，生成器不斷生產資料，Furture技術逐個處理資料。Furture技術自身還可以首尾相連，形成一個併發的pipe filter。這個pipe filter可以用於讀寫資料流，操作資料流。

Future是一個非常強大的技術手段。可以在呼叫的時候不關心資料是否準備好，返回值是否計算好的問題。讓程式中的元件在準備好資料的時候自動跑起來。

併發迴圈

迴圈往往是效能上的熱點。如果效能瓶頸出現在CPU上的話，那麼九成可能性熱點是在一個迴圈體內部。所以如果能讓迴圈體併發執行，那麼效能就會提高很多。

要併發迴圈很簡單，只有在每個迴圈體內部啟動協程。協程作為迴圈體可以併發執行。呼叫啟動前設定一個計數器，每一個迴圈體執行完畢就在計數器上加一個元素，呼叫完成後通過監聽計數器等待迴圈協程全部完成。

//建立計數器
sem := make(chan int, N);
//FOR迴圈體
for i,xi := range data {
    //建立協程
    go func (i int, xi float) {
        doSomething(i,xi);
        //計數
        sem <- 0;
    } (i, xi);
}
// 等待迴圈結束
for i := 0; i < N; ++i { <-sem }

上面是一個併發迴圈例子。通過計數器來等待迴圈全部完成。如果結合上面提到的Future技術的話，則不必等待。可以等到真正需要的結果的地方，再去檢查資料是否完成。

通過併發迴圈可以提供效能，利用多核，解決CPU熱點。正因為協程可以大量建立，才能在迴圈體中如此使用，如果是使用執行緒的話，就需要引入執行緒池之類的東西，防止建立過多執行緒，而協程則簡單的多。

Chain Filter技術

前面提到了Future技術首尾相連，可以形成一個併發的pipe filter。這種方式可以做很多事情，如果每個Filter都由同一個函式組成，還可以有一種簡單的辦法把他們連起來。

由於每個Filter協程都可以併發執行，這樣的結構非常有利於多核環境。下面是一個例子，用這種模式來產生素數。

// A concurrent prime sieve

package main

// Send the sequence 2, 3, 4, ... to channel 'ch'.
func Generate(ch chan<- int) {
    for i := 2; ; i++ {
        ch <- i // Send 'i' to channel 'ch'.
    }
}

// Copy the values from channel 'in' to channel 'out',
// removing those divisible by 'prime'.
func Filter(in <-chan int, out chan<- int, prime int) {
    for {
        i := <-in // Receive value from 'in'.
        if i%prime != 0 {
            out <- i // Send 'i' to 'out'.
        }
    }
}

// The prime sieve: Daisy-chain Filter processes.
func main() {
    ch := make(chan int) // Create a new channel.
    go Generate(ch)      // Launch Generate goroutine.
    for i := 0; i < 10; i++ {
        prime := <-ch
        print(prime, "\n")
        ch1 := make(chan int)
        go Filter(ch, ch1, prime)
        ch = ch1
    }
}

上面的程式建立了10個Filter，每個分別過濾一個素數，所以可以輸出前10個素數。

Chain-Filter通過簡單的程式碼建立併發的過濾器鏈。這種辦法還有一個好處，就是每個通道只有兩個協程會訪問，就不會有激烈的競爭，效能會比較好。

共享變數

協程之間的通訊只能夠通過通道。但是我們習慣於共享變數，而且很多時候使用共享變數能讓程式碼更簡潔。比如一個Server有兩個狀態開和關。其他僅僅希望獲取或改變其狀態，那又該如何做呢。可以將這個變數至於0通道中，並使用一個協程來維護。

下面的例子描述如何用這個方式，實現一個共享變數。

//共享變數有一個讀通道和一個寫通道組成
type sharded_var struct {
    reader chan int
    writer chan int
}

//共享變數維護協程
func sharded_var_whachdog(v sharded_var) {
    go func() {
        //初始值
        var value int = 0
        for {
            //監聽讀寫通道，完成服務
            select {
            case value = <-v.writer:
            case v.reader <- value:
            }
        }
    }()
}

func main() {
    //初始化，並開始維護協程
    v := sharded_var{make(chan int), make(chan int)}
    sharded_var_whachdog(v)

    //讀取初始值
    fmt.Println(<-v.reader)
    //寫入一個值
    v.writer <- 1
    //讀取新寫入的值
    fmt.Println(<-v.reader)
}

這樣，就可以在協程和通道的基礎上實現一個協程安全的共享變量了。定義一個寫通道，需要更新變數的時候，往裡寫新的值。再定義一個讀通道，需要讀的時候，從裡面讀。通過一個單獨的協程來維護這兩個通道。保證資料的一致性。

一般來說，協程之間不推薦使用共享變數來互動，但是按照這個辦法，在一些場合，使用共享變數也是可取的。很多平臺上有較為原生的共享變數支援，到底用那種實現比較好，就見仁見智了。另外利用協程和通道，可以還實現各種常見的併發資料結構，如鎖等等，就不一一贅述。

協程洩漏

協程和記憶體一樣，是系統的資源。對於記憶體，有自動垃圾回收。但是對於協程，沒有相應的回收機制。會不會若干年後，協程普及了，協程洩漏和記憶體洩漏一樣成為程式設計師永遠的痛呢？一般而言，協程執行結束後就會銷燬。協程也會佔用記憶體，如果發生協程洩漏，影響和記憶體洩漏一樣嚴重。輕則拖慢程式，重則壓垮機器。

C和C++都是沒有自動記憶體回收的程式設計語言，但只要有良好的程式設計習慣，就能解決規避問題。對於協程是一樣的，只要有好習慣就可以了。

只有兩種情況會導致協程無法結束。一種情況是協程想從一個通道讀資料，但無人往這個通道寫入資料，或許這個通道已經被遺忘了。還有一種情況是程想往一個通道寫資料，可是由於無人監聽這個通道，該協程將永遠無法向下執行。下面分別討論如何避免這兩種情況。

對於協程想從一個通道讀資料，但無人往這個通道寫入資料這種情況。解決的辦法很簡單，加入超時機制。對於有不確定會不會返回的情況，必須加入超時，避免出現永久等待。另外不一定要使用定時器才能終止協程。也可以對外暴露一個退出提醒通道。任何其他協程都可以通過該通道來提醒這個協程終止。

對於協程想往一個通道寫資料，但通道阻塞無法寫入這種情況。解決的辦法也很簡單，就是給通道加緩衝。但前提是這個通道只會接收到固定數目的寫入。比方說，已知一個通道最多隻會接收N次資料，那麼就將這個通道的緩衝設定為N。那麼該通道將永遠不會堵塞，協程自然也不會洩漏。也可以將其緩衝設定為無限，不過這樣就要承擔記憶體洩漏的風險了。等協程執行完畢後，這部分通道記憶體將會失去引用，會被自動垃圾回收掉。

func never_leak(ch chan int) {
    //初始化timeout，緩衝為1
    timeout := make(chan bool, 1)
    //啟動timeout協程，由於快取為1，不可能洩露
    go func() {
        time.Sleep(1 * time.Second)
        timeout <- true
    }()
    //監聽通道，由於設有超時，不可能洩露
    select {
    case <-ch:
        // a read from ch has occurred
    case <-timeout:
        // the read from ch has timed out
    }
}

上面是個避免洩漏例子。使用超時避免讀堵塞，使用緩衝避免寫堵塞。

和記憶體裡面的物件一樣，對於長期存在的協程，我們不用擔心洩漏問題。一是長期存在，二是數量較少。要警惕的只有那些被臨時建立的協程，這些協程數量大且生命週期短，往往是在迴圈中建立的，要應用前面提到的辦法，避免洩漏發生。協程也是把雙刃劍，如果出問題，不但沒能提高程式效能，反而會讓程式崩潰。但就像記憶體一樣，同樣有洩漏的風險，但越用越溜了。

併發模式之實現

在併發程式設計大行其道的今天，對協程和通道的支援成為各個平臺比不可少的一部分。雖然各家有各家的叫法，但都能滿足協程的基本要求—併發執行和可大量建立。筆者對他們的實現方式總結了一下。

下面列舉一些已經支援協程的常見的語言和平臺。

GoLang 和Scala作為最新的語言，一出生就有完善的基於協程併發功能。Erlang最為老資格的併發程式語言，返老還童。其他二線語言則幾乎全部在新的版本中加入了協程。

令人驚奇的是C/C++和Java這三個世界上最主流的平臺沒有在對協程提供語言級別的原生支援。他們都揹負著厚重的歷史，無法改變，也無需改變。但他們還有其他的辦法使用協程。

Java平臺有很多方法實現協程：

• 修改虛擬機器：對JVM打補丁來實現協程，這樣的實現效果好，但是失去了跨平臺的好處
• 修改位元組碼：在編譯完成後增強位元組碼，或者使用新的JVM語言。稍稍增加了編譯的難度。
• 使用JNI：在Jar包中使用JNI，這樣易於使用，但是不能跨平臺。
• 使用執行緒模擬協程：使協程重量級，完全依賴JVM的執行緒實現。

其中修改位元組碼的方式比較常見。因為這樣的實現辦法，可以平衡效能和移植性。最具代表性的JVM語言Scala就能很好的支援協程併發。流行的Java Actor模型類庫akka也是用修改位元組碼的方式實現的協程。

對於C語言，協程和執行緒一樣。可以使用各種各樣的系統呼叫來實現。協程作為一個比較高階的概念，實現方式實在太多，就不討論了。比較主流的實現有libpcl, coro,lthread等等。

對於C++，有Boost實現，還有一些其他開源庫。還有一門名為μC++語言，在C++基礎上提供了併發擴充套件。

可見這種程式設計模型在眾多的語言平臺中已經得到了廣泛的支援，不再小眾。如果想使用的話，隨時可以加到自己的工具箱中。

結語

本文探討了一個極其簡潔的併發模型。在只有協程和通道這兩個基本元件的情況下。可以提供豐富的功能，解決形形色色實際問題。而且這個模型已經被廣泛的實現，成為潮流。相信這種併發模型的功能遠遠不及此，一定也會有更多更簡潔的用法出現。或許未來CPU核心數目將和人腦神經元數目一樣多，到那個時候，我們又要重新思考併發模型了。