聊聊 Go 流水線編程模式
文末本文轉載自微信公眾號「Golang技術分享」，作者機器鈴砍菜刀 。轉載本文請聯系Golang技術分享公眾號。
流水線工作模型在工業領域內十分常見，它將工作流程分為多個環節，每個環節根據工作強度安排合適的人員數量。良好的流水線設計盡量讓各環節的流通率平衡，最大化提高產能效率。
Go 是一門實用性語言，流水線工作模型與 Go 融合地非常融洽，只不過我們一般使用另一個名詞來表示流水線：pipeline。
pipeline
pipeline 由多個環節組成，具體在 Go 中，環節之間通過 channel 通信，同一個環節任務可以由多個 goroutine 來同時處理。
pipeline
pipeline 的核心是數據，通過 channel 來保證數據流動，每個環節的數據處理由 goroutine 完成。
除了開始環節和結束環節，每個環節都有任意數量的輸入 channel 和輸出 channel。開始環節被稱為發送者或生產者，結束環節被稱為接收者或消費者。
下面我們來看一個簡單的 pipeline 例子，分為三個環節。
第一個環節，generate 函數：它充當生產者角色，將數據寫入 channel，并把該 channel 返回。當所有數據寫入完畢，關閉 channel。
- func generate(nums ...int) <-chan int {
- out := make(chan int)
- go func() {
- for _, n := range nums {
- out <- n
- }
- close(out)
- }()
- return out
- }
第二個環節，square 函數：它是數據處理的角色，從開始環節中的 channel 取出數據，計算平方，將結果寫入新的 channel ，并把該新的 channel 返回。當所有數據計算完畢，關閉該新 channel。
- func square(in <-chan int) <-chan int {
- out := make(chan int)
- go func() {
- for n := range in {
- out <- n * n
- }
- close(out)
- }()
- return out
- }
main 函數負責編排整個 pipeline ，并充當消費者角色：讀取第二個環節的 channel 數據，打印出來。
- func main() {
- // Set up the pipeline.
- c := generate(2, 3)
- out := square(c)
- // Consume the output.
- for n := range out {
- fmt.Println(n)
- }
- }
Fan-out,fan-in
在上述例子中，環節之間通過非緩沖的 channel 傳遞數據，節點中的數據都是單個 goroutine 處理與消費。
這種工作模式并不高效，會讓整個流水線的效率取決于最慢的環節。因為每個環節中的任務量是不同的，這意味著我們需要的機器資源是存在差異的。任務量小的環節，盡量占有少量的機器資源，任務量重的環節，需要更多線程并行處理。
以汽車組裝為例，我們可以將組裝輪胎的工作分發給 4 個人一起干，當輪胎組裝完畢之后，再交由剩下的環節。
多個 goroutine 可以從同一個 channel 讀取數據，直到該通道關閉，這稱為 fan-out(扇出)。
這個稱呼比較形象，它將數據進行分散，所以被稱為扇出。扇出是一種分發任務的模式。
fan-out
單個 goroutine 可以從多個輸入 channel 中讀取數據，直到所有輸入都關閉。具體做法是將輸入 channel 多路復用到同一個 channel 上，當所有輸入 channel 都關閉時，該 channel 也關閉，這稱為 fan-in(扇入)。
它將數據進行聚合，所以被稱為扇入。扇入是一種整合任務結果的模式。
fan-in
在汽車組裝的例子中，分發輪胎任務給每個人是 Fan-out，合并輪胎組裝結果就是 Fan-in。
channel 的多路復用
扇出的編碼模型比較簡單，本文不多研究，我們提供一個扇入編程示例。
創建一個生成器函數 generate，通過 interval 參數控制消息生成頻率。生成器返回消息 channel mc與停止 channel sc，停止 channel 用于停止生成器任務。
- func generate(message string, interval time.Duration) (chan string, chan struct{}) {
- mc := make(chan string)
- sc := make(chan struct{})
- go func() {
- defer func() {
- close(sc)
- }()
- for {
- select {
- case <-sc:
- return
- default:
- time.Sleep(interval)
- mc <- message
- }
- }
- }()
- return mc, sc
- }
stopGenerating 函數通過通過向 sc 中傳入空結構體，通知 generate退出，調用 close(mc) 關閉消息 channel
- func stopGenerating(mc chan string, sc chan struct{}) {
- sc <- struct{}{}
- close(mc)
- }
多路復用函數 multiplex 創建并返回整合消息 channel 和控制并發的 wg。
- func multiplex(mcs ...chan string) (chan string, *sync.WaitGroup) {
- mmc := make(chan string)
- wg := &sync.WaitGroup{}
- for _, mc := range mcs {
- wg.Add(1)
- go func(mc chan string, wg *sync.WaitGroup) {
- defer wg.Done()
- for m := range mc {
- mmc <- m
- }
- }(mc, wg)
- }
- return mmc, wg
- }
在 main 函數中，創建兩個消息 channel 并復用它們生成 mmc ，打印來自 mmc 的每條消息。另外，我們還實現了接收系統斷信號(終端上執行 CTRL+C 即可發送中斷信號)的優雅的關閉機制。
- func main() {
- // create two sample message and stop channels
- mc1, sc1 := generate("message from generator 1", 200*time.Millisecond)
- mc2, sc2 := generate("message from generator 2", 300*time.Millisecond)
- // multiplex message channels
- mmc, wg1 := multiplex(mc1, mc2)
- // create errs channel for graceful shutdown
- errs := make(chan error)
- // wait for interrupt or terminate signal
- go func() {
- sc := make(chan os.Signal, 1)
- signal.Notify(sc, syscall.SIGINT, syscall.SIGTERM)
- errs <- fmt.Errorf("%s signal received", <-sc)
- }()
- // wait for multiplexed messages
- wg2 := &sync.WaitGroup{}
- wg2.Add(1)
- go func() {
- defer wg2.Done()
- for m := range mmc {
- fmt.Println(m)
- }
- }()
- // wait for errors
- if err := <-errs; err != nil {
- fmt.Println(err.Error())
- }
- // stop generators
- stopGenerating(mc1, sc1)
- stopGenerating(mc2, sc2)
- wg1.Wait()
- // close multiplexed messages channel
- close(mmc)
- wg2.Wait()
- }
總結
本文簡單介紹了流水線編程模式，它和我們熟悉的生產者-消費者模式非常相似。
具體到 Go 編程實踐中，pipeline 將數據流分為多個環節，channel 用于數據流動，goroutine 用于處理數據。fan-out 用于分發任務，fan-in 用于數據整合，通過 FAN 模式可以讓流水線更好地并發。
當然，還有些細節需要注意，例如停止通知機制，可參照本文 channel 的多路復用章節示例中的 stopGenerating 函數;如何通過 sync.WaitGroup 做好并發控制，這些都是需要讀者在實際編碼中去體會掌握的。
參考
Go Concurrency Patterns: Pipelines and cancellation：https://go.dev/blog/pipelines
Multiplexing Channels In Go：https://medium.com/@ermanimer/multiplexing-channels-in-go-a7dccdcc4134
