go語言限制協(xié)程并發(fā)數的方案詳情

更新時間：2022年01月24日 14:39:03 作者：DoubleLi

一個線程中可以有任意多個協(xié)程，但某一時刻只能有一個協(xié)程在運行，多個協(xié)程分享該線程分配到的計算機資源，接下來通過本文給大家介紹go語言限制協(xié)程的并發(fā)數的方案詳情，感興趣的朋友一起看看吧

前言

在使用協(xié)程并發(fā)處理某些任務時, 其并發(fā)數量往往因為各種因素的限制不能無限的增大. 例如網絡請求、數據庫查詢等等。從運行效率角度考慮，在相關服務可以負載的前提下（限制最大并發(fā)數），盡可能高的并發(fā)。本文就這個問題探尋一下解決方案和實現。共兩種思路，一是使用帶緩沖的通道實現，二是使用鎖實現。

一、使用帶緩沖的通道限制并發(fā)數

1.1方案詳情

先上代碼如下, 邏輯很簡單.

package golimit

type GoLimit struct {
    ch chan int
}
func NewGoLimit(max int) *GoLimit {
    return &GoLimit{ch: make(chan int, max)}
func (g *GoLimit) Add() {
    g.ch <- 1
func (g *GoLimit) Done() {
    <-g.ch

按允許最大并發(fā)數創(chuàng)建一個帶緩沖的通道, 創(chuàng)建協(xié)程之前調用Add()往通道里寫一個數據, 協(xié)程完成是調用Done()方法讀取一個數據. 若無法往通道里寫數據時, 表示通道已經寫滿, 也就是目前的協(xié)程并發(fā)數為允許的最大數量. Add()方法將被阻塞, 也就無法創(chuàng)建新的協(xié)程. 直到有協(xié)程運行完成, 調用Done()方法讀取了通道了一個數據.

以下是使用示例

package main

import (
    "golimit"
    "log"
    "time"
)
func main() {
    log.Println("開始測試...")
    g := golimit.NewGoLimit(2) //max_num(最大允許并發(fā)數)設置為2
    for i := 0; i < 10; i++ {
        //嘗試增加一個協(xié)程, 若已達到最大并發(fā)數,將阻塞
        g.Add()
        go func(g *golimit.GoLimit, i int) {
            defer g.Done() //一個并發(fā)協(xié)程已經完成
            time.Sleep(time.Second * 2)
            log.Println(i, "done")
        }(g, i)
    }
    log.Println("循環(huán)結束")
    time.Sleep(time.Second * 3)//等待執(zhí)行完成
    log.Println("測試結束")
}

1.2評估總結

優(yōu)點：此方案的實現邏輯簡單明了，易理解、易維護。若能滿足需求，在一般的場景下，此方案為首選。

隱憂：使用通道的緩沖區(qū)的大小來表示最大可并發(fā)數，在允許并發(fā)數較大，如幾千幾萬甚至更大的情況下，通道的性能和內存的負載是否會有問題，我不太清楚，若哪位朋友知道請告知一下。

不足：運行中難以調整最大可并發(fā)數。而在某些場景下是有這種需求的，如A服務依賴的B服務有擴容或縮減，但A服務不能停止，需要調整請求B服務接口的最大可并發(fā)數。二、使用鎖實現協(xié)程并發(fā)數量限制2.1方案詳情

同樣先上代碼（注：此代碼我已經在github上開源https://github.com/zh-five/golimit）

// 協(xié)程并發(fā)數限制庫
package golimit
import (
    "sync"
)
type GoLimit struct {
    max       uint             //并發(fā)最大數量
    count     uint             //當前已有并發(fā)數
    isAddLock bool             //是否已鎖定增加
    zeroChan  chan interface{} //為0時廣播
    addLock   sync.Mutex       //(增加并發(fā)數的)鎖
    dataLock  sync.Mutex       //(修改數據的)鎖
}
func NewGoLimit(max uint) *GoLimit {
    return &GoLimit{max: max, count: 0, isAddLock: false, zeroChan: nil}
}
//并發(fā)計數加1.若 計數>=max_num, 則阻塞,直到 計數<max_num
func (g *GoLimit) Add() {
    g.addLock.Lock()
    g.dataLock.Lock()
    g.count += 1
    if g.count < g.max { //未超并發(fā)時解鎖,后續(xù)可以繼續(xù)增加
        g.addLock.Unlock()
    } else { //已到最大并發(fā)數, 不解鎖并標記. 等數量減少后解鎖
        g.isAddLock = true
    }
    g.dataLock.Unlock()
}
//并發(fā)計數減1
//若計數<max_num, 可以使原阻塞的Add()快速解除阻塞
func (g *GoLimit) Done() {
    g.dataLock.Lock()
    g.count -= 1
    //解鎖
    if g.isAddLock == true && g.count < g.max {
        g.isAddLock = false
        g.addLock.Unlock()
    }
    //0廣播
    if g.count == 0 && g.zeroChan != nil {
        close(g.zeroChan)
        g.zeroChan = nil
    }
    g.dataLock.Unlock()
}
//更新最大并發(fā)計數為, 若是調大, 可以使原阻塞的Add()快速解除阻塞
func (g *GoLimit) SetMax(n uint) {
    g.dataLock.Lock()
    g.max = n
    //解鎖
    if g.isAddLock == true && g.count < g.max {
        g.isAddLock = false
        g.addLock.Unlock()
    }
    //加鎖
    if g.isAddLock == false && g.count >= g.max {
        g.isAddLock = true
        g.addLock.Lock()
    }
    g.dataLock.Unlock()
}
//若當前并發(fā)計數為0, 則快速返回; 否則阻塞等待,直到并發(fā)計數為0
func (g *GoLimit) WaitZero() {
    g.dataLock.Lock()
    //無需等待
    if g.count == 0 {
        g.dataLock.Unlock()
        return
    }
    //無廣播通道, 創(chuàng)建一個
    if g.zeroChan == nil {
        g.zeroChan = make(chan interface{})
    }
    //復制通道后解鎖, 避免從nil讀數據
    c := g.zeroChan
    g.dataLock.Unlock()
    <-c
}
//獲取并發(fā)計數
func (g *GoLimit) Count() uint {
    return g.count
}
//獲取最大并發(fā)計數
func (g *GoLimit) Max() uint {
    return g.max
}

總共使用了兩把鎖，一把是數據鎖（dataLock），用來鎖定數據，保證數據修改安全，加鎖解鎖是在修改數據前后進行的；另一把是增加能否增加協(xié)程的鎖（addLock），增加協(xié)程時必須先加鎖，加鎖成功后修改并發(fā)數，若并發(fā)數小于最大可并發(fā)數，則解鎖，否則不解鎖，促使后續(xù)增加協(xié)程的加鎖操作阻塞，從而限制協(xié)程的并發(fā)數。使用示例如下：

package main
import (
    "github.com/zh-five/golimit"
    "log"
    "time"
)
func main() {
    log.Println("開始測試...")
    g := golimit.NewGoLimit(2) //max_num(最大允許并發(fā)數)設置為2
    for i := 0; i < 10; i++ {
        //并發(fā)計數加1.若 計數>=max_num, 則阻塞,直到 計數<max_num
        g.Add()
        //運行過程中可以隨時修改最大可并發(fā)數據
        //g.SetMax(3)
        go func(g *golimit.GoLimit, i int) {
            defer g.Done() //并發(fā)計數減1
            time.Sleep(time.Second * 2)
            log.Println(i, "done")
        }(g, i)
    }
    log.Println("循環(huán)結束")
    g.WaitZero() //阻塞, 直到所有并發(fā)都完成
    log.Println("測試結束")
}

方案2的GoLimit除了增加了SetMax()方法用于修改最大可并發(fā)數。出于好玩和偷懶增加了一個WaitZero()方法（其實外部使用sync.WaitGroup也可以快速實現此功能），用于阻塞等待所有并發(fā)協(xié)程都執(zhí)行完成。大約可以用于如下場景：有一大批url需要有限制的并發(fā)采集數據，主程序里只需要簡單的調用一下WaitZero()方法，就可以阻塞等等所有采集的協(xié)程完成。