Golang如何控制并发goroutine数量使用semaphore权重信号量-Golang-PHP中文网

在go语言中控制goroutine并发数量的推荐方法是使用x/sync/semaphore信号量。1. semaphore.newweighted创建带容量的信号量；2. 每个goroutine执行前用acquire获取权重；3. 执行结束后用release释放权重；4. 结合context可实现超时控制和优雅退出。这种方法相比waitgroup和channel更灵活，能精确控制并发资源，避免系统过载。

Golang如何控制并发goroutine数量使用semaphore权重信号量

Golang中要控制并发的goroutine数量，使用标准库扩展包

x/sync/semaphore

登录后复制

提供的信号量（weighted semaphore）是一个非常优雅且强大的方法。它就像一个智能的交通管制员，能够精准地限制同时执行的任务数量，有效避免系统资源耗尽或过载。

解决方案

在Go语言中，如果你需要限制同时运行的goroutine数量，例如，你正在处理一个巨大的任务队列，但又不希望一次性启动成千上万个goroutine导致系统崩溃，那么

x/sync/semaphore

登录后复制

包就是你的不二之选。它提供了一种基于权重的信号量机制，你可以给信号量设置一个总容量，每个goroutine在执行前需要“获取”一定量的权重，执行完毕后“释放”这些权重。当总容量不足时，后续的goroutine就会被阻塞，直到有足够的容量被释放。

一个典型的使用场景是：你有一堆需要处理的图片，但你的服务器只能同时处理有限数量的图片压缩任务。

立即学习“go语言免费学习笔记（深入）”；

package main

import (
    "context"
    "fmt"
    "log"
    "runtime"
    "sync"
    "time"

    "golang.org/x/sync/semaphore"
)

func processImage(id int) {
    fmt.Printf("处理图片 %d 开始...\n", id)
    time.Sleep(time.Duration(id%3+1) * time.Second) // 模拟不同处理时间
    fmt.Printf("处理图片 %d 完成。\n", id)
}

func main() {
    maxConcurrency := int64(runtime.NumCPU()) // 通常我会设置为CPU核心数或根据实际负载测试决定
    sem := semaphore.NewWeighted(maxConcurrency)

    totalImages := 20
    var wg sync.WaitGroup

    fmt.Printf("开始处理 %d 张图片，最大并发数: %d\n", totalImages, maxConcurrency)

    for i := 0; i < totalImages; i++ {
        wg.Add(1)
        go func(imageID int) {
            defer wg.Done()

            // 尝试获取一个权重（默认是1），如果容量不足则阻塞
            // 在真实项目中，这里通常会结合 context.WithTimeout/WithCancel 来避免无限等待
            if err := sem.Acquire(context.Background(), 1); err != nil {
                log.Printf("获取信号量失败，图片 %d: %v\n", imageID, err)
                return
            }
            defer sem.Release(1) // 处理完后释放权重

            processImage(imageID)
        }(i)
    }

    wg.Wait()
    fmt.Println("所有图片处理完毕。")
}

登录后复制

这段代码中，

semaphore.NewWeighted(maxConcurrency)

登录后复制

创建了一个容量为

maxConcurrency

登录后复制

的信号量。每个goroutine在调用

sem.Acquire(context.Background(), 1)

登录后复制

时，会尝试获取一个“单位”的执行权。如果当前活跃的goroutine数量已经达到

maxConcurrency

登录后复制

，那么新的goroutine就会在这里等待，直到有其他goroutine调用

sem.Release(1)

登录后复制

释放了执行权。这样，无论你启动多少个goroutine，真正同时运行的始终不会超过你设定的最大并发数。

为什么我们需要控制Goroutine的数量？

说实话，Go语言的goroutine非常轻量级，启动几万甚至几十万个goroutine在理论上是可行的。但实际情况往往比理论复杂得多。我觉得，控制goroutine数量的核心原因在于：资源是有限的。

想象一下，你有一个程序需要从数据库读取大量数据，然后进行复杂的计算。如果每个数据项都启动一个goroutine去处理，而你的数据库连接池只有几十个连接，或者你的CPU核心数只有几个，那么：

资源耗尽： 大量的goroutine可能会瞬间耗尽系统资源，比如内存（每个goroutine虽然小，但量变引起质变）、文件描述符（如果涉及大量文件操作或网络连接）。这可能导致程序崩溃，或者系统变得极其缓慢，甚至影响其他进程。
上下文切换开销： 即使资源没有耗尽，过多的goroutine也会导致Go调度器频繁进行上下文切换。每一次切换都会带来CPU开销，降低整体效率，就像一个理发师同时服务太多顾客，虽然每个顾客都等在那里，但理发师的效率却下降了。
外部系统压力： 如果你的goroutine需要与外部系统（如数据库、缓存、第三方API）交互，不受控制的并发量可能会瞬间压垮这些外部系统，导致它们拒绝服务或性能急剧下降。这就像你突然涌入一家小餐馆，把厨师和服务员都搞懵了。
稳定性与可预测性： 限制并发数能让你的程序行为更可预测，更容易调试和优化。你清楚地知道，无论输入多大，你的系统都不会因为并发过高而崩溃。

在我看来，这是一种对系统负载的“自我保护”机制，确保你的程序在任何情况下都能保持稳定和高效。

信号量与WaitGroup、Channel有何不同？何时选择信号量？

Go语言提供了多种并发原语，它们各有侧重，有时候容易混淆。理解它们之间的差异，能帮助你做出正确的选择。

```
sync.WaitGroup
```
登录后复制
：它的主要作用是等待一组goroutine完成。你调用
```
Add
```
登录后复制
来增加计数，每个goroutine完成时调用
```
Done
```
登录后复制
来减少计数，最后通过
```
Wait
```
登录后复制
阻塞直到计数归零。
```
WaitGroup
```
登录后复制
关心的是“所有任务都完成了没？”，它不限制同时运行的goroutine数量。比如，你启动了100个goroutine，
```
WaitGroup
```
登录后复制
只是等着这100个都跑完，它并不会限制同时只有10个在跑。

绘蛙AI修图
绘蛙平台AI修图工具，支持手脚修复、商品重绘、AI扩图、AI换色

264

查看详情
```
chan
```
登录后复制
(通道)： 通道主要用于goroutine之间的通信，传递数据。当然，你也可以利用带缓冲的通道来实现简单的并发控制，例如，创建一个容量为N的缓冲通道，每次启动goroutine前向通道发送一个“令牌”，goroutine结束后从通道接收一个“令牌”。这种方式虽然能实现并发控制，但语义上不如信号量直接，而且如果需要处理不同“权重”的任务，实现起来会比较麻烦。通道更侧重于数据流和同步。
```
x/sync/semaphore.Weighted
```
登录后复制
(信号量)： 信号量就是专门用来限制并发数量的。它管理的是“资源许可”或者“执行槽位”。它关心的是“当前有多少个任务正在执行？我还能允许多少个新任务开始？”。它的“权重”概念尤其灵活，你可以让一个耗费资源的goroutine获取多个权重，而一个轻量级的goroutine只获取一个权重，这样就能更精细地控制资源消耗。

那么，何时选择信号量呢？

在我实际工作中，我倾向于在以下场景使用信号量：

限制并发访问外部资源： 比如，你的应用需要频繁调用一个有速率限制的第三方API，或者访问一个数据库连接数有限的数据库。使用信号量可以确保你的并发请求不会超出外部系统的承受能力。
控制计算密集型任务的并发： 当你有一批需要大量CPU或内存的任务时，你可能希望只允许有限数量的任务同时运行，以避免系统过载。
当任务有不同“大小”或“成本”时： 如果某些任务比其他任务消耗更多的资源，你可以让它们获取更大的权重，而轻量级任务获取较小的权重。这样，你可以更有效地利用总的资源容量。
需要阻塞等待资源可用时： 信号量的
```
Acquire
```
登录后复制
方法会阻塞，直到有足够的资源可用，这非常符合“等待许可”的语义。

简单来说，如果你需要限制同时运行的“活动单元”的数量，并且这些活动单元可能消耗不同量的“资源配额”，那么信号量通常是最佳选择。

如何处理信号量中的超时与错误？

在实际生产环境中，仅仅使用

sem.Acquire(context.Background(), 1)

登录后复制

是不够的。

context.Background()

登录后复制

意味着它会无限期地等待，直到获取到信号量。但在很多场景下，我们不希望任务无限期地阻塞，而是希望在一定时间后放弃，或者在外部信号（如程序关闭）时停止等待。

这就是

context

登录后复制

包发挥作用的地方。

semaphore.Acquire

登录后复制

方法接受一个

context.Context

登录后复制

参数。我们可以利用它来设置超时或取消信号。

package main

import (
    "context"
    "fmt"
    "log"
    "runtime"
    "sync"
    "time"

    "golang.org/x/sync/semaphore"
)

func processTaskWithTimeout(id int) {
    fmt.Printf("任务 %d 开始...\n", id)
    time.Sleep(time.Duration(id%3+1) * time.Second) // 模拟不同处理时间
    fmt.Printf("任务 %d 完成。\n", id)
}

func main() {
    maxConcurrency := int64(runtime.NumCPU())
    sem := semaphore.NewWeighted(maxConcurrency)

    totalTasks := 10
    var wg sync.WaitGroup

    fmt.Printf("开始处理 %d 个任务，最大并发数: %d\n", totalTasks, maxConcurrency)

    for i := 0; i < totalTasks; i++ {
        wg.Add(1)
        go func(taskID int) {
            defer wg.Done()

            // 设置一个超时Context，例如500毫秒
            ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
            defer cancel() // 确保Context被取消，释放资源

            // 尝试获取信号量，带超时
            if err := sem.Acquire(ctx, 1); err != nil {
                // 错误处理：可能是超时，也可能是Context被取消
                if err == context.DeadlineExceeded {
                    log.Printf("任务 %d: 获取信号量超时，放弃处理。\n", taskID)
                } else if err == context.Canceled {
                    log.Printf("任务 %d: 获取信号量被取消，放弃处理。\n", taskID)
                } else {
                    log.Printf("任务 %d: 获取信号量时发生未知错误: %v\n", taskID, err)
                }
                return
            }
            defer sem.Release(1) // 确保无论如何都释放信号量

            processTaskWithTimeout(taskID)
        }(i)
    }

    wg.Wait()
    fmt.Println("所有任务处理完毕。")
}

登录后复制

在这个例子中，我为每次