前言
并發(fā)編程一直是Golang區(qū)別與其他語言的很大優(yōu)勢,也是實(shí)際工作場景中經(jīng)常遇到的。近日筆者在組內(nèi)分享了我們常見的并發(fā)場景,及代碼示例,以期望大家能在遇到相同場景下,能快速的想到解決方案,或者是拿這些方案與自己實(shí)現(xiàn)的比較,取長補(bǔ)短?,F(xiàn)整理出來與大家共享。
簡單并發(fā)場景
很多時(shí)候,我們只想并發(fā)的做一件事情,比如測試某個(gè)接口的是否支持并發(fā)。那么我們就可以這么做:
func RunScenario1() {
count := 10
var wg sync.WaitGroup
for i := 0; i count; i++ {
wg.Add(1)
go func(index int) {
defer wg.Done()
doSomething(index)
}(i)
}
wg.Wait()
}
使用goroutine來實(shí)現(xiàn)異步,使用WaitGroup來等待所有g(shù)oroutine結(jié)束。這里要注意的是要正確釋放WaitGroup的counter(在goroutine里調(diào)用Done()方法)。
但此種方式有個(gè)弊端,就是當(dāng)goroutine的量過多時(shí),很容易消耗完客戶端的資源,導(dǎo)致程序表現(xiàn)不佳。
規(guī)定時(shí)間內(nèi)的持續(xù)并發(fā)模型
我們?nèi)匀灰詼y試某個(gè)后端API接口為例,如果我們想知道這個(gè)接口在持續(xù)高并發(fā)情況下是否有句柄泄露,這種情況該如何測試呢?
這種時(shí)候,我們需要能控制時(shí)間的高并發(fā)模型:
func RunScenario2() {
timeout := time.Now().Add(time.Second * time.Duration(10))
n := runtime.NumCPU()
waitForAll := make(chan struct{})
done := make(chan struct{})
concurrentCount := make(chan struct{}, n)
for i := 0; i n; i++ {
concurrentCount - struct{}{}
}
go func() {
for time.Now().Before(timeout) {
-done
concurrentCount - struct{}{}
}
waitForAll - struct{}{}
}()
go func() {
for {
-concurrentCount
go func() {
doSomething(rand.Intn(n))
done - struct{}{}
}()
}
}()
-waitForAll
}
上面的代碼里,我們通過一個(gè)buffered channel來控制并發(fā)的數(shù)量(concurrentCount),然后另起一個(gè)channel來周期性的發(fā)起新的任務(wù),而控制的條件就是 time.Now().Before(timeout),這樣當(dāng)超過規(guī)定的時(shí)間,waitForAll 就會得到信號,而使整個(gè)程序退出。
這是一種實(shí)現(xiàn)方式,那么還有其他的方式?jīng)]?我們接著往下看。
基于大數(shù)據(jù)量的并發(fā)模型
前面說的基于時(shí)間的并發(fā)模型,那如果只知道數(shù)據(jù)量很大,但是具體結(jié)束時(shí)間不確定,該怎么辦呢?
比如,客戶給了個(gè)幾TB的文件列表,要求把這些文件從存儲里刪除。再比如,實(shí)現(xiàn)個(gè)爬蟲去爬某些網(wǎng)站的所有內(nèi)容。
而解決此類問題,最常見的就是使用工作池模式了(Worker Pool)。以刪文件為例,我們可以簡單這樣來處理:
Jobs - 可以從文件列表里讀取文件,初始化為任務(wù),然后發(fā)給worker
Worker - 拿到任務(wù)開始做事
Collector - 收集worker處理后的結(jié)果
Worker Pool - 控制并發(fā)的數(shù)量
雖然這只是個(gè)簡單Worker Pool模型,但已經(jīng)能滿足我們的需求:
func RunScenario3() {
numOfConcurrency := runtime.NumCPU()
taskTool := 10
jobs := make(chan int, taskTool)
results := make(chan int, taskTool)
var wg sync.WaitGroup
// workExample
workExampleFunc := func(id int, jobs -chan int, results chan- int, wg *sync.WaitGroup) {
defer wg.Done()
for job := range jobs {
res := job * 2
fmt.Printf("Worker %d do things, produce result %d \n", id, res)
time.Sleep(time.Millisecond * time.Duration(100))
results - res
}
}
for i := 0; i numOfConcurrency; i++ {
wg.Add(1)
go workExampleFunc(i, jobs, results, wg)
}
totalTasks := 100 // 本例就要從文件列表里讀取
wg.Add(1)
go func() {
defer wg.Done()
for i := 0; i totalTasks; i++ {
n := -results
fmt.Printf("Got results %d \n", n)
}
close(results)
}()
for i := 0; i totalTasks; i++ {
jobs - i
}
close(jobs)
wg.Wait()
}
在Go里,分發(fā)任務(wù),收集結(jié)果,我們可以都交給Channel來實(shí)現(xiàn)。從實(shí)現(xiàn)上更加的簡潔。
仔細(xì)看會發(fā)現(xiàn),本模型也是適用于按時(shí)間來控制并發(fā)。只要把totalTask的遍歷換成時(shí)間控制就好了。
等待異步任務(wù)執(zhí)行結(jié)果
goroutine和channel的組合在實(shí)際編程時(shí)經(jīng)常會用到,而加上Select更是無往而不利。
func RunScenario4() {
sth := make(chan string)
result := make(chan string)
go func() {
id := rand.Intn(100)
for {
sth - doSomething(id)
}
}()
go func() {
for {
result - takeSomthing(-sth)
}
}()
select {
case c := -result:
fmt.Printf("Got result %s ", c)
case -time.After(time.Duration(30 * time.Second)):
fmt.Errorf("指定時(shí)間內(nèi)都沒有得到結(jié)果")
}
}
在select的case情況,加上time.After()模型可以讓我們在一定時(shí)間范圍內(nèi)等待異步任務(wù)結(jié)果,防止程序卡死。
定時(shí)反饋異步任務(wù)結(jié)果
上面我們說到持續(xù)的壓測某后端API,但并未實(shí)時(shí)收集結(jié)果。而很多時(shí)候?qū)τ谛阅軠y試場景,實(shí)時(shí)的統(tǒng)計(jì)吞吐率,成功率是非常有必要的。
func RunScenario5() {
concurrencyCount := runtime.NumCPU()
for i := 0; i concurrencyCount; i++ {
go func(index int) {
for {
doUploadMock()
}
}(i)
}
t := time.NewTicker(time.Second)
for {
select {
case -t.C:
// 計(jì)算并打印實(shí)時(shí)數(shù)據(jù)
}
}
}
這種場景就需要使用到Ticker,且上面的Example模型還能控制并發(fā)數(shù)量,也是非常實(shí)用的方式。
知識點(diǎn)總結(jié)
上面我們共提到了五種并發(fā)模式:
- 簡單并發(fā)模型
- 規(guī)定時(shí)間內(nèi)的持續(xù)并發(fā)模型
- 基于大數(shù)據(jù)量的持續(xù)并發(fā)模型
- 等待異步任務(wù)結(jié)果模型
- 定時(shí)反饋異步任務(wù)結(jié)果模型
歸納下來其核心就是使用了Go的幾個(gè)知識點(diǎn):Goroutine, Channel, Select, Time, Timer/Ticker, WaitGroup. 若是對這些不清楚,可以自行Google之。
另完整的Example 代碼可以參考這里:https://github.com/jichangjun/golearn/blob/master/src/carlji.com/experiments/concurrency/main.go
使用方式: go run main.go 場景>
比如 :
參考文檔
https://github.com/golang/go/wiki/LearnConcurrency
這篇是Google官方推薦學(xué)習(xí)Go并發(fā)的資料,從初學(xué)者到進(jìn)階,內(nèi)容非常豐富,且權(quán)威。
Contact me ?
Email: jinsdu@outlook.com
Blog: http://www.cnblogs.com/jinsdu/
Github: https://github.com/CarlJi
您可能感興趣的文章:- golang 并發(fā)編程之生產(chǎn)者消費(fèi)者詳解
- golang并發(fā)編程的實(shí)現(xiàn)
- GO語言并發(fā)編程之互斥鎖、讀寫鎖詳解
- Golang Goroutine的使用
- Go語言學(xué)習(xí)之goroutine詳解
- Go并發(fā)編程之正確使用goroutine的方法