快捷導(dǎo)航

go動(dòng)態(tài)限制并發(fā)數(shù)量的實(shí)現(xiàn)示例

更新時(shí)間：2025年07月24日 11:43:24 作者：譜寫

本文主要介紹了Go并發(fā)控制方法,通過帶緩沖通道和第三方庫實(shí)現(xiàn)并發(fā)數(shù)量限制,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

帶有緩沖大小的通道

package main

import (
	"fmt"
	"sync"
)

func main() {
	count := 10 // 最大支持并發(fā)
	sum := 100 // 任務(wù)總數(shù)
	wg := sync.WaitGroup{} //控制主協(xié)程等待所有子協(xié)程執(zhí)行完之后再退出。

	c := make(chan struct{}, count) // 控制任務(wù)并發(fā)的chan
	defer close(c)

	for i:=0; i<sum;i++{
		wg.Add(1)
		c <- struct{}{} // 作用類似于waitgroup.Add(1)
		go func(j int) {
			defer wg.Done()
			fmt.Println(j)
			<- c // 執(zhí)行完畢，釋放資源
		}(i)
	}
	wg.Wait()
}

代碼中使用帶有緩沖區(qū)的通道實(shí)現(xiàn)控制并發(fā)的數(shù)量。根據(jù)通道中沒有數(shù)據(jù)時(shí)讀取操作陷入阻塞和通道已滿時(shí)繼續(xù)寫入操作陷入阻塞的特性，正好實(shí)現(xiàn)控制并發(fā)數(shù)量。

有中錯(cuò)誤寫法需要避免，資源釋放的操作需要在子協(xié)程中做，如果放到主協(xié)程中將會將會無法控制最大并發(fā)數(shù)量，看例子：

package main

import (
	"fmt"
	"sync"
)

func main() {
	count := 10 // 最大支持并發(fā)
	sum := 100 // 任務(wù)總數(shù)
	wg := sync.WaitGroup{} //控制主協(xié)程等待所有子協(xié)程執(zhí)行完之后再退出。

	c := make(chan struct{}, count) // 控制任務(wù)并發(fā)的chan
	defer close(c)

	for i:=0; i<sum;i++{
		wg.Add(1)
		go func(j int) {
			defer wg.Done()
			c <- struct{}{} // 作用類似于waitgroup.Add(1)
			fmt.Println(j)
		}(i)
		<- c // 執(zhí)行完畢，釋放資源
	}
	wg.Wait()
}

在第22行代碼處的寫法將無法控制最大并發(fā)數(shù)量，如果出現(xiàn)20行代碼還未執(zhí)行完，但是主協(xié)程中已經(jīng)將channel中的數(shù)據(jù)讀出，那么會出現(xiàn)超過10個(gè)協(xié)程同時(shí)運(yùn)行的情況。造成這種情況的主要原因我覺得是go內(nèi)存模型中并沒有保證Go協(xié)程的退出確保發(fā)生在程序的某個(gè)事件之前，也就是說go協(xié)程的退出并沒有保證是發(fā)生在 <-c 之前的。

使用第三方庫

目前有很多第三方庫實(shí)現(xiàn)了協(xié)程池，可以很方便地用來控制協(xié)程的并發(fā)數(shù)量，比較受歡迎的有：

Jeffail/tunny
panjf2000/ants等等

以 tunny 舉例：

package main

import (
	"log"
	"time"

	"github.com/Jeffail/tunny"
)

func main() {
	pool := tunny.NewFunc(10, func(i interface{}) interface{} {
		log.Println(i)
		time.Sleep(time.Second)
		return nil
	})
	defer pool.Close()

	for i := 0; i < 500; i++ {
		go pool.Process(i)
	}
	time.Sleep(time.Second * 4)
}

tunny.NewFunc(3, f) 第一個(gè)參數(shù)是協(xié)程池的大小(poolSize)，第二個(gè)參數(shù)是協(xié)程運(yùn)行的函數(shù)(worker)。
pool.Process(i) 將參數(shù) i 傳遞給協(xié)程池定義好的 worker 處理。
pool.Close() 關(guān)閉協(xié)程池。

其他控制并發(fā)的方法

func main() {
	var wg sync.WaitGroup
	for i := 0; i < math.MaxInt32; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			fmt.Println(i)
			time.Sleep(time.Second)
		}(i)
	}
	wg.Wait()
}

上面這個(gè)可能只是控制并發(fā)，無法做到數(shù)量上的控制具體實(shí)現(xiàn)測試待測試

下面的則是動(dòng)態(tài)控制

package util
 
import (
	"sync"
)
 
type concurrencyLimiter struct {
	runningNum  int32
	limit       int32
	blockingNum int32
	cond        *sync.Cond
	mu          *sync.Mutex
}
 
// NewConcurrencyLimiter 創(chuàng)建一個(gè)并發(fā)限制器，limit為并發(fā)限制數(shù)量，可通過 Reset() 動(dòng)態(tài)調(diào)整limit。
// 每次調(diào)用 Get() 來獲取一個(gè)資源，然后創(chuàng)建一個(gè)協(xié)程，完成任務(wù)后通過 Release() 釋放資源。
func NewConcurrencyLimiter(limit int32) *concurrencyLimiter {
	l := new(sync.Mutex)
	return &concurrencyLimiter{
		limit: limit,
		cond:  sync.NewCond(l),
		mu:    l,
	}
}
 
// Reset 可更新limit，需要保證limit > 0
func (c *concurrencyLimiter) Reset(limit int32) {
	c.mu.Lock()
	defer c.mu.Unlock()
 
	tmp := c.limit
	c.limit = limit
	blockingNum := c.blockingNum
	// 優(yōu)先喚醒阻塞的任務(wù)
	if limit-tmp > 0 && blockingNum > 0 {
		for i := int32(0); i < limit-tmp && blockingNum > 0; i++ {
			c.cond.Signal()
			blockingNum--
		}
	}
}
 
// Get 當(dāng) concurrencyLimiter 沒有資源時(shí)，會阻塞。
func (c *concurrencyLimiter) Get() {
	c.mu.Lock()
	defer c.mu.Unlock()
 
	if c.runningNum < c.limit {
		c.runningNum++
		return
	}
	c.blockingNum++
	for !(c.runningNum < c.limit) {
		c.cond.Wait()
	}
	c.runningNum++
	c.blockingNum--
}
 
// Release 釋放一個(gè)資源
func (c *concurrencyLimiter) Release() {
	c.mu.Lock()
	defer c.mu.Unlock()
 
	if c.blockingNum > 0 {
		c.runningNum--
		c.cond.Signal()
		return
	}
 
	c.runningNum--
}

到此這篇關(guān)于go動(dòng)態(tài)限制并發(fā)數(shù)量的實(shí)現(xiàn)示例的文章就介紹到這了,更多相關(guān)go動(dòng)態(tài)限制并發(fā)數(shù)量內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: