网站推广公司 wordpress,wordpress e,wordpress模板生,如何在海外进行推广在使用协程并发处理某些任务时, 其并发数量往往因为各种因素的限制不能无限的增大. 例如网络请求、数据库查询等等。
从运行效率角度考虑#xff0c;在相关服务可以负载的前提下#xff08;限制最大并发数#xff09;#xff0c;尽可能高的并发。
在Go语言中#xff0c;…在使用协程并发处理某些任务时, 其并发数量往往因为各种因素的限制不能无限的增大. 例如网络请求、数据库查询等等。
从运行效率角度考虑在相关服务可以负载的前提下限制最大并发数尽可能高的并发。
在Go语言中可以使用一些方法来控制协程goroutine的并发数量以防止并发过多导致资源耗尽或性能下降
1、使用信号量Semaphore
可以使用 Go 语言中的 channel 来实现简单的信号量限制并发数量
package mainimport (fmtsync
)func worker(id int, sem chan struct{}) {sem - struct{}{} // 占用一个信号量defer func() {-sem // 方法运行结束释放信号量}()// 执行工作任务fmt.Printf(Worker %d: Working...\n, id)
}func main() {concurrency : 3sem : make(chan struct{}, concurrency)var wg sync.WaitGroupfor i : 0; i 10; i {wg.Add(1)go func(id int) {defer wg.Done()worker(id, sem)}(i)}wg.Wait()close(sem)
}
sem 是一个有缓冲的 channel通过控制 channel 中元素的数量实现了一个简单的信号量机制 2、使用协程池
可以创建一个固定数量的协程池将任务分发给这些协程执行。
package mainimport (fmtsync
)func worker(id int, jobs -chan int, results chan- int) {//jobs等待主要协程往jobs放数据for j : range jobs {fmt.Printf(协程池 %d: 协程池正在工作 %d\n, id, j)results - j}
}func main() {const numJobs 5 //协程要做的工作数量const numWorkers 3 //协程池数量jobs : make(chan int, numJobs)results : make(chan int, numJobs)var wg sync.WaitGroup// 启动协程池for i : 1; i numWorkers; i {wg.Add(1)go func(id int) {defer wg.Done()worker(id, jobs, results)}(i)}// 提交任务for j : 1; j numJobs; j {jobs - j}close(jobs)// 等待所有工作完成go func() {wg.Wait()close(results)}()// 处理结果for result : range results {fmt.Println(Result:, result)}
}jobs 通道用于存储任务results 通道用于存储处理结果。通过创建固定数量的工作协程可以有效地控制并发数量。
3、使用其他包
Go 1.16 引入了 golang.org/x/sync/semaphore 包它提供了一个更为灵活的信号量实现。
案例一
限制对外部API的并发请求
假设我们有一个外部API它对并发请求有限制我们希望确保不超过这个限制。我们可以使用semaphore.Weighted来控制对API的并发访问
package mainimport (contextfmtgolang.org/x/sync/semaphoresynctime
)func main() {/*1、在并发量一定的情况下通过改变允许并发请求数可以更快处理请求任务(在CPU够用的前提下)2、sem : semaphore.NewWeighted(n)参数n就是权重量3、当一个协程需要获取的单位的权重越多运行就会慢(比如权重总量n5个一个协程分配了2个跟一个协程分配1个效率是不一样的)4、信号量没有足够的可用权重的情况发生在所有已分配的权重单位都已经被占用即信号量的当前权重计数达到了它的总容量。在这种情况下任何尝试通过Acquire方法获取更多权重的调用都将无法立即完成从而导致调用者通常是goroutine阻塞直到其他调用者释放一些权重单位。*//*1、权权重较大的任务在资源竞争时有更高的优先级更容易获得执行的机会2、如果当前资源足够满足高权重任务的需求这些任务将立即执行若资源不足则按照权重高低顺序排队等待3、一旦任务开始执行其完成的速度主要取决于任务自身的逻辑复杂度、所需资源以及系统的当前负载等因素与任务在信号量中的权重无关3、高权重的任务并不会中断已经在执行的低权重任务而是等待这些任务自行释放资源。一旦资源释放等待队列中的高权重任务将优先被唤醒4、Acquire 方法会检查当前信号量的可用资源量是否满足请求的权重如果满足则立即减少信号量的资源计数并返回允许任务继续执行。如果不满足任务将阻塞等待直到有足够的资源被释放*/// 记录开始时间startTime : time.Now()// 假设外部API允许的最大并发请求为5(信号量的总容量是5个权重单位)const (maxConcurrentRequests 5)sem : semaphore.NewWeighted(maxConcurrentRequests)var wg sync.WaitGroup// 模拟对外部API的10个并发请求for i : 0; i 10; i {wg.Add(1)go func(requestId int) {defer wg.Done()// 假设我们想要获取2个单位的权重if err : sem.Acquire(context.Background(), 2); err ! nil {fmt.Printf(请求 %d 无法获取信号量: %v\n, requestId, err)return}defer sem.Release(2) // 请求完成后释放信号量// 模拟对API的请求处理fmt.Printf(请求 %d 开始...\n, requestId)time.Sleep(2 * time.Second) // 模拟网络延迟fmt.Printf(请求 %d 完成。\n, requestId)}(i)}wg.Wait()// 记录结束时间endTime : time.Now()// 计算并打印总耗时fmt.Printf(程序总耗时: %v\n, endTime.Sub(startTime))
}信号量没有足够的可用权重的情况发生在所有已分配的权重单位都已经被占用即信号量的当前权重计数达到了它的总容量。在这种情况下任何尝试通过Acquire方法获取更多权重的调用都将无法立即完成从而导致调用者通常是goroutine阻塞直到其他调用者释放一些权重单位。
以下是一些导致信号量没有足够可用权重的具体情况 信号量初始化容量较小如果信号量的总容量设置得较小而并发请求的数量较大那么很快就会出现权重不足的情况。 长时间占用权重如果某些goroutine长时间占用权重单位而不释放这会导致其他goroutine无法获取到权重即使这些goroutine只是少数。 权重分配不均在某些情况下可能存在一些goroutine占用了不成比例的权重单位导致其他goroutine无法获取足够的权重。 权重释放不及时如果goroutine因为错误或异常情况提前退出而没有正确释放它们所占用的权重那么这些权重单位将不会被回收到信号量中。 高频率的请求在短时间内有大量goroutine请求权重即使它们请求的权重不大累积起来也可能超过信号量的总容量。 信号量权重未正确管理如果信号量的权重管理逻辑存在缺陷例如错误地释放了过多的权重或者在错误的时间点释放权重也可能导致可用权重不足。
为了避免信号量没有足够的可用权重可以采取以下措施
合理设置信号量容量根据资源限制和并发需求合理设置信号量的总容量。及时释放权重确保在goroutine完成工作后及时释放权重。使用超时在Acquire调用中使用超时避免无限期地等待权重。监控和日志记录监控信号量的使用情况并记录关键信息以便及时发现和解决问题。权重分配策略设计合理的权重分配策略确保权重的公平和高效分配。
通过这些措施可以更好地管理信号量的使用避免因权重不足导致的并发问题。
案例二
假设有一个在线视频平台它需要处理不同分辨率的视频转码任务。由于高清视频转码比标清视频更消耗计算资源因此平台希望设计一个系统能够优先处理更多标清视频转码请求同时又不完全阻塞高清视频的转码以保持整体服务质量和资源的有效利用。
package mainimport (fmtgolang.org/x/net/contextgolang.org/x/sync/semaphoreruntimesynctime
)// VideoTranscodeJob 视频转码任务
type VideoTranscodeJob struct {resolution stringweight int64
}func main() {cpuCount : runtime.NumCPU()fmt.Printf(当前CPU个数%v\n, cpuCount)/*1、权权重较大的任务在资源竞争时有更高的优先级更容易获得执行的机会2、如果当前资源足够满足高权重任务的需求这些任务将立即执行若资源不足则按照权重高低顺序排队等待3、一旦任务开始执行其完成的速度主要取决于任务自身的逻辑复杂度、所需资源以及系统的当前负载等因素与任务在信号量中的权重无关3、高权重的任务并不会中断已经在执行的低权重任务而是等待这些任务自行释放资源。一旦资源释放等待队列中的高权重任务将优先被唤醒4、Acquire 方法会检查当前信号量的可用资源量是否满足请求的权重如果满足则立即减少信号量的资源计数并返回允许任务继续执行。如果不满足任务将阻塞等待直到有足够的资源被释放*/// 初始化两个信号量一个用于标清一个用于高清假设总共有8个CPU核心可用normalSem : semaphore.NewWeighted(6) // 标清任务分配6个单位权重因为它们消耗资源较少highDefSem : semaphore.NewWeighted(2) // 高清任务分配2个单位权重因为它们更消耗资源var wg sync.WaitGroup//假设有20个需要转码的视频videoJobs : []VideoTranscodeJob{{HD, 2}, {HD, 2}, {SD, 1}, {SD, 1}, {HD, 2},{SD, 1}, {SD, 1}, {HD, 2}, {SD, 1}, {HD, 2},{SD, 4}, {SD, 4}, {HD, 2}, {SD, 1}, {HD, 2},{SD, 1}, {SD, 4}, {HD, 2}, {SD, 6}, {HD, 2},}for _, job : range videoJobs {wg.Add(1)go func(job VideoTranscodeJob) {defer wg.Done()var sem *semaphore.Weightedswitch job.resolution {case SD:sem normalSem //分配权重大当前为6任务在获取执行机会上有优势但并不直接意味着执行速度快case HD:sem highDefSemdefault:panic(无效的分辨率)}if err : sem.Acquire(context.Background(), job.weight); err ! nil {fmt.Printf(名为 %s 视频无法获取信号量: %v\n, job.resolution, err)return}defer sem.Release(job.weight) //释放权重对应的信号量// 模拟转码任务执行fmt.Printf(转码 %s 视频 (权重: %d)...\n, job.resolution, job.weight)//通过利用VideoTranscodeJob的weight值来模拟转码时间的长短HD用时长则设置2比SD的1大*时间就自然长运行就时间长time.Sleep(time.Duration(job.weight*100) * time.Millisecond) // 模拟不同分辨率视频转码所需时间fmt.Printf(------------------------%s 视频转码完成。。。\n, job.resolution)}(job)}wg.Wait()
}标清SD和高清HD分别分配了不同的权重1和2。通过创建两个不同权重的信号量我们可以控制不同类型任务的同时执行数量从而优先保证标清视频的快速处理同时也确保高清视频能够在不影响系统稳定性的情况下进行转码。这展示了带权重的并发控制如何帮助在资源有限的情况下优化任务调度和执行效率。 注意对协程分配的权重单位数不能大于对应上下文semaphore.NewWeighted(n)中参数n的单位数
案例三
package mainimport (contextfmtsynctimegolang.org/x/sync/semaphore
)type weightedTask struct {id intweight int64
}func main() {const (maxTotalWeight 20 // 最大总权重)sem : semaphore.NewWeighted(maxTotalWeight)var wg sync.WaitGrouptasksCh : make(chan weightedTask, 10)// 发送任务for i : 1; i 10; i {tasksCh - weightedTask{id: i, weight: int64(i)} // 假设任务ID即为其权重}close(tasksCh)// 启动任务处理器for task : range tasksCh {wg.Add(1)go func(task weightedTask) {defer wg.Done()if err : sem.Acquire(context.Background(), int64(task.id)); err ! nil {fmt.Printf(任务 %d 无法获取信号量: %v\n, task.id, err)return}defer sem.Release(int64(task.id)) //释放// 模拟任务执行fmt.Printf(任务 %d (权重: %d) 正在运行...\n, task.id, task.weight)time.Sleep(time.Duration(task.weight*100) * time.Millisecond) // 示例中简单用时间模拟权重影响fmt.Printf(任务 %d 完成.\n, task.id)}(task)}wg.Wait()
}总结
选择哪种方法取决于具体的应用场景和需求。使用信号量是一种简单而灵活的方法而协程池则更适用于需要批量处理任务的情况。golang.org/x/sync/semaphore 包提供了一个标准库外的更灵活的信号量实现
