Golang語言實現(xiàn)的流水線模型

文/王曉峰

（中國電信股份有限公司海南分公司 海南省?？谑?570105）

眾所周知，阿里雙11 的實時交易額使用了Flink 技術(shù)，高峰時每秒處理交易數(shù)據(jù)4.7 億個。Flink 這么高的處理能力是得益于底層采用了流水線模型，流水線（pipeline）模型是對現(xiàn)實世界的流水線作業(yè)方式的完美模仿。本文就由淺入深的用Golang 語言實現(xiàn)一個簡單地高并發(fā)流水線。Golang 語言實現(xiàn)的流水線（pipeline）模型，可以允許開發(fā)者以優(yōu)雅的方式構(gòu)建出類似Linux Pipe 的高效利用I/O 和多核CPU 優(yōu)勢的數(shù)據(jù)流水線（data pipeline）。

現(xiàn)實中流水線就是一個從傳送帶取出物品，進(jìn)行處理后，再放到下一個傳送帶三個步驟。而在Golang 語言里，傳送帶就是channel，中間每個環(huán)節(jié)就是從channel里取數(shù)據(jù)，然后進(jìn)行數(shù)據(jù)操作，再把操作結(jié)果放入下一個channel 的三個步驟。因此中間每個操作的輸入是一個channel，輸出也是一個channel。

流水線的最開始的環(huán)節(jié)只是把需要處理的數(shù)據(jù)放入channel，因此沒有輸入，通常稱最開始的環(huán)節(jié)叫做“生產(chǎn)者”；最后一個環(huán)節(jié)是收集處理完的數(shù)據(jù)，因此沒有輸出，通常把最后一個環(huán)節(jié)叫做“消費者”。

從而流水線可看做，生產(chǎn)者把數(shù)據(jù)送入通道（channel），經(jīng)過一系列處理，消費者從通道（channel）得到處理后的數(shù)據(jù)。

1 用Golang語言實現(xiàn)一個鏈?zhǔn)讲僮鞯牧魉€

1.1 簡單流水線需求

假設(shè)需要對一組整數(shù)進(jìn)行處理：要求對每個數(shù)進(jìn)行平方操作后，再對每個數(shù)加10，最后打印出處理后的數(shù)據(jù)。

1.2 簡單流水線設(shè)計思路

根據(jù)前面描述的需求，我們分析首先生產(chǎn)者是產(chǎn)生一組數(shù)據(jù)，消費者是最后對數(shù)據(jù)的打印，中間處理環(huán)節(jié)是對每個數(shù)進(jìn)行平方操作，和對每個數(shù)加10。每個環(huán)節(jié)都是把上一環(huán)節(jié)的輸出通道當(dāng)做本環(huán)節(jié)的輸入（為了保證通道不會阻塞，必須給通道設(shè)置一個合適大小的緩存），因此這里需要分四步進(jìn)行。

（1）生產(chǎn)者輸出數(shù)據(jù)到通道，當(dāng)輸出完畢后，關(guān)閉輸出通道。

（2）從生產(chǎn)者輸出通道上取數(shù)據(jù)，進(jìn)行平方操作后，把結(jié)果放入通道，把通道輸出，同樣輸出完畢后，關(guān)閉輸出通道。

（3）從上一環(huán)節(jié)輸出的通道取數(shù)據(jù)，進(jìn)行加10 操作后，把結(jié)果放入通道，把通道輸出，同樣輸出完畢后，關(guān)閉輸出通道。

（4）最后消費者從上一環(huán)節(jié)輸出的通道中，取出數(shù)據(jù)進(jìn)行打印。

1.3 鏈?zhǔn)秸{(diào)用優(yōu)化

為了使得代碼的可讀性、可維護(hù)性和可變更性變強，進(jìn)行鏈?zhǔn)絻?yōu)化。我們只需把所有函數(shù)封裝成通用通道類型下的方法，再實例化一個通用通道對象，即可把流水線模型的主程序調(diào)用轉(zhuǎn)換為對象方式的鏈?zhǔn)剑ㄒ卜Q流式）調(diào)用。

1.4 優(yōu)化成帶扇入、扇出并行的流水線

在現(xiàn)實世界里，流水線的中間一些環(huán)節(jié)可能需要并行操作，比如以汽車生產(chǎn)流水線為例，在安裝輪子這個環(huán)節(jié)時，是同時進(jìn)行4個輪子的安裝，在四個輪子都安裝好后，再進(jìn)入下一工作環(huán)節(jié)。這個過程就存在了任務(wù)的分發(fā)、和并行任務(wù)結(jié)果的匯聚。在Golang語言的流水線里，就稱作扇入（FAN-IN）、扇出（FAN-OUT）。

扇入（FAN-IN），就是把多個通道上的數(shù)據(jù)匯集到一條通道上，是一種收斂的模式，主要用來收集處理的結(jié)果。

圖1

圖2

扇出（FAN-OUT），多個多個goroutines 共享一個通道，簡單講就是多個方法（或函數(shù)）從一個通道上讀取數(shù)據(jù)，一般用于分發(fā)數(shù)據(jù)到多個協(xié)程并發(fā)處理。如圖1 所示。

為了并發(fā)執(zhí)行，我們可在第二個環(huán)節(jié)，計算平方方法實現(xiàn)3 個goroutine、第三個環(huán)節(jié)，加10 運算方法，實現(xiàn)2 個goroutine。因此在第二和第三環(huán)節(jié)要進(jìn)行扇入和扇出操作。第一和第四環(huán)節(jié)保持不變。

1.5 并行的流水線設(shè)計思路

第一和第四環(huán)節(jié)保持不變。第二和第三個環(huán)節(jié)，需要進(jìn)行扇出處理。把第一環(huán)節(jié)輸出的通道扇出到多個平方計算的協(xié)程上，每個協(xié)程把結(jié)果輸出到各自的輸出通道，然后再把三個通道扇入到一個通道作為第二環(huán)節(jié)的輸出通道。

同樣第三環(huán)節(jié)把第二環(huán)節(jié)輸出的通道扇出到多個個加法協(xié)程上，每個協(xié)程把結(jié)果輸出到各自的輸出通道，然后再把二個通道扇入到一個通道作為第三環(huán)節(jié)的輸出通道。如圖2 所示。

1.6 帶扇入、扇出的的流水線代碼

保持原先定義的通用的通道類型，chan int 的通用類型MyChan type MyChan chan int

在第二和第三環(huán)節(jié)進(jìn)行扇入、扇出修改，修改后的代碼如下：

（1）第一環(huán)節(jié)保持不變，生產(chǎn)者輸出數(shù)據(jù)到通道。

（2）第二環(huán)節(jié)需要進(jìn)行扇入扇出，設(shè)置需要并發(fā)計算平方的數(shù)為threadNum，先進(jìn)性扇出操作：從生產(chǎn)者輸出通道上取數(shù)據(jù)，進(jìn)行threadNum 個協(xié)程平方操作后，把結(jié)果分別放入通道。然后進(jìn)行扇入操作：把多個輸出通道扇入到一個通道（out），并返回這個通道out。

（3）第三環(huán)節(jié)仍然需要進(jìn)行扇入扇出，先進(jìn)性扇出操作：定義并發(fā)函數(shù)addten（threadNum int），從上一環(huán)節(jié)按照threadNum數(shù)量扇出到addten（threadNum int），進(jìn)行加10 操作。然后再進(jìn)行扇入操作：把threadNum 個輸出通道扇入到out 通道。最后把輸出的通道作為返回值返回。

（4）最后消費者從上一環(huán)節(jié)輸出的通道中，取出數(shù)據(jù)進(jìn)行打印，同樣把上一環(huán)節(jié)的通道作為函數(shù)的參數(shù)，但不用再返回通道。

2 結(jié)論

Golang語言的流水線(Pipeline)模型是對現(xiàn)實世界的完美模仿，流水線模型邏輯清晰，易于實現(xiàn)，各環(huán)節(jié)并發(fā)可調(diào)節(jié)，適合大量數(shù)據(jù)的處理工作，流水線模型結(jié)合鏈?zhǔn)秸{(diào)用模式是代碼可讀性大大提高，有利于維護(hù)和修改。Golang 的流水線結(jié)合Contex 可實現(xiàn)對流水線的控制，當(dāng)不需要生產(chǎn)者發(fā)送時數(shù)據(jù)時，簡單的一個cancel（）就能終止數(shù)據(jù)發(fā)送，并關(guān)閉所有通道，關(guān)于Contex 又是一個很大的話題，本文就不再論述了。

流水線涉及到大量并發(fā)和通道的操作，一旦出現(xiàn)錯誤處理就比較復(fù)雜，因此建議力求把流水線的每個環(huán)節(jié)設(shè)計完善，做好測試，盡量確保每個環(huán)節(jié)邏輯清晰。當(dāng)上一環(huán)節(jié)發(fā)送操作結(jié)束時，每個環(huán)節(jié)都應(yīng)該及時關(guān)閉自己的輸出通道。同時也盡量提供足夠大的緩沖保存發(fā)送者發(fā)送的數(shù)據(jù)。