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

package main

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

type entry struct {
    name string
}

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

// process 函数是工作者Goroutine的逻辑
func process(queue chan *entry, wg *sync.WaitGroup) {
    defer wg.Done() // 确保工作者完成后通知WaitGroup
    for {
        // 从队列中接收任务
        entry, ok := <-queue
        // 检查Channel是否已关闭且无更多数据
        if !ok {
            break // Channel已关闭，退出循环
        }
        fmt.Printf("worker: processing %s\n", entry.name)
        time.Sleep(100 * time.Millisecond) // 模拟任务处理时间
        entry.name = "processed_" + entry.name // 模拟数据修改
    }
    fmt.Println("worker finished")
}

// fillQueue 函数负责填充队列并启动工作者
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 cap: %d\n", cap(queue))

    // 启动工作者Goroutine
    var totalThreads int
    if q.maxConcurrent <= len(q.pool) {
        totalThreads = q.maxConcurrent
    } else {
        totalThreads = len(q.pool)
    }

    var wg sync.WaitGroup // 使用WaitGroup等待所有工作者完成
    fmt.Printf("starting %d workers\n", totalThreads)
    for i := 0; i < totalThreads; i++ {
        wg.Add(1) // 每次启动一个工作者，WaitGroup计数加1
        go process(queue, &wg)
    }

    // 核心问题所在：Channel 'queue' 在这里没有被关闭
    // close(queue) // 正确的解决方案应该在这里关闭queue

    fmt.Println("waiting for workers to finish...")
    wg.Wait() // 等待所有工作者完成
    fmt.Println("all workers finished.")
}

func main() {
    // 示例数据
    q := &myQueue{
        pool: []*entry{
            {name: "task1"},
            {name: "task2"},
            {name: "task3"},
        },
        maxConcurrent: 1, // 假设最大并发数为1
    }
    fillQueue(q)
}

push entry: task1
push entry: task2
push entry: task3
entry cap: 3
starting 1 workers
waiting for workers to finish...
worker: processing task1
worker: processing task2
worker: processing task3
fatal error: all goroutines are asleep - deadlock!

// ... (之前的代码保持不变)

func fillQueue(q *myQueue) {
    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 cap: %d\n", cap(queue))

    var totalThreads int
    if q.maxConcurrent <= len(q.pool) {
        totalThreads = q.maxConcurrent
    } else {
        totalThreads = len(q.pool)
    }

    var wg sync.WaitGroup
    fmt.Printf("starting %d workers\n", totalThreads)
    for i := 0; i < totalThreads; i++ {
        wg.Add(1)
        go process(queue, &wg)
    }

    // 关键修改：在所有任务发送完毕后，关闭queue Channel
    close(queue) 

    fmt.Println("waiting for workers to finish...")
    wg.Wait()
    fmt.Println("all workers finished.")
}

// ... (main函数保持不变)

// process 函数使用 for range 遍历 Channel
func process(queue chan *entry, wg *sync.WaitGroup) {
    defer wg.Done()
    for entry := range queue { // Channel关闭后，循环自动结束
        fmt.Printf("worker: processing %s\n", entry.name)
        time.Sleep(100 * time.Millisecond)
        entry.name = "processed_" + entry.name
    }
    fmt.Println("worker finished")
}