gf框架之grpool – 高性能的goroutine池

文章來源：http://gf.johng.cn/504458

grpool

Go語言中的goroutine雖然相對於系統線程來講比較輕量級，可是在高併發量下的goroutine頻繁建立和銷燬對於性能損耗以及GC來講壓力也不小。充分將goroutine複用，減小goroutine的建立/銷燬的性能損耗，這即是grpool對goroutine進行池化封裝的目的。例如，針對於100W個執行任務，使用goroutine的話須要不停建立並銷燬100W個goroutine，而使用grpool也許底層只須要幾千個goroutine便能充分複用地執行完成全部任務。經測試，在高併發下grpool的性能比原生的goroutine高出幾倍到數百倍！而且隨之也極大地下降了內存使用率。

性能測試報告：http://johng.cn/grpool-perfor...

方法列表

func Add(f func())
func Jobs() int
func SetExpire(expire int)
func SetSize(size int)
func Size() int
type Pool
    func New(expire int, sizes ...int) *Pool
    func (p *Pool) Add(f func())
    func (p *Pool) Close()
    func (p *Pool) Jobs() int
    func (p *Pool) SetExpire(expire int)
    func (p *Pool) SetSize(size int)
    func (p *Pool) Size() int

經過grpool.New方法建立一個goroutine池，並給定池中goroutine的有效時間，單位爲秒，第二個參數爲非必需參數，用於限定池中的工做goroutine數量，默認爲不限制。須要注意的是，任務能夠不停地往池中添加，沒有限制，可是工做的goroutine是能夠作限制的。咱們能夠經過Size()方法查詢當前的工做goroutine數量，使用Jobs()方法查詢當前池中待處理的任務數量。

同時，池的大小和goroutine有效期能夠經過SetSize和SetExpire方法在運行時進行動態改變。這一點特性使得協程池的管理更加靈活，可是也增長了必定的維護風險，由於你沒法得知池的屬性在哪一個地方被執行了修改。

同時，爲便於使用，grpool包提供了默認的goroutine池，直接經過grpool.Add便可往默認的池中添加任務，任務參數必須是一個 func() 類型的函數/方法。

使用示例

一、使用默認的goroutine池，限制10個工做goroutine執行1000個任務。

https://gitee.com/johng/gf/bl...

package main

import (
    "time"
    "fmt"
    "gitee.com/johng/gf/g/os/gtime"
    "gitee.com/johng/gf/g/os/grpool"
)

func job() {
    time.Sleep(1*time.Second)
}

func main() {
    grpool.SetSize(10)
    for i := 0; i < 1000; i++ {
        grpool.Add(job)
    }
    gtime.SetInterval(2*time.Second, func() bool {
        fmt.Println("size:", grpool.Size())
        fmt.Println("jobs:", grpool.Jobs())
        return true
    })
    select {}
}

這段程序中的任務函數的功能是sleep 1秒鐘，這樣便能充分展現出goroutine數量限制功能。其中，咱們使用了gtime.SetInterval定時器每隔2秒鐘打印出當前默認池中的工做goroutine數量以及待處理的任務數量。

二、咱們再來看一個新手常常容易出錯的例子

https://gitee.com/johng/gf/bl...

package main

import (
    "fmt"
    "sync"
    "gitee.com/johng/gf/g/os/grpool"
)

func main() {
    wg := sync.WaitGroup{}
    for i := 0; i < 10; i++ {
        wg.Add(1)
        grpool.Add(func() {
            fmt.Println(i)
            wg.Done()
        })
    }
    wg.Wait()
}

咱們這段代碼的目的是要順序地打印出0-9，然而運行後卻輸出：

10
10
10
10
10
10
10
10
10
10

爲何呢？這裏的執行結果不管是採用go關鍵字來執行仍是grpool來執行都是如此。緣由是，對於異步線程/協程來說，函數進行進行異步執行註冊時，該函數並未真正開始執行(註冊時只在goroutine的棧中保存了變量i的內存地址)，而一旦開始執行時函數纔會去讀取變量i的值，而這個時候變量i的值已經自增到了10。
清楚緣由以後，改進方案也很簡單了，就是在註冊異步執行函數的時候，把當時變量i的值也一併傳遞獲取；或者把當前變量i的值賦值給一個不會改變的臨時變量，在函數中使用該臨時變量而不是直接使用變量i。

改進後的示例代碼以下：

1)、使用go關鍵字

https://gitee.com/johng/gf/bl...

package main

import (
    "fmt"
    "sync"
)

func main() {
    wg := sync.WaitGroup{}
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(v int){
            fmt.Println(v)
            wg.Done()
        }(i)
    }
    wg.Wait()
}

執行後，輸出結果爲：

9
0
1
2
3
4
5
6
7
8

注意，異步執行時並不會保證按照函數註冊時的順序執行，如下同理。

2)、使用臨時變量

https://gitee.com/johng/gf/bl...

package main

import (
    "fmt"
    "sync"
    "gitee.com/johng/gf/g/os/grpool"
)

func main() {
    wg := sync.WaitGroup{}
    for i := 0; i < 10; i++ {
        wg.Add(1)
        v := i
        grpool.Add(func() {
            fmt.Println(v)
            wg.Done()
        })
    }
    wg.Wait()
}

執行後，輸出結果爲：

9
0
1
2
3
4
5
6
7
8

這裏能夠看到，使用grpool進行任務註冊時，只能使用func()類型的參數，所以沒法在任務註冊時把變量i的值註冊進去，所以只能採用臨時變量的形式來傳遞當前變量i的值。