并发基础

并发是 Go 语言的核心特性之一。本章将介绍 Go 并发编程的基本概念。

并发 vs 并行

并发：同时处理多个任务（时间片轮转）
并行：同时执行多个任务（多核处理器）

Go 通过 Goroutines 和 Channels 实现高效的并发编程。

Goroutine

Goroutine 是轻量级的线程，由 Go 运行时管理。

创建 Goroutine

package main

import (
    "fmt"
    "time"
)

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

func main() {
    go sayHello()  // 启动 goroutine
    
    time.Sleep(time.Second)  // 等待 goroutine 执行
    fmt.Println("Main function")
}

多个 Goroutines

func printNumbers() {
    for i := 0; i < 5; i++ {
        fmt.Printf("数字: %d\n", i)
        time.Sleep(100 * time.Millisecond)
    }
}

func printLetters() {
    for i := 'a'; i < 'e'; i++ {
        fmt.Printf("字母: %c\n", i)
        time.Sleep(100 * time.Millisecond)
    }
}

func main() {
    go printNumbers()
    go printLetters()
    
    time.Sleep(time.Second)
}

匿名 Goroutine

func main() {
    go func() {
        fmt.Println("匿名 Goroutine")
    }()
    
    time.Sleep(100 * time.Millisecond)
}

带参数的 Goroutine

func greet(name string) {
    fmt.Printf("Hello, %s!\n", name)
}

func main() {
    go greet("张三")
    go greet("李四")
    
    time.Sleep(100 * time.Millisecond)
}

Channel

Channel 是 Goroutine 之间通信的管道。

创建 Channel

// 无缓冲 channel
ch1 := make(chan int)

// 有缓冲 channel
ch2 := make(chan int, 10)

发送和接收

func main() {
    ch := make(chan string)
    
    go func() {
        ch <- "Hello"  // 发送
    }()
    
    msg := <-ch  // 接收
    fmt.Println(msg)
}

双向通信

func producer(ch chan<- int) {
    for i := 0; i < 5; i++ {
        ch <- i
    }
    close(ch)
}

func consumer(ch <-chan int) {
    for value := range ch {
        fmt.Println("接收:", value)
    }
}

func main() {
    ch := make(chan int)
    
    go producer(ch)
    consumer(ch)
}

缓冲 Channel

func main() {
    ch := make(chan int, 3)
    
    ch <- 1  // 不阻塞
    ch <- 2
    ch <- 3
    
    fmt.Println(<-ch)
    fmt.Println(<-ch)
    fmt.Println(<-ch)
}

关闭 Channel

func main() {
    ch := make(chan int, 2)
    
    ch <- 1
    ch <- 2
    close(ch)
    
    for value := range ch {
        fmt.Println(value)
    }
    
    // 检查 channel 是否关闭
    value, ok := <-ch
    fmt.Println(value, ok)  // 0 false
}

Select 语句

Select 用于处理多个 channel 操作。

func main() {
    ch1 := make(chan string)
    ch2 := make(chan string)
    
    go func() {
        time.Sleep(1 * time.Second)
        ch1 <- "来自 ch1"
    }()
    
    go func() {
        time.Sleep(2 * time.Second)
        ch2 <- "来自 ch2"
    }()
    
    for i := 0; i < 2; i++ {
        select {
        case msg1 := <-ch1:
            fmt.Println(msg1)
        case msg2 := <-ch2:
            fmt.Println(msg2)
        }
    }
}

超时处理

func main() {
    ch := make(chan string)
    
    go func() {
        time.Sleep(2 * time.Second)
        ch <- "结果"
    }()
    
    select {
    case result := <-ch:
        fmt.Println(result)
    case <-time.After(1 * time.Second):
        fmt.Println("超时")
    }
}

非阻塞操作

func main() {
    ch := make(chan int)
    
    select {
    case value := <-ch:
        fmt.Println(value)
    default:
        fmt.Println("没有数据")
    }
}

WaitGroup

WaitGroup 用于等待一组 goroutine 完成。

import "sync"

func worker(id int, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Printf("Worker %d 开始\n", id)
    time.Sleep(time.Second)
    fmt.Printf("Worker %d 完成\n", id)
}

func main() {
    var wg sync.WaitGroup
    
    for i := 1; i <= 3; i++ {
        wg.Add(1)
        go worker(i, &wg)
    }
    
    wg.Wait()
    fmt.Println("所有 worker 完成")
}

Mutex

Mutex 用于保护共享资源。

import "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() {
    var counter Counter
    
    for i := 0; i < 1000; i++ {
        go counter.Increment()
    }
    
    time.Sleep(time.Second)
    fmt.Println("计数:", counter.Value())
}

RWMutex

读写锁，允许多个读或一个写。

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

func (ds *DataStore) Read(key string) (string, bool) {
    ds.mu.RLock()
    defer ds.mu.RUnlock()
    value, ok := ds.data[key]
    return value, ok
}

func (ds *DataStore) Write(key, value string) {
    ds.mu.Lock()
    defer ds.mu.Unlock()
    ds.data[key] = value
}

Once

确保函数只执行一次。

var once sync.Once
var instance *Singleton

func GetInstance() *Singleton {
    once.Do(func() {
        instance = &Singleton{}
    })
    return instance
}

实战示例

示例 1：生产者-消费者

func producer(ch chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    defer close(ch)
    
    for i := 0; i < 10; i++ {
        ch <- i
        fmt.Printf("生产: %d\n", i)
    }
}

func consumer(ch <-chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    
    for value := range ch {
        fmt.Printf("消费: %d\n", value)
        time.Sleep(100 * time.Millisecond)
    }
}

func main() {
    ch := make(chan int, 5)
    var wg sync.WaitGroup
    
    wg.Add(1)
    go producer(ch, &wg)
    
    wg.Add(1)
    go consumer(ch, &wg)
    
    wg.Wait()
    fmt.Println("完成")
}

示例 2：工作池

func worker(id int, jobs <-chan int, results chan<- int) {
    for job := range jobs {
        fmt.Printf("Worker %d 处理任务 %d\n", id, job)
        time.Sleep(100 * time.Millisecond)
        results <- job * 2
    }
}

func main() {
    jobs := make(chan int, 10)
    results := make(chan int, 10)
    
    for i := 1; i <= 3; i++ {
        go worker(i, jobs, results)
    }
    
    for i := 1; i <= 9; i++ {
        jobs <- i
    }
    close(jobs)
    
    for i := 1; i <= 9; i++ {
        <-results
    }
}

最佳实践

不要共享内存，通过通信共享
使用 channel 而非锁
避免 goroutine 泄漏
正确关闭 channel
使用 context 管理生命周期

下一章：Goroutines

并发基础 #

并发 vs 并行 #

Goroutine #

创建 Goroutine #

多个 Goroutines #

匿名 Goroutine #

带参数的 Goroutine #

Channel #

创建 Channel #

发送和接收 #

双向通信 #

缓冲 Channel #

关闭 Channel #

Select 语句 #

超时处理 #

非阻塞操作 #

WaitGroup #

Mutex #

RWMutex #

Once #

实战示例 #

示例 1：生产者-消费者 #

示例 2：工作池 #

最佳实践 #