Go Goroutine、GMP 调度、锁与死锁

package main

import (
    "fmt"
    "runtime"
    "time"
)

func main() {
    // 步骤1：设置 P 的数量为 2
    runtime.GOMAXPROCS(2)

    // 步骤2：启动一个会阻塞的 goroutine（模拟系统调用）
    go func() {
        // time.Sleep 底层会调用系统调用，M 会阻塞
        time.Sleep(2 * time.Second)
        fmt.Println("阻塞 goroutine 完成")
    }()

    // 步骤3：启动一个普通计算的 goroutine
    go func() {
        sum := 0
        for i := 0; i < 1000000; i++ {
            sum += i
        }
        fmt.Println("计算 goroutine 完成, sum =", sum)
    }()

    // 步骤4：等待所有 goroutine 完成
    time.Sleep(3 * time.Second)
    fmt.Println("当前 goroutine 数量:", runtime.NumGoroutine())
}

// 步骤1：获取当前 GOMAXPROCS（传 0 表示不修改，只返回当前值）
fmt.Println("默认 P 数量:", runtime.GOMAXPROCS(0))

// 步骤2：设置为 4
runtime.GOMAXPROCS(4)

// 步骤3：再次获取
fmt.Println("设置后 P 数量:", runtime.GOMAXPROCS(0))

package main

import (
    "fmt"
    "sync"
)

// 步骤1：递归函数，每层占用约 1KB 栈空间
func deepRecursion(n int) int {
    if n <= 0 {
        return 0
    }
    var buf [1024]byte // 每层分配 1KB 栈空间
    buf[0] = byte(n % 256)
    return int(buf[0]) + deepRecursion(n-1)
}

func main() {
    var wg sync.WaitGroup
    wg.Add(1)

    // 步骤2：在 goroutine 中深度递归，触发栈扩容
    go func() {
        defer wg.Done()
        result := deepRecursion(10000)
        fmt.Println("递归结果:", result)
        // 步骤3：goroutine 栈从 2KB 开始，按需扩容到足够大小
        // 不会像 C 语言那样 stack overflow
    }()

    wg.Wait()
    fmt.Println("goroutine 初始栈: 2KB, 最大: 1GB")
}

package main

import (
    "io"
    "net/http"
    "sync"
)

// 场景1：并发 IO（同时请求多个 API）
func fetchConcurrent(urls []string) []string {
    results := make([]string, len(urls))
    var wg sync.WaitGroup

    for i, url := range urls {
        // 步骤1：Add 必须在 goroutine 外部调用
        wg.Add(1)
        go func(idx int, u string) {
            defer wg.Done() // 步骤2：goroutine 结束时通知
            resp, _ := http.Get(u)
            body, _ := io.ReadAll(resp.Body)
            results[idx] = string(body)
        }(i, url)
    }

    wg.Wait() // 步骤3：等待所有请求完成
    return results
}

// 场景2：并发计算（并行求和）
func parallelSum(data []int, numWorkers int) int {
    chunkSize := len(data) / numWorkers
    results := make(chan int, numWorkers)

    for i := 0; i < numWorkers; i++ {
        // 步骤1：每个 worker 处理一个数据分片
        go func(start int) {
            sum := 0
            end := start + chunkSize
            if end > len(data) {
                end = len(data)
            }
            for j := start; j < end; j++ {
                sum += data[j]
            }
            // 步骤2：结果通过 channel 传回
            results <- sum
        }(i * chunkSize)
    }

    // 步骤3：汇总所有 worker 的结果
    total := 0
    for i := 0; i < numWorkers; i++ {
        total += <-results
    }
    return total
}

package main

import (
    "fmt"
    "time"
)

func worker(stop <-chan struct{}) {
    ticker := time.NewTicker(500 * time.Millisecond)
    defer ticker.Stop()

    for {
        select {
        // 步骤1：监听退出信号
        case <-stop:
            fmt.Println("worker 收到退出信号，正在停止...")
            return
        // 步骤2：正常工作逻辑
        case t := <-ticker.C:
            fmt.Println("worker 工作中:", t.Format("15:04:05"))
        }
    }
}

func main() {
    // 步骤3：创建退出信号 channel
    stop := make(chan struct{})

    go worker(stop)

    // 步骤4：运行 3 秒后发送退出信号
    time.Sleep(3 * time.Second)
    close(stop) // close 后所有接收者都能收到零值
    time.Sleep(500 * time.Millisecond)
    fmt.Println("主程序退出")
}

package main

import (
    "context"
    "fmt"
    "time"
)

func worker(ctx context.Context, id int) {
    for {
        select {
        // 步骤1：监听 context 取消
        case <-ctx.Done():
            fmt.Printf("worker %d: 收到取消信号, 原因: %v\n", id, ctx.Err())
            return
        default:
            // 步骤2：模拟工作
            fmt.Printf("worker %d: 工作中...\n", id)
            time.Sleep(500 * time.Millisecond)
        }
    }
}

func main() {
    // 步骤3：创建可取消的 context
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel() // 确保最终会取消

    // 步骤4：启动多个 worker
    for i := 1; i <= 3; i++ {
        go worker(ctx, i)
    }

    // 步骤5：运行 2 秒后取消所有 worker
    time.Sleep(2 * time.Second)
    cancel()
    time.Sleep(500 * time.Millisecond)
    fmt.Println("主程序退出")
}

package main

import (
    "sync"
    "time"
)

// 方式1：channel 传回（推荐）
func computeAsync(n int) <-chan int {
    ch := make(chan int, 1)
    go func() {
        // 步骤1：计算结果
        result := n * n
        // 步骤2：通过 channel 返回
        ch <- result
    }()
    return ch
}
// 使用：ch := computeAsync(42); result := <-ch

// 方式2：WaitGroup + 闭包
func computeWithWG(n int) int {
    var wg sync.WaitGroup
    var result int

    wg.Add(1)
    go func() {
        defer wg.Done()
        result = n * n
    }()

    wg.Wait()
    return result
}

// 方式3：Future 模式
type Future struct {
    result chan int
}

func NewFuture(n int) *Future {
    f := &Future{result: make(chan int, 1)}
    go func() {
        // 步骤1：异步执行计算
        time.Sleep(100 * time.Millisecond)
        f.result <- n * n
    }()
    return f
}

func (f *Future) Get() int {
    // 步骤2：按需获取结果（阻塞直到完成）
    return <-f.result
}

// 原因1：channel 发送无接收者
func leakSend() {
    ch := make(chan int) // 无缓冲 channel
    go func() {
        // 步骤1：永远阻塞，因为 main 没有接收
        ch <- 42
        fmt.Println("这行永远不会执行")
    }()
    // 步骤2：函数返回，ch 无人引用
    // 但 goroutine 还在等接收者
}

// 原因2：channel 接收无发送者
func leakReceive() {
    ch := make(chan int)
    go func() {
        // 步骤1：永远阻塞，等待数据
        val := <-ch
        fmt.Println("收到:", val)
    }()
    // 步骤2：函数返回，没人往 ch 发数据
}

// 原因3：context 未取消
func leakContext() {
    ch := make(chan struct{})
    go func() {
        select {
        case <-ch:
            // 步骤1：等待信号，但没人关闭 ch
        }
    }()
    // 步骤2：忘记关闭 ch 或没有 context 超时
}

package main

import (
    "context"
    "fmt"
    "time"
)

// 修复版：使用 context 超时
func safeWorker(ctx context.Context) {
    ch := make(chan int, 1)

    go func() {
        // 步骤1：模拟耗时操作
        time.Sleep(2 * time.Second)
        ch <- 42
    }()

    select {
    case val := <-ch:
        // 步骤2：正常收到结果
        fmt.Println("收到结果:", val)
    case <-ctx.Done():
        // 步骤3：超时或取消，goroutine 可以退出
        fmt.Println("超时退出:", ctx.Err())
    }
}

func main() {
    // 步骤4：设置 1 秒超时
    ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
    defer cancel()

    safeWorker(ctx)
    time.Sleep(500 * time.Millisecond)
}

package main

import (
    "fmt"
    "os"
    "runtime"
    "runtime/pprof"
    "time"
)

func leakyFunc() {
    ch := make(chan int)
    go func() {
        <-ch // 永久阻塞
    }()
}

func main() {
    // 步骤1：记录初始 goroutine 数量
    fmt.Println("初始 goroutine 数:", runtime.NumGoroutine())

    // 步骤2：反复调用泄露函数
    for i := 0; i < 100; i++ {
        leakyFunc()
    }

    time.Sleep(time.Second)
    fmt.Println("泄露后 goroutine 数:", runtime.NumGoroutine())

    // 步骤3：导出 goroutine profile
    f, _ := os.Create("/tmp/goroutine.prof")
    defer f.Close()
    pprof.Lookup("goroutine").WriteTo(f, 2)

    // 步骤4：用 go tool pprof /tmp/goroutine.prof 分析
    // 在 pprof 交互界面输入: top, list leakyFunc
}

package main

import (
    "fmt"
    "sync"
)

func main() {
    var mu sync.Mutex
    var counter int

    // 步骤1：模拟正常模式下的高并发竞争
    var wg sync.WaitGroup
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            // 步骤2：加锁-操作-解锁
            mu.Lock()
            counter++
            mu.Unlock()
        }()
    }

    wg.Wait()
    fmt.Println("最终计数:", counter) // 1000
}

package main

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

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

func NewSafeCache() *SafeCache {
    return &SafeCache{
        data: make(map[string]string),
    }
}

// 步骤1：读操作用 RLock（多读并发）
func (c *SafeCache) Get(key string) (string, bool) {
    c.mu.RLock()
    defer c.mu.RUnlock()
    val, ok := c.data[key]
    return val, ok
}

// 步骤2：写操作用 Lock（独占）
func (c *SafeCache) Set(key, val string) {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.data[key] = val
}

func main() {
    cache := NewSafeCache()

    // 步骤3：并发读
    for i := 0; i < 5; i++ {
        go func(id int) {
            for {
                if val, ok := cache.Get("name"); ok {
                    fmt.Printf("reader %d: %s\n", id, val)
                }
                time.Sleep(100 * time.Millisecond)
            }
        }(i)
    }

    // 步骤4：并发写
    for i := 0; i < 3; i++ {
        go func(id int) {
            for {
                cache.Set("name", fmt.Sprintf("writer-%d", id))
                time.Sleep(500 * time.Millisecond)
            }
        }(i)
    }

    time.Sleep(3 * time.Second)
}

// 方式1：map + RWMutex（适合读多写少）
type SafeMap struct {
    mu   sync.RWMutex
    data map[string]interface{}
}

// 方式2：sync.Map（适合 key 稳定、读远多于写）
var m sync.Map
m.Store("key", "value")      // 存储
val, ok := m.Load("key")     // 读取
m.Delete("key")              // 删除

package main

import (
    "fmt"
    "sync"
    "sync/atomic"
)

func main() {
    var counter int64

    var wg sync.WaitGroup
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            // 步骤1：Add — 原子加法
            atomic.AddInt64(&counter, 1)
        }()
    }
    wg.Wait()

    // 步骤2：Load — 原子读取
    fmt.Println("Add 结果:", atomic.LoadInt64(&counter)) // 1000

    // 步骤3：Store — 原子写入
    atomic.StoreInt64(&counter, 42)
    fmt.Println("Store 后:", atomic.LoadInt64(&counter))

    // 步骤4：Swap — 原子交换，返回旧值
    old := atomic.SwapInt64(&counter, 100)
    fmt.Println("Swap 旧值:", old, "新值:", atomic.LoadInt64(&counter))

    // 步骤5：CompareAndSwap — 原子比较并交换
    success := atomic.CompareAndSwapInt64(&counter, 100, 200)
    fmt.Println("CAS 第一次 (100->200):", success, "值:", atomic.LoadInt64(&counter))

    // 步骤6：CAS 失败（期望值不匹配）
    success = atomic.CompareAndSwapInt64(&counter, 100, 300)
    fmt.Println("CAS 第二次 (100->300):", success, "值:", atomic.LoadInt64(&counter))
}

package main

import (
    "fmt"
    "sync"
    "sync/atomic"
)

// 用 CAS 实现原子加法（模拟 atomic.Add 的底层逻辑）
func casAdd(addr *int64, delta int64) {
    for {
        // 步骤1：读取当前值
        old := atomic.LoadInt64(addr)
        // 步骤2：计算新值
        newVal := old + delta
        // 步骤3：尝试 CAS，成功则返回
        if atomic.CompareAndSwapInt64(addr, old, newVal) {
            return
        }
        // 步骤4：CAS 失败，说明有其他 goroutine 抢先修改，重试
    }
}

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

    for i := 0; i < 10000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            casAdd(&counter, 1)
        }()
    }

    wg.Wait()
    fmt.Println("CAS 计数器结果:", counter) // 10000
}

package main

import (
    "fmt"
    "sync"
    "sync/atomic"
)

type Config struct {
    Host string
    Port int
}

func main() {
    var config atomic.Value

    // 步骤1：首次存储配置
    config.Store(&Config{Host: "localhost", Port: 8080})

    var wg sync.WaitGroup

    // 步骤2：并发读取
    for i := 0; i < 5; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            // 步骤3：Load 返回 interface{}，需要类型断言
            c := config.Load().(*Config)
            fmt.Printf("reader %d: %s:%d\n", id, c.Host, c.Port)
        }(i)
    }

    // 步骤4：并发更新
    config.Store(&Config{Host: "0.0.0.0", Port: 9090})

    wg.Wait()
}

// 场景1：并发安全的标志位（Go 1.19+ 使用 atomic.Bool）
type Service struct {
    running atomic.Bool
}

func (s *Service) Start() {
    // 步骤1：CAS 设置为 running
    if !s.running.CompareAndSwap(false, true) {
        return // 已经在运行
    }
    // 步骤2：执行启动逻辑
}

func (s *Service) Stop() {
    s.running.Store(false)
}

func (s *Service) IsRunning() bool {
    return s.running.Load()
}

// 场景2：并发安全计数器
type Counter struct {
    count atomic.Int64
}

func (c *Counter) Inc() int64  { return c.count.Add(1) }
func (c *Counter) Get() int64  { return c.count.Load() }

// 场景3：sync.Once 底层就是 atomic + Mutex（见第七章）

package main

import (
    "fmt"
    "sync"
)

func main() {
    var wg sync.WaitGroup

    // 步骤1：Add 必须在 goroutine 外部调用
    wg.Add(3)

    // 步骤2：启动 3 个 goroutine
    for i := 1; i <= 3; i++ {
        go func(id int) {
            // 步骤3：defer Done 确保一定会执行
            defer wg.Done()
            fmt.Printf("goroutine %d 完成\n", id)
        }(i)
    }

    // 步骤4：Wait 阻塞直到 counter 归零
    wg.Wait()
    fmt.Println("所有 goroutine 完成")
}

package main

import (
    "sync"
)

// 坑1：Add 在 goroutine 内部调用（竞争条件！）
func badExample() {
    var wg sync.WaitGroup
    for i := 0; i < 5; i++ {
        go func() {
            wg.Add(1) // 错误！可能在 Wait 之后才执行
            defer wg.Done()
        }()
    }
    wg.Wait() // 可能提前返回
}

// 正确写法
func goodExample() {
    var wg sync.WaitGroup
    for i := 0; i < 5; i++ {
        wg.Add(1) // 正确！在启动 goroutine 前调用
        go func() {
            defer wg.Done()
        }()
    }
    wg.Wait()
}

// 坑2：Done 调用次数超过 Add 会导致 counter 为负
func negativeExample() {
    var wg sync.WaitGroup
    wg.Add(1)
    wg.Done()
    wg.Done() // panic: sync: negative WaitGroup counter
}

// 坑3：Wait 之后可以复用，但要确保之前的 Wait 已返回
func reuseExample() {
    var wg sync.WaitGroup
    wg.Add(1)
    go func() {
        defer wg.Done()
    }()

    // 步骤1：正确等待第一轮
    wg.Wait()

    // 步骤2：复用，开始第二轮
    wg.Add(1)
    go func() {
        defer wg.Done()
    }()
    wg.Wait()
}

// Add 的简化逻辑（实际实现见 runtime/sema.go）
func (wg *WaitGroup) Add(delta int) {
    // 步骤1：用 atomic 原子更新 counter（高 32 位）
    state := atomic.AddUint64(&wg.state1, uint64(delta)<<32)
    v := int32(state >> 32) // counter
    w := uint32(state)       // waiter

    if v == 0 {
        // 步骤2：counter 归零，释放所有 waiter 的信号量
        for ; w != 0; w-- {
            runtime_Semrelease(&wg.sema, false, 0)
        }
    }
}

// Wait 的简化逻辑
func (wg *WaitGroup) Wait() {
    // 步骤1：原子递增 waiter 计数（低 32 位）
    state := atomic.AddUint64(&wg.state1, 1)
    v := int32(state >> 32) // counter

    if v > 0 {
        // 步骤2：counter > 0，阻塞在信号量上
        runtime_Semacquire(&wg.sema)
    }
}

// Done 的简化逻辑
func (wg *WaitGroup) Done() {
    // 步骤1：counter 减 1
    wg.Add(-1)
}

package main

import (
    "fmt"
    "sync"
)

type Singleton struct {
    name string
}

var (
    instance *Singleton
    once     sync.Once
)

// 步骤1：GetInstance 保证只初始化一次
func GetInstance() *Singleton {
    once.Do(func() {
        // 步骤2：这段代码只会执行一次
        fmt.Println("初始化 Singleton...")
        instance = &Singleton{name: "我是唯一的实例"}
    })
    return instance
}

func main() {
    var wg sync.WaitGroup

    // 步骤3：并发调用 GetInstance
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            s := GetInstance()
            fmt.Printf("goroutine %d: %s\n", id, s.name)
        }(i)
    }

    wg.Wait()
    // 输出中 "初始化 Singleton..." 只出现一次
}

// 简化的 Once 底层逻辑
type Once struct {
    done atomic.Uint32 // 标志位：0=未执行, 1=已执行
    m    sync.Mutex    // 互斥锁
}

func (o *Once) Do(f func()) {
    // 步骤1：快速路径——atomic 检查是否已执行（无锁）
    if o.done.Load() == 0 {
        o.doSlow(f)
    }
}

func (o *Once) doSlow(f func()) {
    o.m.Lock()
    defer o.m.Unlock()
    // 步骤2：双检查——防止多个 goroutine 同时通过第一次检查
    if o.done.Load() == 0 {
        // 步骤3：执行目标函数
        f()
        // 步骤4：标记为已执行
        o.done.Store(1)
    }
}

package main

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

type Queue struct {
    mu    sync.Mutex
    cond  *sync.Cond
    items []int
}

func NewQueue() *Queue {
    q := &Queue{}
    // 步骤1：cond 必须关联一个 Mutex
    q.cond = sync.NewCond(&q.mu)
    return q
}

// 步骤2：消费者——等待数据
func (q *Queue) Consume() int {
    q.mu.Lock()
    defer q.mu.Unlock()

    // 步骤3：队列为空时等待（必须用 for，不能用 if）
    for len(q.items) == 0 {
        // Wait 内部会：释放锁 -> 阻塞 -> 被唤醒 -> 重新获取锁
        q.cond.Wait()
    }

    item := q.items[0]
    q.items = q.items[1:]
    return item
}

// 步骤4：生产者——添加数据并通知
func (q *Queue) Produce(item int) {
    q.mu.Lock()
    defer q.mu.Unlock()

    q.items = append(q.items, item)
    // 步骤5：唤醒一个等待的消费者
    q.cond.Signal()
    // 步骤6：如果要唤醒所有等待者，用 Broadcast()
    // q.cond.Broadcast()
}

func main() {
    q := NewQueue()

    // 步骤7：启动两个消费者
    var wg sync.WaitGroup
    for i := 1; i <= 2; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            val := q.Consume()
            fmt.Printf("消费者 %d 收到: %d\n", id, val)
        }(i)
    }

    // 步骤8：生产者往队列添加数据
    time.Sleep(500 * time.Millisecond)
    q.Produce(42)
    q.Produce(100)

    wg.Wait()
}

package main

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

func main() {
    var lockA, lockB sync.Mutex

    // 步骤1：goroutine 1 先锁 A 再锁 B
    go func() {
        lockA.Lock()
        fmt.Println("goroutine 1: 获得锁 A")
        time.Sleep(100 * time.Millisecond) // 制造时序

        lockB.Lock() // 等待 goroutine 2 释放锁 B
        fmt.Println("goroutine 1: 获得锁 B")
        lockB.Unlock()
        lockA.Unlock()
    }()

    // 步骤2：goroutine 2 先锁 B 再锁 A（顺序相反！）
    go func() {
        lockB.Lock()
        fmt.Println("goroutine 2: 获得锁 B")
        time.Sleep(100 * time.Millisecond)

        lockA.Lock() // 等待 goroutine 1 释放锁 A
        fmt.Println("goroutine 2: 获得锁 A")
        lockA.Unlock()
        lockB.Unlock()
    }()

    // 步骤3：程序死锁，永远无法结束
    // Go runtime 会检测到并 panic: "all goroutines are deadlock!"
    time.Sleep(2 * time.Second)
}

package main

import (
    "sync"
    "time"
)

type Account struct {
    Balance int
}

// 方法1：固定加锁顺序（推荐）
func safeTransfer(from, to *Account, amount int, muA, muB *sync.Mutex) {
    // 步骤1：按地址排序，保证所有 goroutine 的加锁顺序一致
    first, second := muA, muB
    if &muA > &muB { // 用地址作为排序依据
        first, second = muB, muA
    }
    first.Lock()
    defer first.Unlock()
    second.Lock()
    defer second.Unlock()

    // 步骤2：安全操作
    from.Balance -= amount
    to.Balance += amount
}

// 方法2：使用带超时的锁（避免永久等待）
func tryLockWithTimeout(mu *sync.Mutex, timeout time.Duration) bool {
    done := make(chan struct{})
    go func() {
        mu.Lock()
        close(done)
    }()
    select {
    case <-done:
        return true
    case <-time.After(timeout):
        return false // 超时未获取到锁
    }
}

// 方法3：避免嵌套锁（减小锁粒度）
func noNesting(mu *sync.Mutex, data map[string]int) int {
    // 步骤1：只锁必要部分
    mu.Lock()
    val := data["key"]
    mu.Unlock()

    // 步骤2：不持锁的情况下做耗时操作
    result := process(val)

    // 步骤3：需要写入时再锁
    mu.Lock()
    data["result"] = result
    mu.Unlock()

    return result
}

func process(val int) int {
    return val * 2
}

package main

import (
    "fmt"
    "os"
    "runtime"
    "runtime/pprof"
    "time"
)

func detectDeadlock() {
    // 步骤1：监控 goroutine 数量
    ticker := time.NewTicker(time.Second)
    go func() {
        for {
            <-ticker.C
            fmt.Println("当前 goroutine 数:", runtime.NumGoroutine())
        }
    }()

    // 步骤2：导出 goroutine profile 供分析
    go func() {
        time.Sleep(5 * time.Second)
        f, _ := os.Create("/tmp/goroutine.prof")
        pprof.Lookup("goroutine").WriteTo(f, 2)
        f.Close()
        fmt.Println("profile 已导出到 /tmp/goroutine.prof")
    }()
}

package main

import "fmt"

func main() {
    // 步骤1：nil map 写入会 panic
    var m map[string]int // 声明但未初始化，m == nil
    // m["key"] = 1 // panic: assignment to entry in nil map

    // 步骤2：正确做法——先 make 初始化
    m2 := make(map[string]int)
    m2["key"] = 1 // 正常
    fmt.Println("初始化后写入:", m2)

    // 步骤3：nil map 的读取是安全的（返回零值）
    var m3 map[string]int // nil map
    val := m3["key"]      // 不 panic，返回 0
    fmt.Println("nil map 读取:", val)
}

package main

import "fmt"

func main() {
    // 步骤1：nil slice 可以 append（会分配底层数组）
    var s []int // s == nil
    s = append(s, 1, 2, 3)
    fmt.Println("nil slice append:", s) // [1 2 3]

    // 步骤2：nil slice 的其他操作
    var s2 []int
    fmt.Println("len:", len(s2))    // 0
    fmt.Println("cap:", cap(s2))    // 0
    fmt.Println("nil?:", s2 == nil) // true

    // 步骤3：nil slice 遍历安全
    for _, v := range s2 {
        fmt.Println(v) // 不会执行
    }
}

package main

import (
    "fmt"
    "time"
)

func main() {
    var ch chan int // ch == nil

    // 步骤1：nil channel 发送永久阻塞
    go func() {
        ch <- 42 // 永远阻塞在这里
    }()

    // 步骤2：nil channel 接收永久阻塞
    go func() {
        <-ch // 永远阻塞在这里
    }()

    // 步骤3：nil channel 在 select 中的妙用
    // 利用 nil channel 在 select 中"禁用"某个分支
    ch1 := make(chan int, 1)
    ch1 <- 1

    var ch2 chan int = nil // 故意设为 nil

    select {
    case val := <-ch1:
        fmt.Println("从 ch1 收到:", val)
    case val := <-ch2:
        // 步骤4：ch2 为 nil，这个分支永远不会被选中
        fmt.Println("从 ch2 收到:", val)
    }

    time.Sleep(100 * time.Millisecond)
}

// 最佳实践：统一用 make 初始化
func initCollections() {
    // 步骤1：map 用 make 初始化
    m := make(map[string]int)

    // 步骤2：slice 声明 nil 可以，需要时 append
    var s []int
    s = append(s, 1)

    // 步骤3：channel 用 make 初始化
    ch := make(chan int, 10)

    _ = m
    _ = s
    _ = ch
}

// 写法 A：共享变量 + Mutex（共享内存派）
type CounterMutex struct {
    mu sync.Mutex
    n  int
}

func (c *CounterMutex) Inc() {
    c.mu.Lock()   // 步骤1：抢"那支唯一的笔"
    c.n++         // 步骤2：改公共白板
    c.mu.Unlock() // 步骤3：把笔放回去（漏了这步，全程序卡死）
}

func (c *CounterMutex) Get() int {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.n    // 步骤4：读也必须加锁，否则是数据竞争
}

// 写法 B：CSP —— 数据只归一个 goroutine 所有，别人通过 channel 请求
type CounterCSP struct {
    incCh  chan struct{}  // 写请求通道
    readCh chan chan int  // 读请求通道：把"回执信封"一起递进去
}

func NewCounterCSP(ctx context.Context) *CounterCSP {
    c := &CounterCSP{
        incCh:  make(chan struct{}, 128), // 带缓冲，削峰
        readCh: make(chan chan int),
    }

    // 步骤1：唯一的"数据所有者" goroutine
    go func() {
        n := 0 // 步骤2：n 是局部变量，除它之外没人能碰 —— 天然无竞争
        for {
            select {
            case <-c.incCh:
                n++ // 步骤3：串行处理，不需要任何锁
            case reply := <-c.readCh:
                reply <- n // 步骤4：读也走消息，把快照回传给请求方
            case <-ctx.Done():
                return // 步骤5：明确的退出路径，避免所有者 goroutine 泄露
            }
        }
    }()
    return c
}

func (c *CounterCSP) Inc() { c.incCh <- struct{}{} }

func (c *CounterCSP) Get() int {
    reply := make(chan int) // 步骤6：每次请求自带回执 channel，结果不会串号
    c.readCh <- reply
    return <-reply
}

package main

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

// Job 一条待处理消息
type Job struct {
    ID      int
    Payload string
}

// Result 处理结果（成功/失败都往回报，便于统计）
type Result struct {
    JobID  int
    Output string
    Err    error
}

type Pool struct {
    jobs    chan Job
    results chan Result
    wg      sync.WaitGroup
    workers int
}

// 步骤1：queueSize 决定缓冲深度 —— 有界队列才有背压，别用无界 slice 当队列
func NewPool(workers, queueSize int) *Pool {
    return &Pool{
        jobs:    make(chan Job, queueSize),
        results: make(chan Result, queueSize),
        workers: workers,
    }
}

// 步骤2：Start 一次性启动固定数量的 worker，goroutine 总数从此可控
func (p *Pool) Start(ctx context.Context) {
    for i := 1; i <= p.workers; i++ {
        p.wg.Add(1) // Add 必须在 go 之前（见第 6 章）
        go p.worker(ctx, i)
    }
}

func (p *Pool) worker(ctx context.Context, id int) {
    defer p.wg.Done()
    for {
        select {
        case job, ok := <-p.jobs:
            // 步骤3：ok == false 说明 jobs 已被生产者 close 且取空，正常退场
            if !ok {
                return
            }
            p.results <- p.handle(id, job)
        case <-ctx.Done():
            // 步骤4：外部超时/服务下线，立刻停手，不再取新活
            return
        }
    }
}

// 步骤5：单条消息的处理逻辑。必须用 defer recover 兜住业务 panic，
// 否则一条脏数据引发的 panic 会带崩整个进程
func (p *Pool) handle(workerID int, job Job) (r Result) {
    defer func() {
        if e := recover(); e != nil {
            r = Result{JobID: job.ID, Err: fmt.Errorf("panic: %v", e)}
        }
    }()

    time.Sleep(50 * time.Millisecond) // 模拟业务耗时
    return Result{
        JobID:  job.ID,
        Output: fmt.Sprintf("worker-%d 处理了 %s", workerID, job.Payload),
    }
}

// 步骤6：投递消息。队列满时这里会阻塞，压力自然回传给上游 —— 这是特性不是 bug
func (p *Pool) Submit(job Job) { p.jobs <- job }

// 步骤7：优雅关闭的固定套路：先 close(jobs) → 等 worker 全退 → 再 close(results)
// 顺序反了会 "send on closed channel" panic
func (p *Pool) Stop() {
    close(p.jobs)
    p.wg.Wait()
    close(p.results)
}

func (p *Pool) Results() <-chan Result { return p.results }

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

    pool := NewPool(3, 16) // 3 个 worker，队列容量 16
    pool.Start(ctx)

    // 步骤8：结果消费必须与投递并发进行。
    // 如果先投完再来收，results 写满 16 条后 worker 会全部卡在发送上（死锁）
    var collect sync.WaitGroup
    collect.Add(1)
    go func() {
        defer collect.Done()
        for r := range pool.Results() { // close(results) 后循环自动结束
            if r.Err != nil {
                fmt.Printf("job %d 失败: %v\n", r.JobID, r.Err)
                continue
            }
            fmt.Println(r.Output)
        }
    }()

    // 步骤9：生产者投递消息
    for i := 1; i <= 10; i++ {
        pool.Submit(Job{ID: i, Payload: fmt.Sprintf("msg-%d", i)})
    }

    pool.Stop()      // 步骤10：投递方负责关闭 jobs
    collect.Wait()   // 步骤11：等结果收完再退出 main
    fmt.Println("全部消息处理完毕")
}

// 限制最大并发为 limit，跑完一批就结束
func runWithLimit(tasks []func(), limit int) {
    sem := make(chan struct{}, limit) // 步骤1：容量 = 并发上限，即"令牌总数"
    var wg sync.WaitGroup

    for _, task := range tasks {
        wg.Add(1)
        sem <- struct{}{} // 步骤2：拿一个令牌，令牌用光就在这里阻塞排队

        go func(f func()) {
            defer wg.Done()
            defer func() { <-sem }() // 步骤3：无论成功失败都要归还令牌
            defer func() {
                if e := recover(); e != nil { /* 步骤4：兜住 panic */ }
            }()
            f()
        }(task)
    }
    wg.Wait() // 步骤5：等这一批全部结束
}

// 分片锁：把一把大锁拆成 N 把小锁，按 key 哈希打散，冲突概率降到 1/N
const shardCount = 32

type ShardedMap struct {
    shards [shardCount]struct {
        mu   sync.RWMutex
        data map[string]int
    }
}

func NewShardedMap() *ShardedMap {
    m := &ShardedMap{}
    for i := range m.shards {
        m.shards[i].data = make(map[string]int) // 步骤1：每片独立初始化
    }
    return m
}

// 步骤2：用 key 的哈希决定落在哪一片，不同片的读写完全互不阻塞
func (m *ShardedMap) shard(key string) int {
    h := fnv.New32a()
    h.Write([]byte(key))
    return int(h.Sum32()) % shardCount
}

func (m *ShardedMap) Set(key string, val int) {
    s := &m.shards[m.shard(key)]
    s.mu.Lock()   // 步骤3：只锁这一片
    defer s.mu.Unlock()
    s.data[key] = val
}

func (m *ShardedMap) Get(key string) (int, bool) {
    s := &m.shards[m.shard(key)]
    s.mu.RLock()  // 步骤4：读锁，同片内多读并发
    defer s.mu.RUnlock()
    v, ok := s.data[key]
    return v, ok
}

-- 方案 A：乐观锁（version 字段做 CAS）。冲突时受影响行数为 0，业务层重试
UPDATE stock
SET    count = count - 1, version = version + 1
WHERE  id = 1001 AND version = 7;

-- 方案 A+：更省事的写法 —— 把业务校验直接写进 WHERE，由数据库保证单条 UPDATE 的原子性
-- 这一条就能防超卖，不需要任何锁
UPDATE stock SET count = count - 1 WHERE id = 1001 AND count >= 1;

-- 方案 B：悲观锁。事务内先锁住行，适合"临界区里要做多次读写"的场景
BEGIN;
SELECT count FROM stock WHERE id = 1001 FOR UPDATE;  -- 锁行，其他事务在此阻塞
UPDATE stock SET count = count - 1 WHERE id = 1001;
COMMIT;                                               -- 提交即释放锁

// 乐观锁重试：CAS 失败就重读重试，注意一定要有重试上限
func deductStock(ctx context.Context, db *sql.DB, id int) error {
    const maxRetry = 3
    for i := 0; i < maxRetry; i++ {
        // 步骤1：读出当前值和版本号
        var count, version int
        err := db.QueryRowContext(ctx,
            "SELECT count, version FROM stock WHERE id = ?", id).Scan(&count, &version)
        if err != nil {
            return err
        }
        if count <= 0 {
            return errors.New("库存不足")
        }

        // 步骤2：带版本号条件更新 —— 这就是数据库层面的 CompareAndSwap
        res, err := db.ExecContext(ctx,
            "UPDATE stock SET count = count - 1, version = version + 1 WHERE id = ? AND version = ?",
            id, version)
        if err != nil {
            return err
        }

        // 步骤3：影响行数为 0 说明版本变了（被别人抢先改过），退回重试
        if n, _ := res.RowsAffected(); n > 0 {
            return nil
        }
    }
    return errors.New("并发冲突，重试次数已用尽")
}

package lock

import (
    "context"
    "errors"
    "time"

    "github.com/google/uuid"
    "github.com/redis/go-redis/v9"
)

// 步骤1：解锁脚本 —— "比对 value 再删除"必须是一个原子动作。
// 分成 GET + DEL 两条命令的话，中间锁过期被别人拿到，你的 DEL 就删了别人的锁
var unlockScript = redis.NewScript(`
if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("DEL", KEYS[1])
end
return 0
`)

type DistLock struct {
    rdb   *redis.Client
    key   string
    token string        // 步骤2：本次持锁的唯一凭证，防止误删他人锁
    ttl   time.Duration
}

func New(rdb *redis.Client, key string, ttl time.Duration) *DistLock {
    return &DistLock{rdb: rdb, key: key, token: uuid.NewString(), ttl: ttl}
}

func (l *DistLock) TryLock(ctx context.Context) (bool, error) {
    // 步骤3：SET key token NX PX ttl —— 一条命令同时完成"不存在才设"和"带过期时间"。
    // TTL 是保命符：持锁者崩溃了，锁也能自动释放，否则全局永久死锁
    return l.rdb.SetNX(ctx, l.key, l.token, l.ttl).Result()
}

// 步骤4：带重试的阻塞获取，必须响应 ctx 取消，否则调用方无法超时退出
func (l *DistLock) Lock(ctx context.Context, retryInterval time.Duration) error {
    for {
        ok, err := l.TryLock(ctx)
        if err != nil {
            return err
        }
        if ok {
            return nil
        }
        select {
        case <-ctx.Done():
            return errors.New("获取分布式锁超时: " + ctx.Err().Error())
        case <-time.After(retryInterval):
            // 继续下一轮重试
        }
    }
}

func (l *DistLock) Unlock(ctx context.Context) error {
    // 步骤5：只删自己的锁
    return unlockScript.Run(ctx, l.rdb, []string{l.key}, l.token).Err()
}

// runtime 中 Mutex.state 的位布局（简化示意）
const (
    mutexLocked      = 1 << 0 // 第 0 位：是否被持有
    mutexWoken       = 1 << 1 // 第 1 位：是否有 waiter 已被唤醒
    mutexStarving    = 1 << 2 // 第 2 位：是否进入饥饿模式
    mutexWaiterShift = 3      // 第 3 位起：等待者计数
)

// locked/woken/starving 三个标志位，加上高 29 位的 waiter 数量，
// 全部打包在一个 int32 里，用 atomic 位运算读写

// 简化逻辑：写锁 Lock
func (rw *RWMutex) Lock() {
    rw.w.Lock()                    // 步骤1：先和其他写者互斥
    r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) // 步骤2：把读者计数整体"扣掉"，标记有写者等待
    if r != 0 {                    // 步骤3：还有读者没走
        atomic.Wait(&rw.writerSem) // 步骤4：写者睡在 writerSem 上，等读者全退场
    }
}

// 简化逻辑：读锁 RLock
func (rw *RWMutex) RLock() {
    if atomic.AddInt32(&rw.readerCount, 1) < 0 {
        // 步骤1：读者计数变负，说明有写者正在等待 -> 读者也要排队
        atomic.Wait(&rw.readerSem) // 步骤2：读者睡在 readerSem 上，等写者释放
    }
}

func badUpgrade(rw *sync.RWMutex) {
    rw.RLock()
    defer rw.RUnlock()
    // 危险：在持有读锁时申请写锁
    rw.Lock()   // 写者会把 readerCount 整体扣掉，但本 goroutine 的 +1 还在里面
    defer rw.Unlock()
    // 结果：写者永远等不到 readerCount 归零（自己就是那个读者）-> 死锁
}

package main

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

func main() {
    var mu sync.Mutex
    cond := sync.NewCond(&mu)

    // 步骤1：启动 3 个等待者
    for i := 1; i <= 3; i++ {
        go func(id int) {
            mu.Lock()
            defer mu.Unlock()
            cond.Wait() // 进入等待（Wait 会先释放锁再阻塞）
            fmt.Printf("等待者 %d 被唤醒\n", id)
        }(i)
    }

    time.Sleep(300 * time.Millisecond)

    // 步骤2：Signal 只唤醒 1 个 -> 只有 1 个等待者打印
    mu.Lock()
    cond.Signal()
    mu.Unlock()
    time.Sleep(200 * time.Millisecond) // 此时只有 1 个打印

    // 步骤3：Broadcast 唤醒剩余全部 -> 剩下 2 个一起打印
    mu.Lock()
    cond.Broadcast()
    mu.Unlock()

    time.Sleep(300 * time.Millisecond)
}