go动态限制并发数量的实现示例

本文主要是介绍go动态限制并发数量的实现示例，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

《go动态限制并发数量的实现示例》本文主要介绍了Go并发控制方法,通过带缓冲通道和第三方库实现并发数量限制,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面...

带有缓冲大小的通道

package main

import (
	"fmt"
	"synwww.chinasem.cnc"
)

func main() {
	count := 10 // 最大支持并发
	pythonsum := 100 // 任务总数
	wg := sync.WaitGroup{} //控制主协程等待所有子协程执行完之后再退出。

	c := make(chan struct{}, count) // 控制任务并发的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 // 执行完毕，释放资源
		}(i)
	}
	wg.Wait()
}

代码中使用带有缓冲区的通道实现控制并发的数量。根据通道中没有数据时读取操作陷入阻塞和通道已满时继续写入操作陷入阻塞的特性，正好实现控制并发数量。

有中错误写法需要避免，资源释放的操作需要在子协程中做，如果放到主协程中将会将会无法控制最大并发数量，看例子：

package main

import (
	"fmt"
	"sync"
)

func main() {
	count := 10 // 最大支持并发
	sum := 100 // 任务总数
	wg := sync.WaitGroup{} //控制主协程等待所有子协程执行完之后再退出。

	c := make(chan struct{}, count) // 控制任务并发的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 // 执行完毕，释放资源
	}
	wg.Wait()
}

在第22行代码处的写法将无法控制最大并发数量，如果出现20行代码还未执行完，但是主协程中已经将channel中的数据读出，那么会出现超过10个协程同时运行的情况。造成这种情况的主要原因我觉得是go内存模型中并没有保证Go协程的退出确保发生在程序的某个事件之前，也就是说go协程的退出并没有保证是发生在 <-c 之前的。

使用第三方库

目前有很多第三方库实现了协程池，可以很方便地用来控制协程的并发数量，比较受欢迎的有：

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) 第一个参数是协程池的大小(poolSize)，第二个参数是协程运行的函数(worker)。
pool.Process(i) 将参数 i 传递给协程池定义好的 worker 处理。
pool.Close() 关闭协程池。

其他控制并发的方法

func main() {
	var wg sync.WaitGroupjs
	for i php:= 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()
}

上面这个可能只是控制并发，无法做到数量上的控制具体实现测试待测试

下面的则是动态控制

package util
 
import (
	"sync"
)
 
type concurrencyLimiter struct {
	runningNum  int32
	limit       int32
	blockingNum int32
	cond        *sync.Cond
	mu          *sync.Mutex
}
 
// NewConcurrencyLimiter 创建一个并发限制器，limit为并发限制数量，可通过 Reset() 动态调整limit。
// 每次调用 Get() 来获取一个资源，然后创建一个协程，完成任务后通过 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
	// 优先唤醒阻塞的任务
	if limit-tmp > 0 && blockingNum > 0 {
		for i := int32(0); i < limit-tmp && blockingNum > 0; i++ {
			c.cond.Signal()
			blockingNum--
		}
	}
}
 
// Get 当 concurrencyLimiter 没有资源时，会阻塞。
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 释放一个资源
func (c *concurrencyLimiter) Release() {
	c.mu.Lock()
	defer c.mu.Unlock()
 
	if c.blockingNum > 0 {
		c.runningNum--
		c.cond.Signal()
		return
	}
 
	c.runningNum--
}

到此这篇关于go动态限制并发数量的实现示例的文章就介绍php到这了,更多相关go动态限制并发数量内容请搜索China编程(www.chinasem.cn)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程China编程(www.chinasem.cn)！

这篇关于go动态限制并发数量的实现示例的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！