Go 并发中的死锁：原因分析与sync.WaitGroup实践-Golang-PHP中文网

Go 并发中的死锁：原因分析与sync.WaitGroup实践

本文深入探讨 Go 语言并发编程中常见的“所有 Goroutine 休眠 - 死锁！”错误。通过分析一个具体的观察者模式实现案例，详细解释了非缓冲通道的同步机制、死锁的成因以及 Goroutine 提前退出的问题。文章提供了两种有效的解决方案：正确利用通道进行多 Goroutine 同步，以及更推荐的 sync.WaitGroup 机制，并强调了避免盲目增加通道缓冲区大小的误区，旨在帮助开发者构建健壮的并发程序。

理解 Go 并发中的死锁与 Goroutine 同步

在 go 语言中，并发是通过 goroutine 和 channel 实现的。当 goroutine 之间缺乏正确的同步机制时，很容易遇到“all goroutines are asleep - deadlock!”（所有 goroutine 休眠 - 死锁！）的运行时错误。这通常意味着程序中所有的 goroutine 都处于等待状态，并且没有 goroutine 能够继续执行，从而导致程序挂起。

让我们通过一个具体的观察者模式实现来分析这种问题：

package main

import (
    "fmt"
    "time" // 引入 time 包用于模拟耗时操作
)

// Publisher 发布者
type Publisher struct {
    listeners []chan int
}

// Subscriber 订阅者
type Subscriber struct {
    Channel chan int
    Name    string
}

// Sub 订阅方法：将订阅者的通道添加到发布者的监听列表
func (p *Publisher) Sub(c chan int) {
    p.listeners = append(p.listeners, c)
}

// Pub 发布方法：向所有监听者发送消息
func (p *Publisher) Pub(m int, quit chan int) {
    for _, c := range p.listeners {
        c <- m // 向每个订阅者通道发送数据
    }
    quit <- 0 // 向 quit 通道发送信号
}

// ListenOnChannel 订阅者监听通道并处理数据
func (s *Subscriber) ListenOnChannel() {
    data := <-s.Channel // 从通道接收数据
    fmt.Printf("Name: %v; Data: %v\n", s.Name, data)
    // 模拟一些耗时操作，确保其他 Goroutine 有机会运行
    time.Sleep(10 * time.Millisecond)
}

func main() {
    quit := make(chan int) // 创建一个非缓冲的 quit 通道
    p := &Publisher{}
    subscribers := []*Subscriber{
        {Channel: make(chan int), Name: "1"},
        {Channel: make(chan int), Name: "2"},
        {Channel: make(chan int), Name: "3"},
    }

    // 注册订阅者并启动其监听 Goroutine
    for _, v := range subscribers {
        p.Sub(v.Channel)
        go v.ListenOnChannel() // 启动 Goroutine 监听
    }

    p.Pub(2, quit) // 发布消息并发送 quit 信号

    <-quit // 主 Goroutine 等待从 quit 通道接收信号
    fmt.Println("Main goroutine finished.")
}

登录后复制

运行上述代码，你会得到 fatal error: all goroutines are asleep - deadlock! 的错误。

死锁原因分析：

非缓冲通道的特性： quit := make(chan int) 创建了一个非缓冲通道。非缓冲通道要求发送方和接收方必须同时准备好才能完成通信。如果一方准备好而另一方未准备好，则会阻塞。
Pub 方法中的发送： 在 p.Pub(2, quit) 调用中，Pub 方法在向所有订阅者通道发送数据后，尝试执行 quit <- 0。此时，Pub 方法所在的 Goroutine（即 main Goroutine）尝试向 quit 通道发送数据。
main 函数中的接收： main Goroutine 紧接着在 p.Pub 调用之后执行 <-quit，尝试从 quit 通道接收数据。
循环依赖与阻塞： 问题在于，p.Pub 和 <-quit 都发生在 main Goroutine 中。当 p.Pub 尝试发送 quit <- 0 时，它会阻塞，因为它在等待一个接收者。然而，这个接收者 (<-quit) 位于同一个 main Goroutine 的后续代码中。这意味着 main Goroutine 既是发送者又是接收者，并且两者之间存在一个循环依赖：发送者在等待接收者，而接收者在等待发送者，导致相互阻塞，最终引发死锁。Go 运行时能够检测到这种所有 Goroutine 都被阻塞的情况并报错。

若移除 quit 通道：

如果我们将 quit 通道完全移除，代码如下：

package main

import (
    "fmt"
    "time"
)

type Publisher struct {
    listeners []chan int
}

type Subscriber struct {
    Channel chan int
    Name    string
}

func (p *Publisher) Sub(c chan int) {
    p.listeners = append(p.listeners, c)
}

func (p *Publisher) Pub(m int) { // 移除 quit 参数
    for _, c := range p.listeners {
        c <- m
    }
}

func (s *Subscriber) ListenOnChannel() {
    data := <-s.Channel
    fmt.Printf("Name: %v; Data: %v\n", s.Name, data)
    time.Sleep(10 * time.Millisecond) // 模拟耗时
}

func main() {
    p := &Publisher{}
    subscribers := []*Subscriber{
        {Channel: make(chan int), Name: "1"},
        {Channel: make(chan int), Name: "2"},
        {Channel: make(chan int), Name: "3"},
    }

    for _, v := range subscribers {
        p.Sub(v.Channel)
        go v.ListenOnChannel()
    }

    p.Pub(2) // 移除 quit 参数

    // 没有等待机制
    // fmt.Println("Main goroutine finished.") // 这行可能不会被执行，或者只在部分 Goroutine 完成前执行
}

登录后复制

此时程序不再报错死锁，但通常只会打印第一个订阅者的输出（或少数几个），然后程序就退出了。这是因为 main Goroutine 在启动了其他订阅者 Goroutine 并发送了消息之后，没有任何机制等待这些 Goroutine 完成它们的任务。main Goroutine 会立即执行完毕并退出，导致其他 Goroutine 可能还没来得及从通道接收数据并打印就随着程序的终止而被迫终止。

误区：盲目增加通道缓冲区大小

一个常见的错误是，为了“解决”死锁问题而盲目地增加通道的缓冲区大小，例如 quit := make(chan int, 1)。虽然在某些情况下这可能暂时避免死锁，但它并没有真正解决根本的并发逻辑问题。增加缓冲区只是将问题掩盖起来，而不是修复它。

通道缓冲通常用于性能优化，而不是作为同步问题的解决方案。事实上，在开发阶段，使用非缓冲通道通常更好，因为它能更快地暴露并发逻辑中的错误和死锁，帮助开发者及早发现问题。

因赛AIGC

因赛AIGC解决营销全链路应用场景

查看详情

解决方案一：正确使用通道进行同步

要解决 Goroutine 的同步问题，特别是当 main Goroutine 需要等待其他 Goroutine 完成时，可以修改通道的使用方式。核心思想是让每个“工作”Goroutine（本例中的订阅者 Goroutine）在完成任务后发送一个完成信号，而 main Goroutine 则等待接收所有这些信号。

package main

import (
    "fmt"
    "time"
)

type Publisher struct {
    listeners []chan int
}

type Subscriber struct {
    Channel chan int
    Name    string
}

func (p *Publisher) Sub(c chan int) {
    p.listeners = append(p.listeners, c)
}

func (p *Publisher) Pub(m int) { // Pub 方法不再负责发送 quit 信号
    for _, c := range p.listeners {
        c <- m
    }
}

// ListenOnChannel 订阅者监听通道并处理数据，完成后发送完成信号
func (s *Subscriber) ListenOnChannel(done chan struct{}) { // 接受一个 done 通道
    data := <-s.Channel
    fmt.Printf("Name: %v; Data: %v\n", s.Name, data)
    time.Sleep(10 * time.Millisecond)
    done <- struct{}{} // 完成后向 done 通道发送信号
}

func main() {
    numSubscribers := 3
    done := make(chan struct{}, numSubscribers) // 使用缓冲通道，或者在主 Goroutine中循环接收非缓冲通道
    p := &Publisher{}
    subscribers := make([]*Subscriber, numSubscribers)

    for i := 0; i < numSubscribers; i++ {
        subscribers[i] = &Subscriber{Channel: make(chan int), Name: fmt.Sprintf("%d", i+1)}
        p.Sub(subscribers[i].Channel)
        go subscribers[i].ListenOnChannel(done) // 传递 done 通道给每个订阅者 Goroutine
    }

    p.Pub(2) // 发布消息

    // 等待所有订阅者 Goroutine 完成
    for i := 0; i < numSubscribers; i++ {
        <-done // 接收每个订阅者发送的完成信号
    }
    fmt.Println("所有订阅者已处理消息，Main goroutine finished.")
}

登录后复制

在这个改进版本中：

我们创建了一个 done 通道，其缓冲区大小等于订阅者数量（或者使用非缓冲通道并在 main 中循环接收）。
每个 ListenOnChannel Goroutine 在完成其任务后，会向 done 通道发送一个空结构体 struct{} 作为完成信号。
main Goroutine 则循环 numSubscribers 次，从 done 通道接收信号。只有当接收到所有订阅者的完成信号后，main Goroutine 才会继续执行或退出。这确保了所有订阅者都有机会完成其工作。

解决方案二：使用 sync.WaitGroup (推荐)

Go 语言标准库中的 sync.WaitGroup 提供了一种更简洁、更惯用的方式来等待一组 Goroutine 完成。它是一个计数器：

Add(delta int)：增加计数器的值。
Done()：减少计数器的值（通常在 Goroutine 完成时调用）。
Wait()：阻塞直到计数器归零。

使用 sync.WaitGroup 重构上述观察者模式代码如下：

package main

import (
    "fmt"
    "sync" // 引入 sync 包
    "time"
)

type Publisher struct {
    listeners []chan int
}

type Subscriber struct {
    Channel chan int
    Name    string
}

func (p *Publisher) Sub(c chan int) {
    p.listeners = append(p.listeners, c)
}

func (p *Publisher) Pub(m int) {
    for _, c := range p.listeners {
        c <- m
    }
}

// ListenOnChannel 订阅者监听通道并处理数据，完成后调用 wg.Done()
func (s *Subscriber) ListenOnChannel(wg *sync.WaitGroup) { // 接受 WaitGroup 指针
    defer wg.Done() // 确保在 Goroutine 退出时调用 Done()，无论是否发生 panic

    data := <-s.Channel
    fmt.Printf("Name: %v; Data: %v\n", s.Name, data)
    time.Sleep(10 * time.Millisecond) // 模拟耗时操作
}

func main() {
    var wg sync.WaitGroup // 声明 WaitGroup 变量
    p := &Publisher{}
    subscribers := []*Subscriber{
        {Channel: make(chan int), Name: "1"},
        {Channel: make(chan int), Name: "2"},
        {Channel: make(chan int), Name: "3"},
    }

    for _, v := range subscribers {
        p.Sub(v.Channel)
        wg.Add(1) // 启动 Goroutine 前，计数器加 1
        go v.ListenOnChannel(&wg) // 传递 WaitGroup 的地址
    }

    p.Pub(2) // 发布消息

    wg.Wait() // 阻塞主 Goroutine，直到所有 Goroutine 都调用了 Done()，计数器归零
    fmt.Println("所有订阅者已处理消息，Main goroutine finished.")
}

登录后复制

sync.WaitGroup 优势：

简洁明了： API 简单直观，易于理解和使用。
多 Goroutine 同步： 专为等待多个 Goroutine 完成而设计，比手动管理多个通道更方便。
健壮性： defer wg.Done() 确保即使 Goroutine 内部发生运行时错误，Done() 也会被调用，避免 main Goroutine 永远等待。

总结与最佳实践

理解 Go 语言中的并发同步是编写健壮、高效程序的关键。当遇到“all goroutines are asleep - deadlock!”错误时，应首先检查通道的使用方式，特别是非缓冲通道的发送和接收逻辑。

避免死锁： 确保通道的发送和接收操作不会相互阻塞，形成循环等待。
确保 Goroutine 完成： 当 main Goroutine 依赖于其他 Goroutine 的完成时，必须有明确的同步机制来等待它们。
推荐 sync.WaitGroup： 对于等待一组 Goroutine 完成的场景，sync.WaitGroup 是 Go 语言中最推荐和惯用的解决方案。它提供了清晰的计数器模型，能够有效管理并发任务的生命周期。
谨慎使用缓冲通道： 缓冲通道是性能优化的工具，而非解决并发逻辑错误的银弹。在开发初期，非缓冲通道有助于暴露潜在的并发问题。

通过掌握这些并发同步机制，开发者可以更好地设计和实现复杂的并发程序，避免常见的死锁和竞态条件问题。

以上就是Go 并发中的死锁：原因分析与sync.WaitGroup实践的详细内容，更多请关注php中文网其它相关文章！