建站
咨詢
售后
建站咨詢
新聞中心
基本介紹
GO語言中的?goroutine?雖然相對于系統(tǒng)線程來說比較輕量級（初始棧大小僅2KB），（并且支持動態(tài)擴容），而正常采用java，c++等語言啟用的線程一般都是內核態(tài)的占用的內存資源一般在4m左右，而假設我們的服務器CPU內存為4G，那么很明顯才用的內核態(tài)線程的并發(fā)總數量也就是1024個，相反查看一下Go語言的協(xié)程則可以達到4*1024*1024/2=200w.這么一看就明白了為什么Go語言天生支持高并發(fā)。但是在高并發(fā)量下的?goroutine?頻繁創(chuàng)建和銷毀對于性能損耗以及GC來說壓力也不小。充分將?goroutine?復用，減少?goroutine?的創(chuàng)建/銷毀的性能損耗，這便是?grpool?對?goroutine?進行池化封裝的目的。例如，針對于100W個執(zhí)行任務，使用?goroutine?的話需要不停創(chuàng)建并銷毀100W個?goroutine?，而使用?grpool?也許底層只需要幾萬個?goroutine?便能充分復用地執(zhí)行完成所有任務。
創(chuàng)新互聯(lián)建站致力于成都網站設計、成都網站制作,成都網站設計,集團網站建設等服務標準化，推過標準化降低中小企業(yè)的建站的成本，并持續(xù)提升建站的定制化服務水平進行質量交付，讓企業(yè)網站從市場競爭中脫穎而出。 選擇創(chuàng)新互聯(lián)建站，就選擇了安全、穩(wěn)定、美觀的網站建設服務！
經測試，?goroutine?池對于業(yè)務邏輯的執(zhí)行效率（降低執(zhí)行時間/CPU使用率）提升不大，甚至沒有原生的?goroutine?執(zhí)行快速（池化?goroutine?執(zhí)行調度并沒有底層go調度器高效，因為池化?goroutine?的執(zhí)行調度也是基于底層go調度器），但是由于采用了復用的設計，池化后對內存的使用率得到極大的降低。在v2版本中?grpool?也加入了貫穿全局的鏈路追蹤。
概念：
- ?
Pool?：?goroutine?池，用于管理若干可復用的?goroutine?協(xié)程資源； - ?
Worker?：池對象中參與任務執(zhí)行的?goroutine?，一個?Worker?可以執(zhí)行若干個?Job?，直到隊列中再無等待的?Job?； - ?
Job?：添加到池對象的任務隊列中等待執(zhí)行的任務，是一個?func()?的方法，一個?Job?同時只能被一個?Worker?獲取并執(zhí)行；
使用方式：
import "github.com/gogf/gf/v2/os/grpool"
使用場景：管理大量異步任務的場景、需要異步協(xié)程復用的場景、需要降低內存使用率的場景。
接口文檔：
func Add(f func()) error
func Jobs() int
func Size() int
type Pool
func New(limit ...int) *Pool
func (p *Pool) Add(ctx context.Context, f Func) error
func (p *Pool) AddWithRecover(ctx context.Context, userFunc Func, recoverFunc ...func(err error)) error
func (p *Pool) Cap() int
func (p *Pool) Close()
func (p *Pool) IsClosed() bool
func (p *Pool) Jobs() int
func (p *Pool) Size() int通過?grpool.New?方法創(chuàng)建一個?goroutine?池對象，參數?limit?為非必需參數，用于限定池中的工作?goroutine?數量，默認為不限制。需要注意的是，任務可以不停地往池中添加，沒有限制，但是工作的?goroutine?是可以做限制的。我們可以通過?Size()?方法查詢當前的工作?goroutine?數量，使用?Jobs()?方法查詢當前池中待處理的任務數量。
同時，為便于使用，?grpool?包提供了默認的?goroutine?池，默認的池對象不限制?goroutine?數量，直接通過?grpool.Add?即可往默認的池中添加任務，任務參數必須是一個 ?func()?類型的函數/方法。
使用示例
使用默認的goroutine池，限制100個goroutine執(zhí)行1000個任務
package main
import (
"context"
"fmt"
"github.com/gogf/gf/v2/os/gctx"
"github.com/gogf/gf/v2/os/grpool"
"github.com/gogf/gf/v2/os/gtimer"
"time"
)
var (
ctx = gctx.New()
)
func job(ctx context.Context) {
time.Sleep(1*time.Second)
}
func main() {
pool := grpool.New(100)
for i := 0; i < 1000; i++ {
pool.Add(ctx,job)
}
fmt.Println("worker:", pool.Size())
fmt.Println(" jobs:", pool.Jobs())
gtimer.SetInterval(ctx,time.Second, func(ctx context.Context) {
fmt.Println("worker:", pool.Size())
fmt.Println(" jobs:", pool.Jobs())
fmt.Println()
})
select {}
}這段程序中的任務函數的功能是?sleep? 1秒鐘，這樣便能充分展示出?goroutine?數量限制功能。其中，我們使用了?gtime.SetInterval?定時器每隔1秒鐘打印出當前默認池中的工作?goroutine?數量以及待處理的任務數量。
異步傳參：來個新手容易出錯的例子
package main
import (
"context"
"fmt"
"github.com/gogf/gf/v2/os/gctx"
"github.com/gogf/gf/v2/os/grpool"
"sync"
)
var (
ctx = gctx.New()
)
func main() {
wg := sync.WaitGroup{}
for i := 0; i < 10; i++ {
wg.Add(1)
grpool.Add(ctx,func(ctx context.Context) {
fmt.Println(i)
wg.Done()
})
}
wg.Wait()
} 我們這段代碼的目的是要順序地打印出0-9，然而運行后卻輸出：
10
10
10
10
10
10
10
10
10
10為什么呢？這里的執(zhí)行結果無論是采用go關鍵字來執(zhí)行還是?grpool?來執(zhí)行都是如此。原因是，對于異步線程/協(xié)程來講，函數進行異步執(zhí)行注冊時，該函數并未真正開始執(zhí)行(注冊時只在?goroutine?的棧中保存了變量?i?的內存地址)，而一旦開始執(zhí)行時函數才會去讀取變量?i?的值，而這個時候變量?i?的值已經自增到了10。 清楚原因之后，改進方案也很簡單了，就是在注冊異步執(zhí)行函數的時候，把當時變量i的值也一并傳遞獲??；或者把當前變量i的值賦值給一個不會改變的臨時變量，在函數中使用該臨時變量而不是直接使用變量?i?。
改進后的示例代碼如下：
1)、使用go關鍵字
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()
}
執(zhí)行后，輸出結果為：
0
9
3
4
5
6
7
8
1
2 注意，異步執(zhí)行時并不會保證按照函數注冊時的順序執(zhí)行，以下同理。
2)、使用臨時變量
package main
import (
"context"
"fmt"
"github.com/gogf/gf/v2/os/gctx"
"github.com/gogf/gf/v2/os/grpool"
"sync"
)
var (
ctx = gctx.New()
)
func main() {
wg := sync.WaitGroup{}
for i := 0; i < 10; i++ {
wg.Add(1)
v := i
grpool.Add(ctx, func(ctx context.Context) {
fmt.Println(v)
wg.Done()
})
}
wg.Wait()
} 執(zhí)行后，輸出結果為：
9
0
1
2
3
4
5
6
7
8這里可以看到，使用?grpool?進行任務注冊時，注冊方法為?func(ctx context.Context)?，因此無法在任務注冊時把變量?i?的值注冊進去（請盡量不要通過?ctx?傳遞業(yè)務參數），因此只能采用臨時變量的形式來傳遞當前變量?i?的值。
自動捕獲goroutine錯誤：AddWithRecover
?AddWithRecover?將新作業(yè)推送到具有指定恢復功能的池中。當?userFunc?執(zhí)行過程中出現(xiàn)?panic?時，會調用可選的?Recovery Func?。如果沒有傳入?Recovery Func?或賦空，則忽略?userFunc?引發(fā)的?panic?。該作業(yè)將異步執(zhí)行。
package main
import (
"context"
"fmt"
"github.com/gogf/gf/v2/container/garray"
"github.com/gogf/gf/v2/os/gctx"
"github.com/gogf/gf/v2/os/grpool"
"time"
)
var (
ctx = gctx.New()
)
func main() {
array := garray.NewArray(true)
grpool.AddWithRecover(ctx, func(ctx context.Context) {
array.Append(1)
array.Append(2)
panic(1)
}, func(err error) {
array.Append(1)
})
grpool.AddWithRecover(ctx, func(ctx context.Context) {
panic(1)
array.Append(1)
})
time.Sleep(500 * time.Millisecond)
fmt.Print(array.Len())
}測試一下grpool和原生的goroutine之間的性能
1)、grpool
package main
import (
"context"
"fmt"
"github.com/gogf/gf/v2/os/gctx"
"github.com/gogf/gf/v2/os/grpool"
"github.com/gogf/gf/v2/os/gtime"
"sync"
"time"
)
var (
ctx = gctx.New()
)
func main() {
start := gtime.TimestampMilli()
wg := sync.WaitGroup{}
for i := 0; i < 10000000; i++ {
wg.Add(1)
grpool.Add(ctx,func(ctx context.Context) {
time.Sleep(time.Millisecond)
wg.Done()
})
}
wg.Wait()
fmt.Println(grpool.Size())
fmt.Println("time spent:", gtime.TimestampMilli() - start)
} 2)、goroutine
package main
import (
"fmt"
"github.com/gogf/gf/v2/os/gtime"
"sync"
"time"
)
func main() {
start := gtime.TimestampMilli()
wg := sync.WaitGroup{}
for i := 0; i < 10000000; i++ {
wg.Add(1)
go func() {
time.Sleep(time.Millisecond)
wg.Done()
}()
}
wg.Wait()
fmt.Println("time spent:", gtime.TimestampMilli() - start)
} 3)、運行結果比較
測試結果為兩個程序各運行3次取平均值。
grpool:
goroutine count: 847313
memory spent: ~2.1 G
time spent: 37792 ms
goroutine:
goroutine count: 1000W
memory spent: ~4.8 GB
time spent: 27085 ms
可以看到池化過后，執(zhí)行相同數量的任務，?goroutine?數量減少很多，相對的內存也降低了一倍以上，CPU時間耗時也勉強可以接受。
分享標題：創(chuàng)新互聯(lián)GoFrame教程：GoFrame協(xié)程管理-grpool
文章地址：http://m.5511xx.com/article/djsjcjp.html
