05-Go并发编程

39分钟阅读

5. Go并发编程

5.1 核心知识点

Goroutine

go func() {
    fmt.Println("Hello from goroutine")
}()

Channel

ch := make(chan int)
ch := make(chan int, 10)
 
ch <- 1
value := <-ch
close(ch)

Select

select {
case msg := <-ch1:
    fmt.Println("Received from ch1:", msg)
case msg := <-ch2:
    fmt.Println("Received from ch2:", msg)
case <-time.After(time.Second):
    fmt.Println("Timeout")
}

Sync包

var mu sync.Mutex
mu.Lock()
mu.Unlock()
 
var wg sync.WaitGroup
wg.Add(1)
go func() {
    defer wg.Done()
}()
wg.Wait()
 
var once sync.Once
once.Do(func() {
    fmt.Println("Execute once")
})

5.2 常见面试题

Q1: 如何使用Goroutine和Channel实现生产者-消费者模式？

解题思路：

  1. 解释生产者-消费者模式的基本概念
  2. 演示使用Channel进行通信
  3. 说明缓冲Channel和非缓冲Channel的区别
  4. 展示如何优雅地关闭Channel

生产者-消费者模式的基本概念：

生产者-消费者模式是一种常见的并发模式，用于解耦数据的产生和处理。

核心概念：

  • 生产者：负责产生数据（任务）
  • 消费者：负责处理数据（任务）
  • Channel：生产者和消费者之间的通信管道
  • 缓冲区：临时存储数据，平衡生产者和消费者的速度

通俗理解：

生产者-消费者就像"餐厅"：

生产者 = 厨师
- 负责做菜
- 把菜放到传菜口

消费者 = 服务员
- 从传菜口取菜
- 把菜端给客人

Channel = 传菜口
- 连接厨师和服务员
- 临时存放菜品

缓冲区 = 传菜口的容量
- 可以放几道菜
- 平衡厨师和服务员的速度

多个Goroutine之间的通信：

在Go中，多个goroutine之间通过channel进行通信，这是Go并发编程的核心思想。

通信流程：

生产者1 ──┐
           ├──→ Channel ──→ 消费者1
生产者2 ──┤              ├──→ 消费者2
生产者3 ──┘              └──→ 消费者3

详细流程：

  1. 生产者发送数据：

    • 生产者1向channel发送数据1
    • 生产者2向channel发送数据2
    • 生产者3向channel发送数据3

  2. Channel接收数据：

    • 消费者1从channel接收数据
    • 消费者2从channel接收数据
    • 消费者3从channel接收数据

  3. 数据流动：

    • 数据从生产者流向channel
    • 数据从channel流向消费者
    • channel作为缓冲区，平衡生产者和消费者的速度

缓冲Channel vs 非缓冲Channel：

缓冲Channel：

// 缓冲Channel：有容量，可以存储多个值
ch := make(chan int, 10)
 
// 特点：
// - 发送方不会阻塞（除非channel满了）
// - 接收方不会阻塞（除非channel空了）
// - 可以作为缓冲区，平衡生产者和消费者的速度
 
// 示例：
func main() {
    ch := make(chan int, 3)  // 缓冲大小为3
    
    ch <- 1  // 不会阻塞
    ch <- 2  // 不会阻塞
    ch <- 3  // 不会阻塞
    
    // ch <- 4  // 会阻塞，因为channel满了
    
    fmt.Println(<-ch)  // 输出1
    fmt.Println(<-ch)  // 输出2
    fmt.Println(<-ch)  // 输出3
}

非缓冲Channel：

// 非缓冲Channel：没有容量，只能存储一个值
ch := make(chan int)
 
// 特点：
// - 发送方会阻塞，直到有接收方
// - 接收方会阻塞，直到有发送方
// - 同步通信，保证数据被接收
 
// 示例：
func main() {
    ch := make(chan int)
    
    go func() {
        ch <- 1  // 会阻塞，直到有接收方
        fmt.Println("发送完成")
    }()
    
    time.Sleep(1 * time.Second)  // 模拟延迟
    fmt.Println(<-ch)  // 接收数据
}

代码实现：

基础版本：

package main
 
import (
    "fmt"
    "sync"
    "time"
)
 
// 生产者：产生任务
func producer(id int, jobs chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for i := 1; i <= 5; i++ {
        job := (id-1)*5 + i  // 生成唯一任务ID
        fmt.Printf("生产者 %d: 生产任务 %d\n", id, job)
        jobs <- job  // 发送任务到jobs channel
        time.Sleep(100 * time.Millisecond)
    }
    fmt.Printf("生产者 %d: 退出\n", id)
}
 
// 消费者：处理任务
func consumer(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {  // 从jobs channel读取任务
        result := job * 2
        fmt.Printf("消费者 %d: 处理任务 %d\n", id, job)
        results <- result  // 发送结果到results channel
        time.Sleep(150 * time.Millisecond)
    }
    fmt.Printf("消费者 %d: 退出\n", id)
}
 
func main() {
    // 创建channel
    jobs := make(chan int, 100)      // 任务channel
    results := make(chan int, 100)    // 结果channel
    
    var producerWg, consumerWg sync.WaitGroup
    
    // 启动3个生产者
    for i := 1; i <= 3; i++ {
        producerWg.Add(1)
        go producer(i, jobs, &producerWg)
    }
    
    // 启动2个消费者
    for i := 1; i <= 2; i++ {
        consumerWg.Add(1)
        go consumer(i, jobs, results, &consumerWg)
    }
    
    // 等待所有生产者完成，然后关闭jobs channel
    go func() {
        producerWg.Wait()
        close(jobs)
    }()
    
    // 等待所有消费者完成，然后关闭results channel
    go func() {
        consumerWg.Wait()
        close(results)
    }()
    
    // 收集结果
    for result := range results {
        fmt.Printf("主程序: 收到结果 %d\n", result)
    }
    
    fmt.Println("所有任务完成！")
}

详细版本（带注释）：

package main
 
import (
    "fmt"
    "sync"
    "time"
)
 
// 生产者：产生任务
func producer(id int, jobs chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()  // 通知WaitGroup，生产者完成了
    
    // 模拟生产5个任务
    for i := 1; i <= 5; i++ {
        job := (id-1)*5 + i  // 生成唯一任务ID
        fmt.Printf("生产者 %d: 生产任务 %d\n", id, job)
        
        // 发送任务到jobs channel
        jobs <- job
        
        time.Sleep(100 * time.Millisecond)
    }
    
    fmt.Printf("生产者 %d: 退出\n", id)
}
 
// 消费者：处理任务
func consumer(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()  // 通知WaitGroup，消费者完成了
    
    // 使用range遍历jobs channel
    // 当jobs channel关闭时，range会自动退出
    for job := range jobs {
        result := job * 2
        fmt.Printf("消费者 %d: 处理任务 %d\n", id, job)
        
        // 发送结果到results channel
        results <- result
        
        time.Sleep(150 * time.Millisecond)
    }
    
    fmt.Printf("消费者 %d: 退出\n", id)
}
 
func main() {
    // 创建channel
    jobs := make(chan int, 100)      // 任务channel，缓冲大小100
    results := make(chan int, 100)    // 结果channel，缓冲大小100
    
    var producerWg, consumerWg sync.WaitGroup
    
    // 启动3个生产者
    // 每个生产者都会发送任务到jobs channel
    for i := 1; i <= 3; i++ {
        producerWg.Add(1)  // 增加生产者计数
        go producer(i, jobs, &producerWg)
    }
    
    // 启动2个消费者
    // 每个消费者从jobs channel读取任务
    // 处理完后发送到results channel
    for i := 1; i <= 2; i++ {
        consumerWg.Add(1)
        go consumer(i, jobs, results, &consumerWg)
    }
    
    // 启动一个goroutine，等待所有生产者完成
    // 然后关闭jobs channel
    go func() {
        producerWg.Wait()  // 等待所有生产者完成
        close(jobs)        // 关闭jobs channel
        fmt.Println("jobs channel已关闭")
    }()
    
    // 启动一个goroutine，等待所有消费者完成
    // 然后关闭results channel
    go func() {
        consumerWg.Wait()  // 等待所有消费者完成
        close(results)      // 关闭results channel
        fmt.Println("results channel已关闭")
    }()
    
    // 收集结果
    // 使用range遍历results channel
    // 当results channel关闭时，range会自动退出
    for result := range results {
        fmt.Printf("主程序: 收到结果 %d\n", result)
    }
    
    fmt.Println("所有任务完成！")
}

通信流程详解：

主程序启动
    ↓
创建3个生产者goroutine
    ↓
创建2个消费者goroutine
    ↓
生产者1 ──┐
           ├──→ jobs channel ──→ 消费者1 ──┐
生产者2 ──┤                            ├──→ results channel ──→ 主程序
生产者3 ──┘              消费者2 ──┘

详细流程：

  1. 主程序启动：

    • 创建jobs channel（缓冲大小100）
    • 创建results channel（缓冲大小100）
    • 创建2个WaitGroup（producerWg和consumerWg）

  2. 启动生产者：

    • 启动3个生产者goroutine
    • 每个生产者从jobs channel接收任务
    • 处理完成后通知producerWg

  3. 启动消费者：

    • 启动2个消费者goroutine
    • 每个消费者处理5个任务
    • 将结果发送到results channel
    • 处理完成后通知consumerWg

  4. 关闭jobs channel：

    • 启动一个goroutine等待所有生产者完成
    • 所有生产者完成后，关闭jobs channel
    • 生产者会自动退出（因为range检测到channel关闭）

  5. 关闭results channel：

    • 启动一个goroutine等待所有消费者完成
    • 所有消费者完成后，关闭results channel
    • 主程序会自动退出（因为range检测到channel关闭）

  6. 收集结果：

    • 主程序使用range遍历results channel
    • 收集所有结果
    • results channel关闭后，range自动退出

优雅地关闭Channel：

关闭Channel的规则：

  1. 只在发送方关闭Channel：

    • 接收方不应该关闭Channel
    • 多个发送方不应该同时关闭Channel

  2. 使用WaitGroup同步：

    • 等待所有发送方完成
    • 然后关闭Channel

  3. 使用range遍历Channel：

    • range会自动检测Channel关闭
    • Channel关闭后，range会自动退出

常见错误分析：

  • 错误1：忘记关闭Channel导致死锁
  • 错误2：在多个goroutine中关闭Channel导致panic
  • 错误3：不理解range和Channel的配合使用

Q2: 如何使用Select实现多路复用？

解题思路：

  1. 解释Select的基本概念和用途
  2. 演示Select的随机选择特性
  3. 说明Select的default分支
  4. 展示Select的超时处理

Select的基本概念：

Select是Go语言中用于处理多个Channel操作的并发原语，可以同时监听多个Channel，并在其中任意一个Channel就绪时执行对应的操作。

核心概念：

  • 多路复用：同时监听多个Channel
  • 随机选择：多个Channel同时就绪时，随机选择一个
  • 阻塞等待：没有Channel就绪时，阻塞等待
  • default分支：没有Channel就绪时，执行default分支

通俗理解：

Select就像"多路选择器"：

场景：你在等电话、短信、邮件
- 电话响了：接电话
- 短信来了：看短信
- 邮件来了：查邮件
- 都没有：继续等待

Select的作用：
- 同时监听电话、短信、邮件
- 哪个先来，就处理哪个
- 都没有来，就继续等待

Select的随机选择特性：

当多个Channel同时就绪时，Select会随机选择一个执行。

// 示例：随机选择
package main
 
import (
    "fmt"
    "time"
)
 
func main() {
    ch1 := make(chan string, 1)
    ch2 := make(chan string, 1)
    
    // 同时向两个channel发送数据
    ch1 <- "from ch1"
    ch2 <- "from ch2"
    
    // Select会随机选择一个
    select {
    case msg := <-ch1:
        fmt.Println("Received from ch1:", msg)
    case msg := <-ch2:
        fmt.Println("Received from ch2:", msg)
    }
}

输出示例：

第一次运行：Received from ch1: from ch1
第二次运行：Received from ch2: from ch2
第三次运行：Received from ch1: from ch1

Select的default分支：

当所有Channel都没有就绪时，执行default分支。

// 示例：default分支
package main
 
import (
    "fmt"
    "time"
)
 
func main() {
    ch := make(chan string)
    
    // Select带default分支
    select {
    case msg := <-ch:
        fmt.Println("Received:", msg)
    default:
        fmt.Println("No message, doing other work")
    }
    
    // 向channel发送数据
    go func() {
        time.Sleep(100 * time.Millisecond)
        ch <- "hello"
    }()
    
    // 再次Select
    select {
    case msg := <-ch:
        fmt.Println("Received:", msg)
    default:
        fmt.Println("No message, doing other work")
    }
}

输出：

No message, doing other work
Received: hello

Select的超时处理：

使用time.After()实现超时处理。

// 示例：超时处理
package main
 
import (
    "fmt"
    "time"
)
 
func main() {
    ch := make(chan string)
    
    // 启动一个goroutine，延迟发送数据
    go func() {
        time.Sleep(2 * time.Second)
        ch <- "hello"
    }()
    
    // Select带超时
    select {
    case msg := <-ch:
        fmt.Println("Received:", msg)
    case <-time.After(1 * time.Second):
        fmt.Println("Timeout waiting for message")
    }
}

输出：

Timeout waiting for message

代码实现：

基础版本：

package main
 
import (
    "fmt"
    "time"
)
 
func worker(id int, jobs <-chan int, results chan<- int) {
    for job := range jobs {
        fmt.Printf("Worker %d: processing job %d\n", id, job)
        time.Sleep(time.Duration(job) * 100 * time.Millisecond)
        results <- job * 2
    }
}
 
func main() {
    jobs := make(chan int, 10)
    results := make(chan int, 10)
    
    // 启动3个worker
    for i := 1; i <= 3; i++ {
        go worker(i, jobs, results)
    }
    
    // 发送任务
    go func() {
        for i := 1; i <= 9; i++ {
            jobs <- i
        }
        close(jobs)
    }()
    
    // 使用Select接收结果，带超时
    for i := 1; i <= 9; i++ {
        select {
        case result := <-results:
            fmt.Printf("Result: %d\n", result)
        case <-time.After(500 * time.Millisecond):
            fmt.Println("Timeout waiting for result")
        }
    }
    
    // 演示多路复用
    ch1 := make(chan string, 1)
    ch2 := make(chan string, 1)
    
    go func() {
        time.Sleep(100 * time.Millisecond)
        ch1 <- "from ch1"
    }()
    
    go func() {
        time.Sleep(200 * time.Millisecond)
        ch2 <- "from ch2"
    }()
    
    // 使用Select监听多个channel
    for i := 0; i < 2; i++ {
        select {
        case msg := <-ch1:
            fmt.Println("Received from ch1:", msg)
        case msg := <-ch2:
            fmt.Println("Received from ch2:", msg)
        case <-time.After(300 * time.Millisecond):
            fmt.Println("Timeout")
        }
    }
}

详细版本（带注释）：

package main
 
import (
    "fmt"
    "time"
)
 
// Worker：处理任务
func worker(id int, jobs <-chan int, results chan<- int) {
    for job := range jobs {
        fmt.Printf("Worker %d: 处理任务 %d\n", id, job)
        time.Sleep(time.Duration(job) * 100 * time.Millisecond)
        results <- job * 2
    }
}
 
func main() {
    // 创建channel
    jobs := make(chan int, 10)
    results := make(chan int, 10)
    
    // 启动3个worker
    // 每个worker都会从jobs channel接收任务
    // 处理完成后将结果发送到results channel
    for i := 1; i <= 3; i++ {
        go worker(i, jobs, results)
    }
    
    // 发送任务
    go func() {
        for i := 1; i <= 9; i++ {
            jobs <- i
        }
        close(jobs)
    }()
    
    // 使用Select接收结果，带超时
    // 如果500ms内没有收到结果，输出超时
    for i := 1; i <= 9; i++ {
        select {
        case result := <-results:
            fmt.Printf("结果: %d\n", result)
        case <-time.After(500 * time.Millisecond):
            fmt.Println("等待结果超时")
        }
    }
    
    // 演示多路复用
    // 创建两个channel
    ch1 := make(chan string, 1)
    ch2 := make(chan string, 1)
    
    // 向ch1发送数据（100ms后）
    go func() {
        time.Sleep(100 * time.Millisecond)
        ch1 <- "from ch1"
    }()
    
    // 向ch2发送数据（200ms后）
    go func() {
        time.Sleep(200 * time.Millisecond)
        ch2 <- "from ch2"
    }()
    
    // 使用Select监听多个channel
    // 哪个channel先有数据，就处理哪个
    for i := 0; i < 2; i++ {
        select {
        case msg := <-ch1:
            fmt.Println("从ch1收到:", msg)
        case msg := <-ch2:
            fmt.Println("从ch2收到:", msg)
        case <-time.After(300 * time.Millisecond):
            fmt.Println("超时")
        }
    }
}

Select的多路复用流程：

主程序启动
    ↓
创建3个worker goroutine
    ↓
发送9个任务到jobs channel
    ↓
使用Select监听results channel
    ↓
    ├─→ results channel有数据 → 接收结果
    ├─→ 超时 → 输出超时
    └─→ 继续等待

Select的最佳实践：

  1. 使用Select处理超时：
// ❌ 不好的做法：没有超时处理
select {
case msg := <-ch:
    fmt.Println("Received:", msg)
}
 
// ✅ 好的做法：使用Select处理超时
select {
case msg := <-ch:
    fmt.Println("Received:", msg)
case <-time.After(1 * time.Second):
    fmt.Println("Timeout")
}
  1. 使用Select的default分支：
// ❌ 不好的做法：阻塞等待
msg := <-ch
fmt.Println("Received:", msg)
 
// ✅ 好的做法：使用Select的default分支
select {
case msg := <-ch:
    fmt.Println("Received:", msg)
default:
    fmt.Println("No message")
}
  1. 避免在Select中使用nil Channel：
// ❌ 不好的做法：在Select中使用nil Channel
var ch1 chan string
select {
case msg := <-ch1:  // ch1是nil，永远不会执行
    fmt.Println("Received:", msg)
}
 
// ✅ 好的做法：初始化Channel
ch1 := make(chan string)
select {
case msg := <-ch1:
    fmt.Println("Received:", msg)
}

常见错误分析：

  • 错误1：不理解Select的随机选择特性
  • 错误2：忘记处理超时导致阻塞
  • 错误3：在Select中使用nil Channel

Q3: 如何避免Goroutine泄漏？

解题思路：

  1. 解释Goroutine泄漏的原因
  2. 演示如何使用context控制Goroutine生命周期
  3. 说明如何正确关闭Channel
  4. 展示如何监控Goroutine数量

Goroutine泄漏的原因：

Goroutine泄漏是指Goroutine没有正常退出，一直占用内存和资源。

常见原因：

  • 没有退出条件：Goroutine无限循环，没有退出条件
  • Channel阻塞：Goroutine等待Channel，但Channel永远不会关闭
  • Context未使用：Goroutine没有使用Context，无法被取消
  • 资源未释放：Goroutine持有资源，但无法释放

通俗理解：

Goroutine泄漏就像"员工不离职"：

场景：公司雇佣员工
- 员工1：一直在工作，没有下班时间
- 员工2：等待任务，但任务永远不会来
- 员工3：持有公司资源，但无法归还

问题：
- 员工一直占用工资
- 员工一直占用工位
- 员工一直占用资源

解决：
- 设置下班时间（Context）
- 任务完成后通知（关闭Channel）
- 监控员工数量（监控Goroutine）

使用Context控制Goroutine生命周期：

Context是Go语言中用于控制Goroutine生命周期的标准方式。

核心概念：

  • Context：携带取消信号、截止时间、键值对
  • WithCancel：创建可取消的Context
  • WithTimeout：创建带超时的Context
  • WithDeadline：创建带截止时间的Context

Context的使用：

// 示例：使用Context控制Goroutine
package main
 
import (
    "context"
    "fmt"
    "time"
)
 
func worker(ctx context.Context, id int) {
    for {
        select {
        case <-ctx.Done():
            fmt.Printf("Worker %d: %v\n", id, ctx.Err())
            return
        default:
            fmt.Printf("Worker %d: working\n", id)
            time.Sleep(500 * time.Millisecond)
        }
    }
}
 
func main() {
    // 创建可取消的Context
    ctx, cancel := context.WithCancel(context.Background())
    
    // 启动worker
    go worker(ctx, 1)
    
    // 2秒后取消
    time.Sleep(2 * time.Second)
    cancel()
    
    time.Sleep(1 * time.Second)
    fmt.Println("Main: done")
}

输出：

Worker 1: working
Worker 1: working
Worker 1: working
Worker 1: working
Worker 1: context canceled
Main: done

Context的超时控制：

// 示例：使用Context控制超时
package main
 
import (
    "context"
    "fmt"
    "time"
)
 
func worker(ctx context.Context, id int) {
    for {
        select {
        case <-ctx.Done():
            fmt.Printf("Worker %d: %v\n", id, ctx.Err())
            return
        default:
            fmt.Printf("Worker %d: working\n", id)
            time.Sleep(500 * time.Millisecond)
        }
    }
}
 
func main() {
    // 创建带超时的Context
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()
    
    // 启动worker
    go worker(ctx, 1)
    
    // 等待worker退出
    time.Sleep(3 * time.Second)
    fmt.Println("Main: done")
}

输出：

Worker 1: working
Worker 1: working
Worker 1: working
Worker 1: context deadline exceeded
Main: done

正确关闭Channel：

关闭Channel是防止Goroutine泄漏的重要方式。

关闭Channel的规则：

  1. 只在发送方关闭Channel：

    • 接收方不应该关闭Channel
    • 多个发送方不应该同时关闭Channel

  2. 使用range遍历Channel：

    • range会自动检测Channel关闭
    • Channel关闭后，range会自动退出

  3. 使用WaitGroup同步：

    • 等待所有发送方完成
    • 然后关闭Channel

示例：

// 示例：正确关闭Channel
package main
 
import (
    "fmt"
    "sync"
)
 
func producer(id int, jobs <-chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    for i := 1; i <= 5; i++ {
        jobs <- i
        fmt.Printf("Producer %d: sent job %d\n", id, i)
    }
}
 
func consumer(id int, jobs <-chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {
        fmt.Printf("Consumer %d: received job %d\n", id, job)
    }
    fmt.Printf("Consumer %d: jobs channel closed\n", id)
}
 
func main() {
    jobs := make(chan int, 10)
    
    var producerWg, consumerWg sync.WaitGroup
    
    // 启动生产者
    for i := 1; i <= 3; i++ {
        producerWg.Add(1)
        go producer(i, jobs, &producerWg)
    }
    
    // 启动消费者
    for i := 1; i <= 2; i++ {
        consumerWg.Add(1)
        go consumer(i, jobs, &consumerWg)
    }
    
    // 等待生产者完成，然后关闭jobs channel
    go func() {
        producerWg.Wait()
        close(jobs)
    }()
    
    // 等待消费者完成
    consumerWg.Wait()
    fmt.Println("All done")
}

监控Goroutine数量：

监控Goroutine数量可以帮助发现Goroutine泄漏。

// 示例：监控Goroutine数量
package main
 
import (
    "fmt"
    "runtime"
    "time"
)
 
func monitorGoroutines() {
    ticker := time.NewTicker(1 * time.Second)
    defer ticker.Stop()
    
    for range ticker.C {
        fmt.Printf("Goroutines: %d\n", runtime.NumGoroutine())
    }
}
 
func main() {
    // 启动监控
    go monitorGoroutines()
    
    // 启动一些goroutine
    for i := 1; i <= 10; i++ {
        go func(id int) {
            time.Sleep(2 * time.Second)
            fmt.Printf("Goroutine %d: exiting\n", id)
        }(i)
    }
    
    // 等待goroutine退出
    time.Sleep(3 * time.Second)
    fmt.Println("All done")
}

输出：

Goroutines: 11
Goroutines: 11
Goroutine 1: exiting
Goroutine 2: exiting
...
Goroutine 10: exiting
Goroutines: 1
All done

代码实现：

基础版本：

package main
 
import (
    "context"
    "fmt"
    "runtime"
    "sync"
    "time"
)
 
// Worker：处理任务
func worker(ctx context.Context, id int, jobs <-chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    
    for {
        select {
        case <-ctx.Done():
            // Context被取消，退出
            fmt.Printf("Worker %d: %v\n", id, ctx.Err())
            return
        case job, ok := <-jobs:
            if !ok {
                // Channel关闭，退出
                fmt.Printf("Worker %d: jobs channel closed\n", id)
                return
            }
            fmt.Printf("Worker %d: processing job %d\n", id, job)
            time.Sleep(100 * time.Millisecond)
        }
    }
}
 
// 监控Goroutine数量
func monitorGoroutines() {
    ticker := time.NewTicker(1 * time.Second)
    defer ticker.Stop()
    
    for range ticker.C {
        fmt.Printf("Goroutines: %d\n", runtime.NumGoroutine())
    }
}
 
func main() {
    // 创建可取消的Context
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    
    jobs := make(chan int, 10)
    
    var wg sync.WaitGroup
    
    // 启动5个worker
    for i := 1; i <= 5; i++ {
        wg.Add(1)
        go worker(ctx, i, jobs, &wg)
    }
    
    // 启动监控
    go monitorGoroutines()
    
    // 发送任务
    go func() {
        for i := 1; i <= 20; i++ {
            jobs <- i
            time.Sleep(50 * time.Millisecond)
        }
        close(jobs)
    }()
    
    // 2秒后取消
    time.Sleep(2 * time.Second)
    cancel()
    
    // 等待worker退出
    wg.Wait()
    time.Sleep(1 * time.Second)
    fmt.Println("All workers stopped")
}

详细版本（带注释）：

package main
 
import (
    "context"
    "fmt"
    "runtime"
    "sync"
    "time"
)
 
// Worker：处理任务
func worker(ctx context.Context, id int, jobs <-chan int, wg *sync.WaitGroup) {
    defer wg.Done()  // 通知WaitGroup，worker完成了
    
    for {
        select {
        case <-ctx.Done():
            // Context被取消，退出
            // 可能是调用cancel()或超时
            fmt.Printf("Worker %d: %v\n", id, ctx.Err())
            return
        case job, ok := <-jobs:
            if !ok {
                // Channel关闭，退出
                // 使用range遍历channel时，channel关闭会自动退出
                // 但这里使用select，需要手动检查
                fmt.Printf("Worker %d: jobs channel closed\n", id)
                return
            }
            // 处理任务
            fmt.Printf("Worker %d: 处理任务 %d\n", id, job)
            time.Sleep(100 * time.Millisecond)
        }
    }
}
 
// 监控Goroutine数量
func monitorGoroutines() {
    ticker := time.NewTicker(1 * time.Second)
    defer ticker.Stop()
    
    // 每秒输出一次Goroutine数量
    for range ticker.C {
        fmt.Printf("Goroutines: %d\n", runtime.NumGoroutine())
    }
}
 
func main() {
    // 创建可取消的Context
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()  // 确保main退出时取消所有worker
    
    jobs := make(chan int, 10)
    
    var wg sync.WaitGroup
    
    // 启动5个worker
    // 每个worker都会监听ctx.Done()和jobs channel
    for i := 1; i <= 5; i++ {
        wg.Add(1)  // 增加worker计数
        go worker(ctx, i, jobs, &wg)
    }
    
    // 启动监控
    go monitorGoroutines()
    
    // 发送任务
    go func() {
        for i := 1; i <= 20; i++ {
            jobs <- i
            time.Sleep(50 * time.Millisecond)
        }
        close(jobs)  // 所有任务发送完毕，关闭jobs channel
    }()
    
    // 2秒后取消
    // 这会触发所有worker的ctx.Done()
    time.Sleep(2 * time.Second)
    cancel()
    
    // 等待worker退出
    wg.Wait()
    
    // 等待监控输出
    time.Sleep(1 * time.Second)
    fmt.Println("所有worker已停止")
}

Goroutine泄漏的预防措施：

  1. 使用Context控制Goroutine生命周期：
// ❌ 不好的做法：Goroutine没有退出条件
func worker() {
    for {
        // 无限循环，没有退出条件
        fmt.Println("Working...")
        time.Sleep(100 * time.Millisecond)
    }
}
 
// ✅ 好的做法：使用Context控制Goroutine生命周期
func worker(ctx context.Context) {
    for {
        select {
        case <-ctx.Done():
            return  // Context被取消，退出
        default:
            fmt.Println("Working...")
            time.Sleep(100 * time.Millisecond)
        }
    }
}
  1. 正确关闭Channel：
// ❌ 不好的做法：忘记关闭Channel
func producer(jobs <-chan int) {
    for i := 1; i <= 10; i++ {
        jobs <- i
    }
    // 忘记关闭jobs channel
    // consumer会一直阻塞
}
 
// ✅ 好的做法：正确关闭Channel
func producer(jobs <-chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    for i := 1; i <= 10; i++ {
        jobs <- i
    }
    close(jobs)  // 关闭jobs channel
    // consumer会自动退出
}
  1. 监控Goroutine数量：
// ❌ 不好的做法：不监控Goroutine数量
func main() {
    for i := 0; i < 100; i++ {
        go func() {
            time.Sleep(1 * time.Hour)
        }()
    }
    // 不知道有多少Goroutine在运行
}
 
// ✅ 好的做法：监控Goroutine数量
func main() {
    go func() {
        ticker := time.NewTicker(1 * time.Second)
        defer ticker.Stop()
        for range ticker.C {
            fmt.Printf("Goroutines: %d\n", runtime.NumGoroutine())
        }
    }()
    
    for i := 0; i < 100; i++ {
        go func() {
            time.Sleep(1 * time.Hour)
        }()
    }
}

常见错误分析：

  • 错误1：Goroutine没有退出条件导致泄漏
  • 错误2：忘记关闭Channel导致Goroutine阻塞
  • 错误3：不理解context的使用

Q4: 如何实现工作池（Worker Pool）？

解题思路：

  1. 解释工作池的基本概念
  2. 演示如何创建固定数量的Worker
  3. 说明如何分配任务给Worker
  4. 展示如何优雅地关闭工作池

工作池的基本概念：

工作池（Worker Pool）是一种并发模式，用于管理和复用固定数量的工作线程（Goroutine）来处理任务队列。

核心概念：

  • Worker：工作线程，负责处理任务的Goroutine
  • 任务队列：存放待处理任务的Channel
  • 结果队列：存放处理结果的Channel
  • 工作池：管理多个Worker的容器

通俗理解：

工作池就像"快递公司"：

Worker = 快递员
- 固定数量的快递员
- 负责配送包裹

任务队列 = 待配送的包裹
- 存放所有待处理的包裹
- 快递员从这里取包裹

结果队列 = 已配送的包裹
- 存放已完成的配送记录

工作池 = 快递公司
- 管理所有快递员
- 分配任务给快递员
- 监控配送状态

为什么要使用工作池：

  1. 资源控制：限制并发Goroutine的数量，避免创建过多的Goroutine导致资源耗尽
  2. 任务复用：复用Goroutine，避免频繁创建和销毁的开销
  3. 负载均衡：合理分配任务给Worker，提高整体处理效率
  4. 优雅关闭：统一管理Worker的生命周期，确保资源正确释放

工作池的应用场景：

  1. 批量处理任务：如批量处理图片、批量发送邮件
  2. IO密集型任务：如网络请求、文件读写
  3. CPU密集型任务：如数据处理、计算任务
  4. 限流控制：如API请求限流、数据库操作限流

工作池的优势和劣势：

优势：

  • 资源可控：固定数量的Goroutine，避免资源耗尽
  • 性能稳定：复用Goroutine，减少创建销毁开销
  • 易于管理：统一管理Worker的生命周期
  • 负载均衡：合理分配任务给Worker

劣势：

  • 复杂度增加：需要管理Worker的状态和生命周期
  • 调试困难：多个Goroutine并发执行，调试和排错困难
  • 适用性限制：不适合所有场景，需要根据任务特点选择

工作池的实现细节：

1. Worker的实现：

Worker是工作池的基本单元，负责从任务队列中获取任务并处理。

type Worker struct {
    ID       int           // Worker的ID
    JobQueue chan Job      // 任务队列
    Result   chan Result   // 结果队列
    Quit     chan bool     // 退出信号
}
 
func (w *Worker) Start() {
    go func() {
        for {
            select {
            case job := <-w.JobQueue:
                // 处理任务
                result := w.processJob(job)
                w.Result <- result
            case <-w.Quit:
                // 退出Worker
                return
            }
        }
    }()
}
 
func (w *Worker) processJob(job Job) Result {
    // 处理任务的具体逻辑
    return Result{
        JobID:  job.ID,
        Output: fmt.Sprintf("Processed %s", job.Data),
        Error:  nil,
    }
}

关键点：

  • Worker是一个无限循环的Goroutine
  • 使用select监听任务队列和退出信号
  • 处理任务后将结果发送到结果队列
  • 收到退出信号后退出循环

2. 工作池的实现：

工作池管理多个Worker，负责启动、停止和任务分配。

type WorkerPool struct {
    Workers   []*Worker     // Worker列表
    JobQueue  chan Job      // 任务队列
    Result    chan Result   // 结果队列
    Quit      chan bool     // 退出信号
    WorkerWg  sync.WaitGroup // 用于等待所有Worker退出
}
 
func NewWorkerPool(numWorkers int) *WorkerPool {
    jobQueue := make(chan Job, 100)
    result := make(chan Result, 100)
    
    workers := make([]*Worker, numWorkers)
    for i := 0; i < numWorkers; i++ {
        workers[i] = NewWorker(i+1, jobQueue, result)
    }
    
    return &WorkerPool{
        Workers:  workers,
        JobQueue: jobQueue,
        Result:   result,
        Quit:     make(chan bool),
    }
}
 
func (wp *WorkerPool) Start() {
    for _, worker := range wp.Workers {
        worker.Start()
        wp.WorkerWg.Add(1)
    }
}
 
func (wp *WorkerPool) Stop() {
    close(wp.JobQueue)
    for _, worker := range wp.Workers {
        worker.Stop()
    }
    close(wp.Quit)
}
 
func (wp *WorkerPool) AddJob(job Job) {
    wp.JobQueue <- job
}
 
func (wp *WorkerPool) GetResults() <-chan Result {
    return wp.Result
}

关键点：

  • 工作池创建固定数量的Worker
  • 所有Worker共享同一个任务队列和结果队列
  • 启动工作池时启动所有Worker
  • 停止工作池时关闭任务队列并发送退出信号

3. 任务分配机制：

任务分配通过Channel实现，多个Worker竞争获取任务。

// 任务分配流程
// 1. 主程序将任务发送到任务队列
pool.AddJob(job)
 
// 2. 多个Worker竞争获取任务
// Worker 1: job := <-w.JobQueue
// Worker 2: job := <-w.JobQueue
// Worker 3: job := <-w.JobQueue
 
// 3. Worker处理任务并发送结果
w.Result <- result
 
// 4. 主程序从结果队列获取结果
result := <-pool.GetResults()

关键点：

  • 使用Channel实现任务分配，天然支持并发
  • 多个Worker竞争获取任务，实现负载均衡
  • 结果队列用于收集处理结果

工作池的优化方向：

1. 动态调整Worker数量：

根据任务队列的长度动态调整Worker的数量。

func (wp *WorkerPool) adjustWorkers() {
    for {
        select {
        case <-time.After(time.Second):
            if len(wp.JobQueue) > 50 && len(wp.Workers) < 10 {
                // 任务队列过长，增加Worker
                wp.addWorker()
            } else if len(wp.JobQueue) < 10 && len(wp.Workers) > 3 {
                // 任务队列过短，减少Worker
                wp.removeWorker()
            }
        case <-wp.Quit:
            return
        }
    }
}
 
func (wp *WorkerPool) addWorker() {
    worker := NewWorker(len(wp.Workers)+1, wp.JobQueue, wp.Result)
    worker.Start()
    wp.Workers = append(wp.Workers, worker)
}
 
func (wp *WorkerPool) removeWorker() {
    if len(wp.Workers) > 0 {
        worker := wp.Workers[len(wp.Workers)-1]
        worker.Stop()
        wp.Workers = wp.Workers[:len(wp.Workers)-1]
    }
}

2. 任务优先级：

实现优先级队列，优先处理重要任务。

type PriorityJob struct {
    Job
    Priority int
}
 
type PriorityQueue chan PriorityJob
 
func (pq PriorityQueue) Len() int { return len(pq) }
 
func (pq PriorityQueue) Less(i, j int) bool {
    return pq[i].Priority > pq[j].Priority
}
 
func (pq PriorityQueue) Swap(i, j int) {
    pq[i], pq[j] = pq[j], pq[i]
}
 
func (pq *PriorityQueue) Push(x interface{}) {
    *pq = append(*pq, x.(PriorityJob))
}
 
func (pq *PriorityQueue) Pop() interface{} {
    old := *pq
    n := len(old)
    item := old[n-1]
    *pq = old[0 : n-1]
    return item
}

3. 任务超时处理：

为任务设置超时时间，避免长时间阻塞。

func (w *Worker) processJobWithTimeout(job Job, timeout time.Duration) Result {
    resultChan := make(chan Result, 1)
    
    go func() {
        resultChan <- w.processJob(job)
    }()
    
    select {
    case result := <-resultChan:
        return result
    case <-time.After(timeout):
        return Result{
            JobID:  job.ID,
            Output: "",
            Error:  fmt.Errorf("job %d timeout", job.ID),
        }
    }
}

4. 任务重试机制：

为失败的任务实现重试机制。

func (w *Worker) processJobWithRetry(job Job, maxRetries int) Result {
    var lastError error
    
    for i := 0; i < maxRetries; i++ {
        result := w.processJob(job)
        if result.Error == nil {
            return result
        }
        lastError = result.Error
        time.Sleep(time.Second * time.Duration(i+1))
    }
    
    return Result{
        JobID:  job.ID,
        Output: "",
        Error:  fmt.Errorf("job %d failed after %d retries: %v", job.ID, maxRetries, lastError),
    }
}

工作池的变体和扩展：

1. 限流工作池：

限制任务提交的速度，避免过载。

type RateLimitedWorkerPool struct {
    *WorkerPool
    rateLimiter chan struct{}
}
 
func NewRateLimitedWorkerPool(numWorkers int, rateLimit int) *RateLimitedWorkerPool {
    return &RateLimitedWorkerPool{
        WorkerPool:  NewWorkerPool(numWorkers),
        rateLimiter: make(chan struct{}, rateLimit),
    }
}
 
func (rlwp *RateLimitedWorkerPool) AddJob(job Job) {
    rlwp.rateLimiter <- struct{}{}
    go func() {
        rlwp.WorkerPool.AddJob(job)
        <-rlwp.rateLimiter
    }()
}

2. 分组工作池：

将任务分组，每组由专门的Worker处理。

type GroupedWorkerPool struct {
    groups map[string]*WorkerPool
    mu     sync.RWMutex
}
 
func NewGroupedWorkerPool() *GroupedWorkerPool {
    return &GroupedWorkerPool{
        groups: make(map[string]*WorkerPool),
    }
}
 
func (gwp *GroupedWorkerPool) AddGroup(name string, numWorkers int) {
    gwp.mu.Lock()
    defer gwp.mu.Unlock()
    
    if _, exists := gwp.groups[name]; !exists {
        gwp.groups[name] = NewWorkerPool(numWorkers)
        gwp.groups[name].Start()
    }
}
 
func (gwp *GroupedWorkerPool) AddJob(group string, job Job) {
    gwp.mu.RLock()
    defer gwp.mu.RUnlock()
    
    if pool, exists := gwp.groups[group]; exists {
        pool.AddJob(job)
    }
}

实际应用中的注意事项：

1. 任务队列大小：

任务队列的大小需要根据实际情况设置：

  • 队列过小：可能导致任务提交阻塞
  • 队列过大：占用过多内存
// 根据任务大小和Worker数量设置队列大小
queueSize := numWorkers * 10
jobQueue := make(chan Job, queueSize)

2. Worker数量：

Worker数量需要根据任务类型和系统资源设置：

  • IO密集型任务：可以设置较多的Worker
  • CPU密集型任务：Worker数量不宜超过CPU核心数
// IO密集型任务
numWorkers := runtime.NumCPU() * 2
 
// CPU密集型任务
numWorkers := runtime.NumCPU()

3. 优雅关闭：

确保所有任务处理完成后再关闭工作池。

func (wp *WorkerPool) GracefulShutdown() {
    close(wp.JobQueue)
    
    // 等待所有任务处理完成
    for len(wp.JobQueue) > 0 {
        time.Sleep(100 * time.Millisecond)
    }
    
    // 停止所有Worker
    for _, worker := range wp.Workers {
        worker.Stop()
    }
    
    close(wp.Quit)
}

4. 错误处理：

统一处理任务执行过程中的错误。

func (wp *WorkerPool) HandleErrors() {
    go func() {
        for result := range wp.Result {
            if result.Error != nil {
                log.Printf("Job %d failed: %v", result.JobID, result.Error)
                // 可以选择重试或记录错误
            }
        }
    }()
}

代码实现：

package main
 
import (
    "fmt"
    "sync"
    "time"
)
 
type Job struct {
    ID   int
    Data string
}
 
type Result struct {
    JobID  int
    Output string
    Error  error
}
 
type Worker struct {
    ID       int
    JobQueue chan Job
    Result   chan Result
    Quit     chan bool
}
 
func NewWorker(id int, jobQueue chan Job, result chan Result) *Worker {
    return &Worker{
        ID:       id,
        JobQueue: jobQueue,
        Result:   result,
        Quit:     make(chan bool),
    }
}
 
func (w *Worker) Start() {
    go func() {
        for {
            select {
            case job := <-w.JobQueue:
                fmt.Printf("Worker %d: processing job %d\n", w.ID, job.ID)
                time.Sleep(100 * time.Millisecond)
                w.Result <- Result{
                    JobID:  job.ID,
                    Output: fmt.Sprintf("Processed %s", job.Data),
                    Error:  nil,
                }
            case <-w.Quit:
                fmt.Printf("Worker %d: stopping\n", w.ID)
                return
            }
        }
    }()
}
 
func (w *Worker) Stop() {
    go func() {
        w.Quit <- true
    }()
}
 
type WorkerPool struct {
    Workers   []*Worker
    JobQueue  chan Job
    Result    chan Result
    Quit      chan bool
    WorkerWg  sync.WaitGroup
}
 
func NewWorkerPool(numWorkers int) *WorkerPool {
    jobQueue := make(chan Job, 100)
    result := make(chan Result, 100)
    
    workers := make([]*Worker, numWorkers)
    for i := 0; i < numWorkers; i++ {
        workers[i] = NewWorker(i+1, jobQueue, result)
    }
    
    return &WorkerPool{
        Workers:  workers,
        JobQueue: jobQueue,
        Result:   result,
        Quit:     make(chan bool),
    }
}
 
func (wp *WorkerPool) Start() {
    for _, worker := range wp.Workers {
        worker.Start()
        wp.WorkerWg.Add(1)
    }
}
 
func (wp *WorkerPool) Stop() {
    close(wp.JobQueue)
    for _, worker := range wp.Workers {
        worker.Stop()
    }
    close(wp.Quit)
}
 
func (wp *WorkerPool) AddJob(job Job) {
    wp.JobQueue <- job
}
 
func (wp *WorkerPool) GetResults() <-chan Result {
    return wp.Result
}
 
func main() {
    pool := NewWorkerPool(3)
    pool.Start()
    defer pool.Stop()
    
    for i := 1; i <= 10; i++ {
        job := Job{
            ID:   i,
            Data: fmt.Sprintf("Task %d", i),
        }
        pool.AddJob(job)
    }
    
    completed := 0
    for result := range pool.GetResults() {
        if result.Error != nil {
            fmt.Printf("Job %d failed: %v\n", result.JobID, result.Error)
        } else {
            fmt.Printf("Job %d completed: %s\n", result.JobID, result.Output)
        }
        completed++
        if completed >= 10 {
            break
        }
    }
}

常见错误分析：

  • 错误1：工作池没有正确关闭导致资源泄漏
  • 错误2：任务队列大小设置不当导致阻塞
  • 错误3：不理解Worker的生命周期管理