Go语言并发编程实战指南

本文最后更新于 1 分钟前，文中所描述的信息可能已发生改变。

Go 语言（Golang）因其简单而强大的并发模型而备受推崇。通过 Goroutines 和 Channels，Go 提供了一种优雅且高效的方式来处理并发任务。本文将深入探讨 Go 并发编程的核心概念和最佳实践。

并发与并行

在开始之前，让我们明确两个关键概念：

并发（Concurrency）：指程序的逻辑结构，是程序处理多个任务的能力，即使它们可能不是同时执行的
并行（Parallelism）：指程序的运行状态，是程序同时执行多个任务的能力，通常需要多核处理器支持

正如 Rob Pike 所说："并发是关于处理多个事情，而并行是同时做多个事情。"

Goroutines 基础

Goroutines 是 Go 并发模型的核心，它们是轻量级的"线程"，由 Go 运行时管理。创建一个 Goroutine 非常简单：

package main

import (
    "fmt"
    "time"
)

func sayHello() {
    fmt.Println("Hello, Goroutine!")
}

func main() {
    go sayHello() // 启动一个 Goroutine
    time.Sleep(1 * time.Second) // 给 Goroutine 运行的时间
    fmt.Println("Main function")
}

注意：使用 time.Sleep() 只是为了示例目的。在实际应用中，我们应该使用适当的同步机制。

Goroutines 与线程的区别

Goroutines 与传统线程相比有几个显著优势：

创建成本低：Goroutines 初始只需要 2KB 的栈空间，而线程可能需要 MB 级别
调度开销小：Go 运行时使用协作式调度器，上下文切换成本低
可扩展性强：一个程序可以轻松运行数十万个 Goroutines

Channels 通信

Go 的并发哲学是："通过通信来共享内存，而不是通过共享内存来通信"。Channels 是 Goroutines 之间通信的管道。

基本用法

package main

import "fmt"

func main() {
    // 创建一个无缓冲通道
    ch := make(chan string)

    // 启动 Goroutine 发送数据
    go func() {
        ch <- "Hello from Goroutine!" // 发送数据到通道
    }()

    msg := <-ch // 从通道接收数据
    fmt.Println(msg)
}

缓冲通道

缓冲通道允许发送一定数量的数据而不阻塞：

ch := make(chan int, 3) // 创建带有3个缓冲槽的通道

ch <- 1 // 不会阻塞
ch <- 2 // 不会阻塞
ch <- 3 // 不会阻塞
// ch <- 4 // 会阻塞，因为缓冲区已满

通道方向

通道可以指定方向，限制其操作：

func send(ch chan<- int) { // 只发送通道
    ch <- 42
}

func receive(ch <-chan int) { // 只接收通道
    val := <-ch
    fmt.Println(val)
}

关闭通道

当不再需要发送数据时，可以关闭通道：

close(ch)

通过 range 循环可以迭代通道直到其关闭：

for item := range ch {
    fmt.Println(item)
}

可以使用第二个返回值检查通道是否已关闭：

value, ok := <-ch
if !ok {
    fmt.Println("Channel closed")
}

并发模式

工作池模式（Worker Pool）

工作池是一种常见的并发模式，用于处理大量独立任务：

package main

import (
    "fmt"
    "time"
)

func worker(id int, jobs <-chan int, results chan<- int) {
    for job := range jobs {
        fmt.Printf("Worker %d started job %d\n", id, job)
        time.Sleep(time.Second) // 模拟工作耗时
        fmt.Printf("Worker %d finished job %d\n", id, job)
        results <- job * 2 // 发送结果
    }
}

func main() {
    jobs := make(chan int, 100)
    results := make(chan int, 100)

    // 启动3个worker
    for w := 1; w <= 3; w++ {
        go worker(w, jobs, results)
    }

    // 发送5个任务
    for j := 1; j <= 5; j++ {
        jobs <- j
    }
    close(jobs)

    // 收集所有结果
    for a := 1; a <= 5; a++ {
        <-results
    }
}

扇出扇入模式（Fan-out, Fan-in）

这种模式用于将工作分配给多个Goroutine处理，然后汇总结果：

package main

import (
    "fmt"
    "sync"
)

// 扇出：多个goroutine从同一个通道读取数据
func fanOut(ch <-chan int, n int) []<-chan int {
    outputs := make([]<-chan int, n)
    for i := 0; i < n; i++ {
        outputs[i] = process(ch)
    }
    return outputs
}

// 处理函数
func process(input <-chan int) <-chan int {
    output := make(chan int)
    go func() {
        defer close(output)
        for val := range input {
            output <- val * val // 简单的平方操作
        }
    }()
    return output
}

// 扇入：将多个通道的输出合并到一个通道
func fanIn(inputs []<-chan int) <-chan int {
    output := make(chan int)
    var wg sync.WaitGroup

    collect := func(ch <-chan int) {
        defer wg.Done()
        for val := range ch {
            output <- val
        }
    }

    wg.Add(len(inputs))
    for _, ch := range inputs {
        go collect(ch)
    }

    go func() {
        wg.Wait()
        close(output)
    }()

    return output
}

func main() {
    // 创建输入通道
    input := make(chan int)

    // 启动goroutine发送数据
    go func() {
        for i := 1; i <= 10; i++ {
            input <- i
        }
        close(input)
    }()

    // 扇出为3个处理单元
    processors := fanOut(input, 3)

    // 扇入结果
    results := fanIn(processors)

    // 打印结果
    for result := range results {
        fmt.Println(result)
    }
}

同步机制

sync.WaitGroup

WaitGroup 用于等待一组 Goroutine 完成：

package main

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

func worker(id int, wg *sync.WaitGroup) {
    defer wg.Done() // 完成时减少计数器

    fmt.Printf("Worker %d starting\n", id)
    time.Sleep(time.Second)
    fmt.Printf("Worker %d done\n", id)
}

func main() {
    var wg sync.WaitGroup

    for i := 1; i <= 5; i++ {
        wg.Add(1) // 增加计数器
        go worker(i, &wg)
    }

    wg.Wait() // 等待所有worker完成
    fmt.Println("All workers done")
}

sync.Mutex

互斥锁用于保护共享资源，防止数据竞争：

package main

import (
    "fmt"
    "sync"
)

type Counter struct {
    mu    sync.Mutex
    value int
}

func (c *Counter) Increment() {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.value++
}

func (c *Counter) Value() int {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.value
}

func main() {
    counter := Counter{}
    var wg sync.WaitGroup

    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            counter.Increment()
        }()
    }

    wg.Wait()
    fmt.Println("Counter:", counter.Value())
}

sync.RWMutex

读写锁允许多个读操作同时进行，但写操作是独占的：

type SafeMap struct {
    mu   sync.RWMutex
    data map[string]string
}

func (m *SafeMap) Get(key string) (string, bool) {
    m.mu.RLock() // 读锁
    defer m.mu.RUnlock()
    val, ok := m.data[key]
    return val, ok
}

func (m *SafeMap) Set(key, value string) {
    m.mu.Lock() // 写锁
    defer m.mu.Unlock()
    m.data[key] = value
}

select 语句

select 语句是 Go 并发编程的强大工具，它允许 Goroutine 等待多个通信操作：

package main

import (
    "fmt"
    "time"
)

func main() {
    ch1 := make(chan string)
    ch2 := make(chan string)

    go func() {
        time.Sleep(1 * time.Second)
        ch1 <- "one"
    }()

    go func() {
        time.Sleep(2 * time.Second)
        ch2 <- "two"
    }()

    for i := 0; i < 2; i++ {
        select {
        case msg1 := <-ch1:
            fmt.Println("Received", msg1)
        case msg2 := <-ch2:
            fmt.Println("Received", msg2)
        case <-time.After(3 * time.Second):
            fmt.Println("Timeout")
        }
    }
}

select 特性：

如果多个 case 同时就绪，会随机选择一个执行
可以使用 default 子句实现非阻塞操作
可以与 time.After() 结合实现超时机制

并发安全与陷阱

数据竞争

数据竞争是并发编程中常见的问题，发生在多个 Goroutine 同时访问共享数据且至少有一个是写操作时：

package main

import (
    "fmt"
    "sync"
)

func main() {
    counter := 0
    var wg sync.WaitGroup

    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            counter++ // 数据竞争！
        }()
    }

    wg.Wait()
    fmt.Println("Counter:", counter) // 结果可能小于1000
}

使用 Go 的 race detector 检测数据竞争：

bash

go run -race main.go

Goroutine 泄漏

如果 Goroutine 在完成其工作前无法退出，就会发生 Goroutine 泄漏：

func leak() {
    ch := make(chan int)
    go func() {
        val := <-ch // 永远不会收到数据
        fmt.Println(val)
    }()
    // ch 通道无数据写入，Goroutine 永远阻塞
}

避免泄漏的方法：

使用 context 包管理 Goroutine 生命周期
确保所有情况下 Goroutine 都能退出
设置适当的超时机制

实际应用示例

并发网络爬虫

package main

import (
    "fmt"
    "net/http"
    "sync"
    "time"
)

// 抓取URL并返回状态码
func fetch(url string) (int, error) {
    resp, err := http.Get(url)
    if err != nil {
        return 0, err
    }
    defer resp.Body.Close()
    return resp.StatusCode, nil
}

func main() {
    urls := []string{
        "https://www.google.com",
        "https://www.github.com",
        "https://www.stackoverflow.com",
        "https://www.medium.com",
        "https://www.golang.org",
    }

    var wg sync.WaitGroup
    results := make(map[string]int)
    var mu sync.Mutex

    for _, url := range urls {
        wg.Add(1)
        go func(url string) {
            defer wg.Done()

            start := time.Now()
            status, err := fetch(url)
            elapsed := time.Since(start)

            mu.Lock()
            defer mu.Unlock()

            if err != nil {
                fmt.Printf("Error fetching %s: %v\n", url, err)
                results[url] = -1
            } else {
                fmt.Printf("Fetched %s in %v with status %d\n", url, elapsed, status)
                results[url] = status
            }
        }(url)
    }

    wg.Wait()

    fmt.Println("\nResults:")
    for url, status := range results {
        fmt.Printf("%s: %d\n", url, status)
    }
}

最佳实践

合理使用 Goroutines：不要创建过多 Goroutine，尤其是在资源受限的环境
使用 channels 而非锁：优先选择通过通信（channels）共享数据，而非共享内存（锁）
使用 context 管理生命周期：控制 Goroutine 的取消和超时
避免数据竞争：适当使用互斥锁或通道
定期运行 race detector：用 -race 标志构建和测试代码
处理 panics：在 Goroutine 中使用 recover 处理 panic
监控 Goroutines：了解运行中的 Goroutine 数量，可用 runtime 包

小结

Go 并发编程模型提供了简单而强大的工具，使得开发高性能、并发的应用程序变得更加容易。通过 Goroutines、Channels 和同步原语，Go 解决了传统并发编程中的许多痛点。

了解这些概念和模式后，你可以更有效地利用并发来提高程序性能，同时避免常见的并发陷阱。记住，并发是一种强大的工具，但需要谨慎使用，合理设计并发模型对于创建稳定、高效的程序至关重要。

并发与并行 ​

Goroutines 基础 ​

Goroutines 与线程的区别 ​

Channels 通信 ​

基本用法 ​

缓冲通道 ​

通道方向 ​

关闭通道 ​

并发模式 ​

工作池模式（Worker Pool） ​

扇出扇入模式（Fan-out, Fan-in） ​

同步机制 ​

sync.WaitGroup ​

sync.Mutex ​

sync.RWMutex ​

select 语句 ​

并发安全与陷阱 ​

数据竞争 ​

Goroutine 泄漏 ​

实际应用示例 ​

并发网络爬虫 ​

最佳实践 ​

小结 ​