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

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

性能測試報告：johng.cn/grpool-perf…

方法列表

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個任務。

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數量以及待處理的任務數量。

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

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關鍵字

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)、使用臨時變量

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的值。

 

https://segmentfault.com/q/1010000013455064

首先, 我以爲要不要限制 goroutine 數量得看瓶頸或關鍵問題點.

其次, 全局單例 Grpool 頗有問題, 如下是 sync.WaitGroup 官方文檔

func (wg *WaitGroup) Add(delta int)

Add adds delta, which may be negative, to the WaitGroup counter.
If the counter becomes zero, all goroutines blocked on Wait are released.
If the counter goes negative, Add panics.

這意味着時間相近的多個請求和可能會綁在一塊兒, 變成多個 pool.Wait() 同時返回.

如下是資源池的簡單示例, 但願有益

package main

type Pool struct {
    q chan int
}

func NewPool(max_size int) *Pool {
    return &Pool{q: make(chan int, max_size)}
}

func (p *Pool) Acquire() {
    p.q <- 1
}

func (p *Pool) Release() {
    <-p.q
}

func main() {
    p := NewPool(10)
    p.Acquire()
    // ...
    p.Release()
}