Go并发编程：解决Goroutine与Channel协作中的死锁问题-Golang-PHP中文网

Go并发编程：解决Goroutine与Channel协作中的死锁问题

本文深入探讨了go语言中goroutine与channel协作时可能遇到的死锁问题。通过分析一个典型的“工作者”模式示例，揭示了未正确关闭channel是导致死锁的常见原因。文章详细阐述了channel关闭机制及其对接收操作的影响，并提供了基于close()函数的解决方案。此外，还介绍了使用for range遍历channel和sync.waitgroup等go语言最佳实践，以构建更健健壮、高效的并发程序。

在Go语言中，Goroutine和Channel是实现并发的核心原语。它们提供了一种安全、高效地进行数据通信和同步的方式。然而，不恰当的使用方式也可能导致程序陷入死锁，即所有Goroutine都在等待某个事件发生而该事件永远不会发生的状态。本文将通过一个实际案例，深入剖析Go并发编程中的死锁问题及其解决方案。

理解Go并发中的死锁现象

考虑一个常见的“工作者”模式：一个主Goroutine负责将任务放入队列（Channel），多个工作Goroutine从队列中取出任务并处理。当任务全部处理完毕后，主Goroutine需要等待所有工作Goroutine完成。

以下是原始代码片段，它试图实现这样一个系统，但最终导致了死锁：

package main

import (
    "fmt"
    "sync" // 引入sync包，用于后续的WaitGroup示例
    "time" // 引入time包，用于模拟工作耗时
)

// entry 模拟任务结构
type entry struct {
    name string
}

// myQueue 模拟任务池
type myQueue struct {
    pool []*entry
    maxConcurrent int
}

// process 函数：工作Goroutine，从queue中接收任务并处理
func process(queue chan *entry, waiters chan bool) {
    for {
        // 从queue中接收任务
        entry, ok := <-queue
        // 如果channel已关闭且无更多数据，ok为false，此时应退出循环
        if !ok {
            break
        }
        fmt.Printf("worker: processing %s\n", entry.name)
        // 模拟任务处理
        time.Sleep(100 * time.Millisecond)
        entry.name = "processed_" + entry.name
    }
    fmt.Println("worker finished")
    // 通知主Goroutine本工作Goroutine已完成
    waiters <- true
}

// fillQueue 函数：主Goroutine，填充任务队列并启动工作Goroutine
func fillQueue(q *myQueue) {
    // 创建一个有缓冲的channel作为任务队列
    queue := make(chan *entry, len(q.pool))
    for _, entry := range q.pool {
        fmt.Printf("push entry: %s\n", entry.name)
        queue <- entry // 将任务推入队列
    }
    fmt.Printf("entry queue capacity: %d\n", cap(queue))

    // 确定启动的工作Goroutine数量
    totalThreads := q.maxConcurrent
    if q.maxConcurrent > len(q.pool) {
        totalThreads = len(q.pool)
    }

    // 创建一个有缓冲的channel用于接收工作Goroutine的完成信号
    waiters := make(chan bool, totalThreads)
    fmt.Printf("waiters channel capacity: %d\n", cap(waiters))

    // 启动工作Goroutine
    var threads int
    for threads = 0; threads < totalThreads; threads++ {
        fmt.Println("start worker")
        go process(queue, waiters)
    }
    fmt.Printf("threads started: %d\n", threads)

    // 等待所有工作Goroutine完成
    for ; threads > 0; threads-- {
        fmt.Println("wait for thread")
        ok := <-waiters // 阻塞等待工作Goroutine发送完成信号
        fmt.Printf("received thread end: %t\n", ok)
    }
    fmt.Println("All workers finished, fillQueue exiting.")
}

func main() {
    // 示例数据
    myQ := &myQueue{
        pool: []*entry{
            {name: "task1"},
            {name: "task2"},
            {name: "task3"},
        },
        maxConcurrent: 1, // 假设只启动一个工作Goroutine
    }
    fillQueue(myQ)
}

登录后复制

当运行上述代码时，会得到类似以下的输出和死锁错误：

push entry: task1
push entry: task2
push entry: task3
entry queue capacity: 3
waiters channel capacity: 1
start worker
threads started: 1
wait for thread
worker: processing task1
worker: processing task2
worker: processing task3
fatal error: all goroutines are asleep - deadlock!

登录后复制

死锁分析：

fillQueue函数将所有任务推入queue通道后，开始启动一个工作Goroutine（因为maxConcurrent为1）。
工作Goroutine process从queue中取出并处理了所有3个任务。
process函数中的for循环继续尝试从queue中接收数据：entry, ok := <-queue。
此时queue通道中已没有数据，且queue通道从未被关闭。这意味着<-queue操作会无限期阻塞，等待新的数据到来。
由于process Goroutine被阻塞，它永远无法执行到waiters <- true这一行。
fillQueue函数在for ; threads > 0; threads--循环中，也阻塞在ok := <-waiters，等待工作Goroutine发送完成信号。
结果是：process Goroutine等待queue通道，而fillQueue Goroutine等待waiters通道。两者都在无限期等待对方，从而导致了死锁。

死锁根源：未关闭的Channel

问题的核心在于process Goroutine无法得知queue通道中是否还有后续数据。当通道被关闭后，再尝试从通道中接收数据时，ok变量会返回false，表示通道已关闭且无更多数据。工作Goroutine正是需要这个信号来安全地退出循环。

在Go语言中，close(channel)操作用于通知接收方，该通道不再有数据发送。一旦通道关闭，再向其发送数据会引发panic，但从已关闭的通道接收数据会立即返回零值和ok=false。

解决方案：正确关闭Channel

要解决死锁，我们必须在所有任务都发送到queue通道之后，由发送方（即fillQueue函数）关闭queue通道。

豆包AI编程

豆包推出的AI编程助手

483

查看详情

package main

import (
    "fmt"
    "sync"
    "time"
)

type entry struct {
    name string
}

type myQueue struct {
    pool []*entry
    maxConcurrent int
}

// process 函数：工作Goroutine，从queue中接收任务并处理
func process(queue chan *entry, waiters chan bool) {
    for {
        entry, ok := <-queue
        if !ok { // channel已关闭且无更多数据，退出循环
            break
        }
        fmt.Printf("worker: processing %s\n", entry.name)
        time.Sleep(100 * time.Millisecond)
        entry.name = "processed_" + entry.name
    }
    fmt.Println("worker finished")
    waiters <- true // 通知主Goroutine本工作Goroutine已完成
}

// fillQueue 函数：主Goroutine，填充任务队列并启动工作Goroutine
func fillQueue(q *myQueue) {
    queue := make(chan *entry, len(q.pool))
    // 使用defer确保在fillQueue函数退出时关闭queue通道
    defer close(queue) 

    for _, entry := range q.pool {
        fmt.Printf("push entry: %s\n", entry.name)
        queue <- entry
    }
    fmt.Printf("entry queue capacity: %d\n", cap(queue))

    totalThreads := q.maxConcurrent
    if q.maxConcurrent > len(q.pool) {
        totalThreads = len(q.pool)
    }

    waiters := make(chan bool, totalThreads)
    fmt.Printf("waiters channel capacity: %d\n", cap(waiters))

    var threads int
    for threads = 0; threads < totalThreads; threads++ {
        fmt.Println("start worker")
        go process(queue, waiters)
    }
    fmt.Printf("threads started: %d\n", threads)

    for ; threads > 0; threads-- {
        fmt.Println("wait for thread")
        ok := <-waiters
        fmt.Printf("received thread end: %t\n", ok)
    }
    fmt.Println("All workers finished, fillQueue exiting.")
}

func main() {
    myQ := &myQueue{
        pool: []*entry{
            {name: "task1"},
            {name: "task2"},
            {name: "task3"},
        },
        maxConcurrent: 1, 
    }
    fillQueue(myQ)
}

登录后复制

关键改动： 在fillQueue函数中，添加了defer close(queue)。这确保了在fillQueue函数完成所有任务发送并即将退出时，queue通道会被正确关闭。一旦queue关闭，process Goroutine在接收完所有数据后，<-queue操作会返回ok=false，从而允许它退出循环并发送完成信号到waiters通道，最终解决死锁。

优化与最佳实践

除了正确关闭通道，Go语言还提供了一些更简洁和健壮的并发编程模式。

1. 使用 for range 遍历 Channel

对于消费者Goroutine，for range结构是遍历通道的更简洁方式。当通道关闭且所有值都被接收后，for range循环会自动退出。

// process 函数使用 for range 简化
func processOptimized(queue chan *entry, wg *sync.WaitGroup) {
    defer wg.Done() // 确保Goroutine完成时通知WaitGroup

    for entry := range queue { // 当queue关闭且无更多数据时，循环自动退出
        fmt.Printf("worker: processing %s\n", entry.name)
        time.Sleep(100 * time.Millisecond)
        entry.name = "processed_" + entry.name
    }
    fmt.Println("worker finished")
}

登录后复制

2. 使用 sync.WaitGroup 管理 Goroutine

手动管理waiters通道来等待所有Goroutine完成是可行的，但Go标准库提供了sync.WaitGroup这一更惯用且功能强大的工具。WaitGroup允许您等待一组Goroutine完成，而无需创建额外的通道。

// fillQueue 函数使用 WaitGroup 优化
func fillQueueOptimized(q *myQueue) {
    queue := make(chan *entry, len(q.pool))
    defer close(queue) // 确保在fillQueue退出时关闭queue通道

    var wg sync.WaitGroup // 声明一个WaitGroup

    for _, entry := range q.pool {
        fmt.Printf("push entry: %s\n", entry.name)
        queue <- entry
    }
    fmt.Printf("entry queue capacity: %d\n", cap(queue))

    totalThreads := q.maxConcurrent
    if q.maxConcurrent > len(q.pool) {
        totalThreads = len(q.pool)
    }

    for i := 0; i < totalThreads; i++ {
        wg.Add(1) // 每启动一个Goroutine，WaitGroup计数器加1
        fmt.Println("start worker")
        go processOptimized(queue, &wg) // 传入WaitGroup指针
    }
    fmt.Printf("threads started: %d\n", totalThreads)

    wg.Wait() // 阻塞直到所有Goroutine都调用了Done()
    fmt.Println("All workers finished, fillQueue exiting.")
}

func main() {
    myQ := &myQueue{
        pool: []*entry{
            {name: "task1"},
            {name: "task2"},
            {name: "task3"},
            {name: "task4"}, // 增加任务以更好地体现并发
            {name: "task5"},
        },
        maxConcurrent: 3, // 启动3个工作Goroutine
    }
    fillQueueOptimized(myQ)
}

登录后复制

sync.WaitGroup 的使用步骤：