GoFrame 協(xié)程管理-grpool

2022-04-11 14:56 更新

基本介紹

Go語言中的?goroutine?雖然相對于系統(tǒng)線程來說比較輕量級（初始棧大小僅2KB），（并且支持動態(tài)擴容），而正常采用java，c++等語言啟用的線程一般都是內(nèi)核態(tài)的占用的內(nèi)存資源一般在4m左右，而假設(shè)我們的服務(wù)器CPU內(nèi)存為4G，那么很明顯才用的內(nèi)核態(tài)線程的并發(fā)總數(shù)量也就是1024個，相反查看一下Go語言的協(xié)程則可以達到4*1024*1024/2=200w.這么一看就明白了為什么Go語言天生支持高并發(fā)。但是在高并發(fā)量下的?goroutine?頻繁創(chuàng)建和銷毀對于性能損耗以及GC來說壓力也不小。充分將?goroutine?復(fù)用，減少?goroutine?的創(chuàng)建/銷毀的性能損耗，這便是?grpool?對?goroutine?進行池化封裝的目的。例如，針對于100W個執(zhí)行任務(wù)，使用?goroutine?的話需要不停創(chuàng)建并銷毀100W個?goroutine?，而使用?grpool?也許底層只需要幾萬個?goroutine?便能充分復(fù)用地執(zhí)行完成所有任務(wù)。

經(jīng)測試，?goroutine?池對于業(yè)務(wù)邏輯的執(zhí)行效率（降低執(zhí)行時間/CPU使用率）提升不大，甚至沒有原生的?goroutine?執(zhí)行快速（池化?goroutine?執(zhí)行調(diào)度并沒有底層go調(diào)度器高效，因為池化?goroutine?的執(zhí)行調(diào)度也是基于底層go調(diào)度器），但是由于采用了復(fù)用的設(shè)計，池化后對內(nèi)存的使用率得到極大的降低。在v2版本中?grpool?也加入了貫穿全局的鏈路追蹤。

概念：

  1. ?Pool?：?goroutine?池，用于管理若干可復(fù)用的?goroutine?協(xié)程資源；
  2. ?Worker?：池對象中參與任務(wù)執(zhí)行的?goroutine?，一個?Worker?可以執(zhí)行若干個?Job?，直到隊列中再無等待的?Job?；
  3. ?Job?：添加到池對象的任務(wù)隊列中等待執(zhí)行的任務(wù)，是一個?func()?的方法，一個?Job?同時只能被一個?Worker?獲取并執(zhí)行；

使用方式：

import "github.com/gogf/gf/v2/os/grpool"

使用場景：管理大量異步任務(wù)的場景、需要異步協(xié)程復(fù)用的場景、需要降低內(nèi)存使用率的場景。

接口文檔：

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?池對象，參數(shù)?limit?為非必需參數(shù)，用于限定池中的工作?goroutine?數(shù)量，默認為不限制。需要注意的是，任務(wù)可以不停地往池中添加，沒有限制，但是工作的?goroutine?是可以做限制的。我們可以通過?Size()?方法查詢當(dāng)前的工作?goroutine?數(shù)量，使用?Jobs()?方法查詢當(dāng)前池中待處理的任務(wù)數(shù)量。

同時，為便于使用，?grpool?包提供了默認的?goroutine?池，默認的池對象不限制?goroutine?數(shù)量，直接通過?grpool.Add?即可往默認的池中添加任務(wù)，任務(wù)參數(shù)必須是一個 ?func()?類型的函數(shù)/方法。

使用示例

使用默認的goroutine池，限制100個goroutine執(zhí)行1000個任務(wù)

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 {}
}

這段程序中的任務(wù)函數(shù)的功能是?sleep? 1秒鐘，這樣便能充分展示出?goroutine?數(shù)量限制功能。其中，我們使用了?gtime.SetInterval?定時器每隔1秒鐘打印出當(dāng)前默認池中的工作?goroutine?數(shù)量以及待處理的任務(wù)數(shù)量。

異步傳參：來個新手容易出錯的例子

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í)行結(jié)果無論是采用go關(guān)鍵字來執(zhí)行還是?grpool?來執(zhí)行都是如此。原因是，對于異步線程/協(xié)程來講，函數(shù)進行異步執(zhí)行注冊時，該函數(shù)并未真正開始執(zhí)行(注冊時只在?goroutine?的棧中保存了變量?i?的內(nèi)存地址)，而一旦開始執(zhí)行時函數(shù)才會去讀取變量?i?的值，而這個時候變量?i?的值已經(jīng)自增到了10。 清楚原因之后，改進方案也很簡單了，就是在注冊異步執(zhí)行函數(shù)的時候，把當(dāng)時變量i的值也一并傳遞獲?。换蛘甙旬?dāng)前變量i的值賦值給一個不會改變的臨時變量，在函數(shù)中使用該臨時變量而不是直接使用變量?i?。

改進后的示例代碼如下：

1)、使用go關(guān)鍵字

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í)行后，輸出結(jié)果為：

0
9
3
4
5
6
7
8
1
2

注意，異步執(zhí)行時并不會保證按照函數(shù)注冊時的順序執(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í)行后，輸出結(jié)果為：

9
0
1
2
3
4
5
6
7
8

這里可以看到，使用?grpool?進行任務(wù)注冊時，注冊方法為?func(ctx context.Context)?，因此無法在任務(wù)注冊時把變量?i?的值注冊進去（請盡量不要通過?ctx?傳遞業(yè)務(wù)參數(shù)），因此只能采用臨時變量的形式來傳遞當(dāng)前變量?i?的值。

自動捕獲goroutine錯誤：AddWithRecover

?AddWithRecover?將新作業(yè)推送到具有指定恢復(fù)功能的池中。當(dāng)?userFunc?執(zhí)行過程中出現(xiàn)?panic?時，會調(diào)用可選的?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)、運行結(jié)果比較

測試結(jié)果為兩個程序各運行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í)行相同數(shù)量的任務(wù)，?goroutine?數(shù)量減少很多，相對的內(nèi)存也降低了一倍以上，CPU時間耗時也勉強可以接受。


以上內(nèi)容是否對您有幫助：
在線筆記
App下載
App下載

掃描二維碼

下載編程獅App

公眾號
微信公眾號

編程獅公眾號