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

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

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

x/sync/semaphore

解决方案

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

x/sync/semaphore

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

立即学习“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

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

maxConcurrency

sem.Release(1)

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

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

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

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

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

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

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

• sync.WaitGroup

  ： 它的主要作用是等待一组goroutine完成。你调用

  Add

  来增加计数，每个goroutine完成时调用

  Done

  来减少计数，最后通过

  Wait

  阻塞直到计数归零。

  WaitGroup

  关心的是“所有任务都完成了没？”，它不限制同时运行的goroutine数量。比如，你启动了100个goroutine，

  WaitGroup

  只是等着这100个都跑完，它并不会限制同时只有10个在跑。
  

  妙笔工坊

  妙笔工坊是一个集短剧解说，AI视频生成，口播数字人，小说推文生成的ai智能工具

  下载

• chan

  (通道)： 通道主要用于goroutine之间的通信，传递数据。当然，你也可以利用带缓冲的通道来实现简单的并发控制，例如，创建一个容量为N的缓冲通道，每次启动goroutine前向通道发送一个“令牌”，goroutine结束后从通道接收一个“令牌”。这种方式虽然能实现并发控制，但语义上不如信号量直接，而且如果需要处理不同“权重”的任务，实现起来会比较麻烦。通道更侧重于数据流和同步。

• x/sync/semaphore.Weighted

  (信号量)： 信号量就是专门用来限制并发数量的。它管理的是“资源许可”或者“执行槽位”。它关心的是“当前有多少个任务正在执行？我还能允许多少个新任务开始？”。它的“权重”概念尤其灵活，你可以让一个耗费资源的goroutine获取多个权重，而一个轻量级的goroutine只获取一个权重，这样就能更精细地控制资源消耗。

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

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

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

   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("所有任务处理完毕。")
}

在这个例子中，我为每次

Acquire

context

sem.Acquire

context.DeadlineExceeded

处理错误时，一个常见的模式是使用

defer sem.Release(1)

processTaskWithTimeout

在实际应用中，你可能还会有一个全局的

context.Context

context

cancel