Go并发模式实战

在前几期内容的基础上，本篇文章将深入探讨Go语言最强大的特性——并发编程的各种实战模式，带你掌握Go高并发编程的精髓。

一、基础并发模式

1. Worker Pool模式

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

    // 启动3个worker
    for w := 1; w <= 3; w++ {
        go func(id int) {
            for job := range jobs {
                fmt.Printf("worker %d processing job %d\n", id, job)
                time.Sleep(time.Second) // 模拟耗时任务
                results <- job * 2
            }
        }(w)
    }

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

    // 收集结果
    for r := 1; r <= 5; r++ {
        fmt.Println("result:", <-results)
    }
}

2. Pub/Sub模式

type PubSub struct {
    mu     sync.RWMutex
    subs   map[string][]chan string
}

func (ps *PubSub) Subscribe(topic string) chan string {
    ps.mu.Lock()
    defer ps.mu.Unlock()

    ch := make(chan string, 1)
    ps.subs[topic] = append(ps.subs[topic], ch)
    return ch
}

func (ps *PubSub) Publish(topic string, msg string) {
    ps.mu.RLock()
    defer ps.mu.RUnlock()

    for _, ch := range ps.subs[topic] {
        go func(ch chan string) {
            ch <- msg
        }(ch)
    }
}

二、sync包深度用法

1. sync.Cond条件变量

func condExample() {
    var mu sync.Mutex
    cond := sync.NewCond(&mu)
    var ready bool

    // 等待goroutine
    go func() {
        time.Sleep(2 * time.Second)

        mu.Lock()
        ready = true
        cond.Broadcast() // 唤醒所有等待者
        mu.Unlock()
    }()

    mu.Lock()
    for !ready {
        cond.Wait() // 等待条件满足
    }
    fmt.Println("条件满足!")
    mu.Unlock()
}

2. sync.Once用法

var (
    instance *Singleton
    once     sync.Once
)

type Singleton struct {
    Value int
}

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

三、Context完美实践

1. 超时控制

func ctxTimeout() {
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()

    select {
    case <-time.After(3 * time.Second):
        fmt.Println("操作完成")
    case <-ctx.Done():
        fmt.Println("操作超时:", ctx.Err())
    }
}

2. 级联取消

func hierarchicalCancel() {
    rootCtx := context.Background()
    ctx1, cancel1 := context.WithCancel(rootCtx)
    ctx2, _ := context.WithCancel(ctx1)

    go func() {
        time.Sleep(1 * time.Second)
        cancel1() // 取消ctx1也会取消ctx2
    }()

    select {
    case <-ctx2.Done():
        fmt.Println("ctx2被取消:", ctx2.Err())
    }
}

四、原子操作

1. atomic基本用法

func atomicCounter() {
    var counter int64
    var wg sync.WaitGroup

    for i := 0; i < 100; i++ {
        wg.Add(1)
        go func() {
            atomic.AddInt64(&counter, 1)
            wg.Done()
        }()
    }

    wg.Wait()
    fmt.Println("最终计数:", atomic.LoadInt64(&counter))
}

2. 自旋锁实现

type SpinLock struct {
    state *int32
}

func (sl *SpinLock) Lock() {
    for !atomic.CompareAndSwapInt32(sl.state, 0, 1) {
        runtime.Gosched() // 释放CPU时间片
    }
}

func (sl *SpinLock) Unlock() {
    atomic.StoreInt32(sl.state, 0)
}

五、Select高级用法

1. 多通道优先级

func prioritySelect() {
    ch1 := make(chan string, 1)
    ch2 := make(chan string, 1)

    ch2 <- "高优先级数据"

    select {
    case msg := <-ch1:
        fmt.Println("收到一般消息:", msg)
    case msg := <-ch2:
        fmt.Println("收到高优先级消息:", msg)
    default:
        fmt.Println("没有消息")
    }
}

2. 超时控制模板

func selectWithTimeout() {
    result := make(chan string)

    go func() {
        time.Sleep(3 * time.Second)
        result <- "操作结果"
    }()

    select {
    case res := <-result:
        fmt.Println(res)
    case <-time.After(2 * time.Second):
        fmt.Println("操作超时")
    }
}

六、并发安全数据结构

1. sync.Map高级用法

func syncMapDemo() {
    var m sync.Map

    // 存储
    m.Store("key1", "value1")
    m.Store("key2", "value2")

    // 加载
    if val, ok := m.Load("key1"); ok {
        fmt.Println("key1:", val)
    }

    // 遍历
    m.Range(func(key, value interface{}) bool {
        fmt.Println(key, "->", value)
        return true
    })
}

2. ConcurrentSlice实现

type ConcurrentSlice struct {
    sync.RWMutex
    items []interface{}
}

func (cs *ConcurrentSlice) Append(item interface{}) {
    cs.Lock()
    defer cs.Unlock()

    cs.items = append(cs.items, item)
}

func (cs *ConcurrentSlice) Len() int {
    cs.RLock()
    defer cs.RUnlock()

    return len(cs.items)
}

预告：Go标准库深度探索

在掌握了强大的并发模式之后，下一期我们将深度探索Go丰富的标准库：

《Go标准库深度探索》 内容预告：

• net/http包详解：构建高性能HTTP服务与客户端
• encoding秘籍：JSON/XML/ProtoBuf全解析
• os/io实战：文件系统与IO操作最佳实践
• time高级用法：定时器与时区处理技巧
• regexp深入：正则表达式高级匹配技巧
• template引擎：文本与HTML模板全攻略

通过这些标准库的深入学习，你的Go开发能力将提升到专业工程师水平！