Goroutine + Channel 實踐

背景

在最近開發的項目中，後端需要編寫許多提供HTTP接口的API，另外技術選型相對寬鬆，因此選擇Golang + Beego框架進行開發。之所以選擇Golang，主要是考慮到開發的模塊，都需要接受瞬時大併發、請求需要經歷多個步驟、處理時間較長、無法同步立即返回結果的場景，Golang的goroutine以及channel所提供的語言層級的特性，正好可以滿足這方面的需要。

goroutine不同於thread，threads是操作系統中的對於一個獨立運行實例的描述，不同操作系統，對於thread的實現也不盡相同；但是，操作系統並不知道goroutine的存在，goroutine的調度是有Golang運行時進行管理的。啟動thread雖然比process所需的資源要少，但是多個thread之間的上下文切換仍然是需要大量的工作的（寄存器/Program Count/Stack Pointer/...），Golang有自己的調度器，許多goroutine的數據都是共享的，因此goroutine之間的切換會快很多，啟動goroutine所耗費的資源也很少，一個Golang程序同時存在幾百個goroutine是很正常的。

channel，即“管道”，是用來傳遞數據（叫消息更為合適）的一個數據結構，即可以從channel裡面塞數據，也可以從中獲取數據。channel本身並沒有什麼神奇的地方，但是channel加上了goroutine，就形成了一種既簡單又強大的請求處理模型，即N個工作goroutine將處理的中間結果或者最終結果放入一個channel，另外有M個工作goroutine從這個channel拿數據，再進行進一步加工，通過組合這種過程，從而勝任各種複雜的業務模型。

模型

自己在實踐的過程中，產生了幾種通過goroutine + channel實現的工作模型，本文分別對這些模型進行介紹。

V0.1: go關鍵字

直接加上go關鍵字，就可以讓一個函數脫離原先的主函數獨立運行，即主函數直接繼續進行剩下的操作，而不需要等待某個十分耗時的操作完成。比如我們在寫一個服務模塊，接收到前端請求之後，然後去做一個比較耗時的任務。比如下面這個：

<code>func (m *SomeController) PorcessSomeTask() {
    var task models.Task
    if err := task.Parse(m.Ctx.Request); err != nil {
        m.Data["json"] = err 
        m.ServeJson()
        return
    }
    task.Process()
    m.ServeJson()/<code>

如果Process函數需要耗費大量時間的話，這個請求就會被block住。有時候，前端只需要發出一個請求給後端，並且不需要後端立即所處響應。遇到這樣的需求，直接在耗時的函數前面加上go關鍵字就可以將請求之間返回給前端了，保證了體驗。

<code>func (m *SomeController) PorcessSomeTask() {
    var task models.Task
    if err := task.Parse(m.Ctx.Request); err != nil {
        m.Data["json"] = err 
        m.ServeJson()
        return
    }
    go task.Process()
    m.ServeJson()
/<code>

不過，這種做法也是有許多限制的。比如：

• 只能在前端不需要立即得到後端處理的結果的情況下
• 這種請求的頻率不應該很大，因為目前的做法沒有控制併發

V0.2: 併發控制

上一個方案有一個缺點就是無法控制併發，如果這一類請求同一個時間段有很多的話，每一個請求都啟動一個goroutine，如果每個goroutine中還需要使用其他系統資源，消耗將是不可控的。

遇到這種情況，一個解決方案是：將請求都轉發給一個channel，然後初始化多個goroutine讀取這個channel中的內容，並進行處理。假設我們可以新建一個全局的channel

<code>var TASK_CHANNEL = make(chan models.Task)
/<code>

然後，啟動多個goroutine：

<code>for i := 0; i < WORKER_NUM; i ++ {
    go func() {
        for {
            select {
            case task :=                 task.Process()
            }
        }
    } ()
}
/<code>

服務端接收到請求之後，將任務傳入channel中即可：

<code>func (m *SomeController) PorcessSomeTask() {
    var task models.Task
    if err := task.Parse(m.Ctx.Request); err != nil {
        m.Data["json"] = err 
        m.ServeJson()
        return
    }
    //go task.Process()
    TASK_CHANNEL     m.ServeJson()
}
/<code> 

這樣一來，這個操作的併發度就可以通過WORKER_NUM來控制了。

V0.3: 處理channel滿的情況

不過，上面方案有一個bug：那就是channel初始化時是沒有設置長度的，因此當所有WORKER_NUM個goroutine都正在處理請求時，再有請求過來的話，仍然會出現被block的情況，而且會比沒有經過優化的方案還要慢（因為需要等某一個goroutine結束時才能處理它）。因此，需要在channel初始化時增加一個長度：

<code>var TASK_CHANNEL = make(chan models.Task, TASK_CHANNEL_LEN)
/<code>

這樣一來，我們將TASK_CHANNEL_LEN設置得足夠大，請求就可以同時接收TASK_CHANNEL_LEN個請求而不用擔心被block。不過，這其實還是有問題的：那如果真的同時有大於TASK_CHANNEL_LEN個請求過來呢？一方面，這就應該算是架構方面的問題了，可以通過對模塊進行擴容等操作進行解決。另一方面，模塊本身也要考慮如何進行“優雅降級了”。遇到這種情況，我們應該希望模塊能夠及時告知調用方，“我已經達到處理極限了，無法給你處理請求了”。其實，這種需求，可以很簡單的在Golang中實現：如果channel發送以及接收操作在select語句中執行並且發生阻塞，default語句就會立即執行

<code>select {
case TASK_CHANNEL     //do nothing
default:
    //warnning!
    return fmt.Errorf("TASK_CHANNEL is full!")
}
//...
/<code>

V0.4: 接收發送給channel之後返回的結果

如果處理程序比較複雜的時候，通常都會出現在一個goroutine中，還會發送一些中間處理的結果發送給其他goroutine去做，經過多道“工序”才能最終將結果產出。

那麼，我們既需要把某一箇中間結果發送給某個channel，也要能獲取到處理這次請求的結果。解決的方法是：將一個channel實例包含在請求中，goroutine處理完成後將結果寫回這個channel。

<code>type TaskResponse struct {
    //...
}

type Task struct {
    TaskParameter   SomeStruct
    ResChan         *chan TaskResponse
}

//...

task := Task {
    TaskParameter   : xxx,
    ResChan         : make(chan TaskResponse),
}

TASK_CHANNEL res := //... 

/<code>

（這邊可能會有疑問：為什麼不把一個複雜的任務都放在一個goroutine中依次的執行呢？是因為這裡需要考慮到不同子任務，所消耗的系統資源不盡相同，有些是CPU集中的，有些是IO集中的，所以需要對這些子任務設置不同的併發數，因此需要經由不同的channel + goroutine去完成。）

V0.5: 等待一組goroutine的返回

將任務經過分組，交由不同的goroutine進行處理，最終再將每個goroutine處理的結果進行合併，這個是比較常見的處理流程。這裡需要用到WaitGroup來對一組goroutine進行同步。一般的處理流程如下：

<code>var wg sync.WaitGroup
for i := 0; i < someLen; i ++ {
    wg.Add(1)
    go func(t Task) {
        defer wg.Done()
        //對某一段子任務進行處理
    } (tasks[i])
}

wg.Wait()
//處理剩下的工作
/<code>

V0.6: 超時機制

即使是複雜、耗時的任務，也必須設置超時時間。一方面可能是業務對此有時限要求（用戶必須在XX分鐘內看到結果），另一方面模塊本身也不能都消耗在一直無法結束的任務上，使得其他請求無法得到正常處理。因此，也需要對處理流程增加超時機制。

我一般設置超時的方案是：和之前提到的“接收發送給channel之後返回的結果”結合起來，在等待返回channel的外層添加select，並在其中通過time.After()來判斷超時。

<code>task := Task {
    TaskParameter   : xxx,
    ResChan         : make(chan TaskResponse),
}

select {
case res :=     //...
case     //處理超時
}
/<code>

V0.7: 廣播機制

既然有了超時機制，那也需要一種機制來告知其他goroutine結束手上正在做的事情並退出。很明顯，還是需要利用channel來進行交流，第一個想到的肯定就是向某一個chan發送一個struct即可。比如執行任務的goroutine在參數中，增加一個chan struct{}類型的參數，當接收到該channel的消息時，就退出任務。但是，還需要解決兩個問題：

1. 怎樣能在執行任務的同時去接收這個消息呢？
2. 如何通知所有的goroutine？

對於第一個問題，比較優雅的作法是：使用另外一個channel作為函數d輸出，再加上select，就可以一邊輸出結果，一邊接收退出信號了。

另一方面，對於同時有未知數目個執行goroutine的情況，一次次調用done 當關閉一個channel時，所有因為接收該channel而阻塞的語句會立即返回。示例代碼如下：

<code>// 執行方
func doTask(done     out := make(chan Result)
    go func() {
        // close 是為了讓調用方的range能夠正常退出
        defer close(out)
        for t := range tasks {
            select {
            case result             case                 return
            }
        }
    }()

    return out
}

// 調用方
func Process(tasks     done := make(chan struct{})
    out := doTask(done, tasks)

    go func() {
                //done 
        //通知所有的執行goroutine退出
        close(done)
    }()

    // 因為goroutine執行完畢，或者超時，導致out被close，range退出
    for res := range out {
        fmt.Println(res)
        //...
    }
}/<code>

閱讀更多 昭昭的生活樂趣 的文章

關鍵字: 數據結構 實踐 併發控制