一、并发编程

// context 包的核心接口
// Context 接口定义了四个方法，每个方法都有特定的用途
type Context interface {
    // Deadline 返回两个值：
    //   deadline：过期时间点（time.Time）
    //   ok：是否设置了过期时间。false 表示没有设置过期时间
    // 用途：在执行耗时操作前，可以先检查是否快过期了，避免做无用功
    Deadline() (deadline time.Time, ok bool)

    // Done 返回一个 channel
    // 当 Context 被取消或过期时，这个 channel 会被关闭（close）
    // 用法：配合 select-case 监听，一旦 channel 关闭就退出当前操作
    // 这是最常用的方法，几乎所有超时控制都依赖它
    Done() <-chan struct{}

    // Err 返回 Context 被取消的原因
    //   context.Canceled：被主动取消（调用了 cancel 函数）
    //   context.DeadlineExceeded：超时过期
    // 当 Done() 的 channel 还没关闭时，Err() 返回 nil
    Err() error

    // Value 根据 key 获取 Context 中存储的值
    // 用途：在请求链路中传递数据，如 trace ID、用户信息等
    // 注意：key 必须是可比较的类型，通常用自定义类型的指针
    Value(key any) any
}

package main

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

func main() {
    // ==================== 1. context.WithValue ====================
    // WithValue 在父 Context 的基础上，附加一个 key-value 对
    // 返回一个新的 Context（原 Context 不受影响）
    // 用途：在请求链路中传递数据，如 trace ID、用户 ID 等

    // 创建根 Context（context.Background() 返回一个空的 Context）
    ctx := context.Background()

    // 往 Context 中存入一个 key-value
    // key 用自定义类型，避免和其他人的 key 冲突
    type ctxKey string
    ctx = context.WithValue(ctx, ctxKey("userID"), 12345)

    // 从 Context 中取出值
    // 注意：返回值是 any 类型，需要做类型断言
    val := ctx.Value(ctxKey("userID"))
    fmt.Printf("userID: %v\n", val) // 输出: userID: 12345

    // ==================== 2. context.WithCancel ====================
    // WithCancel 返回一个可取消的 Context 和一个取消函数 cancel
    // 调用 cancel() 后，ctx.Done() 返回的 channel 会被关闭
    // 用途：手动控制取消时机

    ctx2, cancel := context.WithCancel(context.Background())
    // 启动一个 goroutine，监听取消信号
    go func() {
        select {
        case <-ctx2.Done():
            // 当 cancel() 被调用时，Done() 的 channel 关闭，这里就会执行
            fmt.Println("收到取消信号，停止工作")
        }
    }()
    // 模拟做一些工作
    time.Sleep(100 * time.Millisecond)
    // 主动取消
    cancel()
    time.Sleep(50 * time.Millisecond) // 等待 goroutine 打印

    // ==================== 3. context.WithDeadline ====================
    // WithDeadline 在指定的时间点过期
    // 参数 deadline 是一个绝对时间点
    ctx3, cancel3 := context.WithDeadline(context.Background(), time.Now().Add(2*time.Second))
    defer cancel3() // 好习惯：即使没用也要 defer cancel，释放资源

    select {
    case <-ctx3.Done():
        fmt.Println("到达截止时间:", ctx3.Err()) // 输出: context deadline exceeded
    }

    // ==================== 4. context.WithTimeout ====================
    // WithTimeout 在指定的时间段后过期
    // 本质上和 WithDeadline 一样，只是参数从"时间点"变成了"时间段"
    // WithTimeout(parent, d) 等价于 WithDeadline(parent, time.Now().Add(d))
    ctx4, cancel4 := context.WithTimeout(context.Background(), 500*time.Millisecond)
    defer cancel4()

    select {
    case <-ctx4.Done():
        fmt.Println("超时了:", ctx4.Err()) // 输出: context deadline exceeded
    }
}

package main

import (
    "context"
    "fmt"
    "net/http"
)

// ==================== 定义 Context Key ====================
// 用自定义类型作为 key，避免不同包之间的 key 冲突
// 这是一个未导出的类型（小写开头），只有当前包能用
type traceIDKey string

// 定义 key 常量
const TraceIDKey traceIDKey = "trace_id"

// ==================== 中间件：注入 trace ID ====================
// 在 HTTP 请求进来时，生成一个 trace ID 并存入 Context
func traceMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        // 生成一个唯一的 trace ID（实际项目中可以用 UUID）
        traceID := "trace-" + r.RemoteAddr

        // 将 trace ID 存入 Context
        // r.Context() 是 Go 1.7+ 提供的，返回请求自带的 Context
        ctx := context.WithValue(r.Context(), TraceIDKey, traceID)

        // 创建一个新的请求，带上新的 Context
        // 因为 http.Request 的 Context 是不可变的，只能创建新请求
        r = r.WithContext(ctx)

        // 继续处理
        next.ServeHTTP(w, r)
    })
}

// ==================== 业务处理函数 ====================
func handleUser(w http.ResponseWriter, r *http.Request) {
    // 从 Context 中取出 trace ID
    // 注意要做类型断言，因为 Value() 返回 any 类型
    traceID, ok := r.Context().Value(TraceIDKey).(string)
    if !ok {
        traceID = "unknown"
    }

    // 调用数据库查询，把 trace ID 传递下去
    // 这就是"安全传递数据"——通过 Context 一路传下去
    user, err := queryUser(r.Context(), 1)
    if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }

    fmt.Fprintf(w, "traceID=%s, user=%+v\n", traceID, user)
}

// ==================== 数据库查询函数 ====================
// 所有公共方法都应该接受 context.Context 作为第一个参数
// 这样可以在整条调用链中传递 Context
func queryUser(ctx context.Context, userID int) (string, error) {
    // 数据库查询时也可以用到 trace ID（比如记录日志）
    traceID, _ := ctx.Value(TraceIDKey).(string)
    fmt.Printf("[DB] traceID=%s, 正在查询用户 %d\n", traceID, userID)

    // 模拟数据库查询
    // 实际项目中，这里会检查 ctx.Done() 是否已关闭
    select {
    case <-ctx.Done():
        // Context 被取消（比如客户端断开了连接），直接返回
        return "", ctx.Err()
    default:
        // 正常执行查询
    }

    return "张三", nil
}

func main() {
    mux := http.NewServeMux()
    mux.HandleFunc("/user", handleUser)

    // 使用中间件包裹
    http.ListenAndServe(":8080", traceMiddleware(mux))
}

package main

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

func main() {
    // 父 Context（根节点）
    // context.Background() 是所有 Context 的根
    parentCtx, parentCancel := context.WithCancel(context.Background())

    // 子 Context：从父 Context 派生
    // 子 Context 继承了父 Context 的取消控制
    childCtx, childCancel := context.WithCancel(parentCtx)
    defer childCancel() // 子 Context 也要 cancel，释放资源

    // 孙子 Context：从子 Context 派生
    grandchildCtx, _ := context.WithTimeout(childCtx, 10*time.Second)

    // ==================== 规则一：父取消，子也取消 ====================
    // 当父亲取消时，所有派生的子 Context 都会被取消
    // 这就是"控制从上至下"
    go func() {
        <-grandchildCtx.Done()
        fmt.Println("孙子 Context 被取消了:", grandchildCtx.Err())
    }()

    // 取消父 Context
    // 即使孙子 Context 设置了 10 秒超时，也会因为父取消而立即取消
    parentCancel()
    time.Sleep(100 * time.Millisecond)

    // ==================== 规则二：子不能控制父 ====================
    // 子 Context 设置的超时时间不能比父的长
    // 如果父 5 秒后过期，子设置 10 秒，子实际上还是 5 秒就过期
    parent2, _ := context.WithTimeout(context.Background(), 5*time.Second)
    child2, _ := context.WithTimeout(parent2, 10*time.Second)
    // child2 实际上 5 秒后就过期了，因为 parent2 先过期

    // 但子可以设置更短的超时
    parent3, _ := context.WithTimeout(context.Background(), 10*time.Second)
    child3, _ := context.WithTimeout(parent3, 3*time.Second)
    // child3 在 3 秒后过期，parent3 在 10 秒后过期
    // 子比父先过期是允许的

    _ = child2
    _ = child3
    fmt.Println("父子关系演示完成")
}

// ==================== valueCtx 的简化实现 ====================
// 帮助你理解 context.WithValue 内部是怎么工作的

// valueCtx 嵌入了 Context 接口（这就是装饰器模式）
// 它在原有 Context 的基础上，多存了一个 key-value 对
type valueCtx struct {
    Context       // 匿名嵌入：这是"装饰器模式"的核心
    key, val any  // 只能存储一个 key-value 对
}

// Value 方法的实现：查找 key 时，先看自己有没有，没有就找父节点
func (c *valueCtx) Value(key any) any {
    // 如果 key 匹配自己的 key，直接返回
    if c.key == key {
        return c.val
    }
    // 否则委托给父 Context 去找
    // 这就是"查找从下至上"的实现原理
    return c.Context.Value(key)
}

// ==================== 为什么要这样设计？ ====================
// 假设有以下调用链：
//   ctx1 = WithValue(background, "a", 1)
//   ctx2 = WithValue(ctx1, "b", 2)
//   ctx3 = WithValue(ctx2, "c", 3)
//
// 查找 "a" 的过程：
//   ctx3.Value("a") → 不是 "c"，找父节点 ctx2
//   ctx2.Value("a") → 不是 "b"，找父节点 ctx1
//   ctx1.Value("a") → 是 "a"，返回 1
//
// 查找 "d"（不存在）的过程：
//   ctx3.Value("d") → 找父 → ctx2 → 找父 → ctx1 → 找父 → background
//   background.Value("d") → 返回 nil（根节点没有存任何值）

// ==================== cancelCtx 的简化实现 ====================

type cancelCtx struct {
    Context

    // done 是一个 channel，被 close 后表示取消
    // 用 sync.Once 确保只关闭一次（避免重复 close 导致 panic）
    done atomic.Value // 存储一个 chan struct{}

    // children 记录所有从这个 cancelCtx 派生的子 cancelCtx
    // 取消时会遍历 children，挨个取消
    children map[canceler]struct{}
}

// canceler 接口：任何可以被取消的类型都实现这个接口
type canceler interface {
    cancel(removeFromParent bool, err error)
    Done() <-chan struct{}
}

// ==================== 取消的传播过程 ====================
// 假设有以下 Context 树：
//
//              cancelCtxA
//             /    |    \
//        valueCtx  cancelCtxB  timerCtxC
//                   |            |
//               valueCtx     cancelCtxD
//
// 当 cancelCtxA 被取消时：
// 1. 遍历 cancelCtxA.children，找到 cancelCtxB 和 timerCtxC
// 2. 调用 cancelCtxB.cancel() → 关闭 B 的 done channel
// 3. 调用 timerCtxC.cancel() → 关闭 C 的 done channel，停掉 C 的定时器
// 4. cancelCtxB 再遍历自己的 children，取消 valueCtx...
// 5. 子子孙孙，全部取消

// ==================== 关键细节：children 维护 ====================
// 注意：不是所有 Context 类型都会被加入 children
// 只有 cancelCtx 和 timerCtx 会被加入
// valueCtx 不会被加入，因为它没有取消功能

// 儿子把自己加入父亲的 children 的过程：
// 1. 新建一个 cancelCtx（或 timerCtx）
// 2. 从自己开始往上找，找到第一个 cancelCtx 类型的祖先
// 3. 把自己加入那个祖先的 children map 中
// 4. 如果祖先已经取消了，立刻取消自己

// ==================== cancel 方法做了两件事 ====================
func (c *cancelCtx) cancel(removeFromParent bool, err error) {
    // 1. 遍历所有 children，挨个调用 cancel（递归取消）
    for child := range c.children {
        child.cancel(false, err) // false 表示不需要从父节点移除自己（因为父节点也在取消）
    }
    c.children = nil

    // 2. 关闭 done channel（符合"谁创建谁关闭"的原则）
    close(c.done.(chan struct{}))
}

// ==================== timerCtx 的简化实现 ====================

type timerCtx struct {
    cancelCtx          // 嵌入 cancelCtx，复用取消功能
    timer    *time.Timer // 定时器，到时间后自动触发取消
    deadline time.Time   // 过期时间点
}

// ==================== WithTimeout 的本质 ====================
// WithTimeout 和 WithDeadline 本质是一样的
// WithTimeout(parent, d) = WithDeadline(parent, time.Now().Add(d))
//
// WithDeadline 内部的实现：
// 1. 创建一个 timerCtx
// 2. 用 time.AfterFunc 设置定时器
// 3. 定时器触发时，自动调用 cancelCtx.cancel()
// 4. 这样就实现了"到了时间自动取消"

func WithDeadline(parent Context, d time.Time) (Context, CancelFunc) {
    // 如果父 Context 的过期时间更早，就没必要创建 timerCtx
    // 因为父先过期，子一定会跟着过期
    if cur, ok := parent.Deadline(); ok && cur.Before(d) {
        // 父比子先过期，直接用 WithCancel 就够了
        return WithCancel(parent)
    }

    c := &timerCtx{
        cancelCtx: newCancelCtx(parent),
        deadline:  d,
    }

    // 把自己加入父节点的 children（传播取消信号）
    propagateCancel(parent, c)

    // 计算距离过期还有多久
    dur := time.Until(d)
    if dur <= 0 {
        // 已经过期了，直接取消
        c.cancel(true, context.DeadlineExceeded)
        return c, func() { c.cancel(false, Canceled) }
    }

    // 设置定时器，到时间自动取消
    c.timer = time.AfterFunc(dur, func() {
        c.cancel(true, context.DeadlineExceeded)
    })

    return c, func() { c.cancel(true, Canceled) }
}

package main

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

// ==================== 模拟一个耗时的数据库查询 ====================
func slowDBQuery(ctx context.Context) (string, error) {
    // 在执行查询前，先检查 context 是否已经超时
    // 这是一种常见的优化：如果已经超时了，就不用发请求了
    if ctx.Err() != nil {
        return "", ctx.Err()
    }

    // 模拟耗时查询（3 秒）
    select {
    case <-time.After(3 * time.Second):
        return "查询结果", nil
    case <-ctx.Done():
        // context 超时或被取消
        return "", ctx.Err()
    }
}

func main() {
    // 创建一个 2 秒超时的 Context
    // 数据库查询需要 3 秒，但 context 只有 2 秒
    // 所以查询会在 2 秒后被取消
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    // 好习惯：defer cancel() 确保资源被释放
    defer cancel()

    result, err := slowDBQuery(ctx)
    if err != nil {
        fmt.Println("查询失败:", err)
        // 输出: 查询失败: context deadline exceeded
        return
    }
    fmt.Println("查询成功:", result)
}

package main

import (
    "context"
    "fmt"
    "time"

    "golang.org/x/sync/errgroup"
)

func main() {
    // errgroup.WithContext 返回一个 Group 和一个 Context
    // 如果 Group 中的任何一个任务返回 error，这个 Context 就会被取消
    // 这样其他正在运行的任务就能通过 ctx.Done() 收到取消信号
    g, ctx := errgroup.WithContext(context.Background())

    // 任务 1：模拟一个会成功的任务
    g.Go(func() error {
        time.Sleep(1 * time.Second)
        fmt.Println("任务 1 完成")
        return nil
    })

    // 任务 2：模拟一个会失败的任务
    g.Go(func() error {
        time.Sleep(500 * time.Millisecond)
        fmt.Println("任务 2 失败了")
        return fmt.Errorf("任务 2 出错了")
        // 此时 ctx 会被取消，任务 1 和任务 3 会收到取消信号
    })

    // 任务 3：监听 ctx，一旦取消就退出
    g.Go(func() error {
        select {
        case <-ctx.Done():
            fmt.Println("任务 3 收到取消信号，提前退出")
            return ctx.Err()
        case <-time.After(5 * time.Second):
            fmt.Println("任务 3 正常完成")
            return nil
        }
    })

    // Wait 等待所有任务完成
    // 返回第一个出错的任务的 error
    if err := g.Wait(); err != nil {
        fmt.Println("最终结果:", err)
    }
}

// ==================== 规则一：只用作方法参数 ====================
// context.Context 应该作为方法的第一个参数传递
// 不要存到结构体字段里（除非你的结构体本身就是表达"请求上下文"的概念）

// ✅ 正确用法：作为第一个参数
func GetUser(ctx context.Context, userID int) (*User, error) {
    // ...
    return nil, nil
}

// ❌ 错误用法：存到结构体字段
// type UserService struct {
//     ctx context.Context  // 不要这样做！
// }

// ==================== 规则二：所有公共方法都加 context 参数 ====================
// 除非是 util、helper 之类的工具方法
// 这是 Go 社区的共识，也是代码规范

// ✅ 正确
func (s *Service) DoSomething(ctx context.Context, req *Request) (*Response, error) {
    // ...
    return nil, nil
}

// ==================== 规则三：不要用 nil Context ====================
// 如果没有 Context 可传，用 context.Background() 或 context.TODO()

// ✅ 正确
result := ProcessData(context.Background(), data)

// ❌ 错误（会 panic）
// result := ProcessData(nil, data)

// ==================== 规则四：父无法访问子的值 ====================
// 父 Context 无法获取子 Context 中 WithValue 存储的值
// 如果确实需要在父中访问子设置的值，可以在父中放一个 map，后续修改这个 map

// ==================== 规则五：WithTimeout 和 WithCancel 返回的 cancel 函数必须调用 ====================
// 即使函数提前返回了，也要调用 cancel 释放资源
// 最简单的方式是用 defer

// ✅ 正确
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel() // 确保无论如何都会调用 cancel

package main

import (
    "fmt"
    "sync"
)

// ==================== 1. Mutex 基本用法 ====================
// 用 Mutex 保护一个计数器
type Counter struct {
    mu    sync.Mutex // 互斥锁，保护 count 字段
    count int        // 被保护的共享变量
}

func (c *Counter) Increment() {
    c.mu.Lock()         // 加锁：其他 goroutine 调用 Lock 会阻塞等待
    c.count++           // 安全地修改共享变量
    c.mu.Unlock()       // 解锁：让等待的 goroutine 可以继续
}

func (c *Counter) Get() int {
    c.mu.Lock()
    defer c.mu.Unlock() // 用 defer 确保一定会解锁，即使 panic 也会执行
    return c.count
}

// ==================== 2. RWMutex 基本用法 ====================
// 读写锁：适合"读多写少"的场景
// 多个读操作可以并发执行，写操作需要独占
type SafeCache struct {
    mu   sync.RWMutex // 读写锁
    data map[string]string
}

func (c *SafeCache) Get(key string) (string, bool) {
    // 读锁：允许多个 goroutine 同时读
    // 但如果有 goroutine 持有写锁，读锁会阻塞
    c.mu.RLock()
    defer c.mu.RUnlock()
    val, ok := c.data[key]
    return val, ok
}

func (c *SafeCache) Set(key, val string) {
    // 写锁：独占，其他 goroutine 既不能读也不能写
    c.mu.Lock()
    defer c.mu.Unlock()
    c.data[key] = val
}

func main() {
    // 测试 Mutex
    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.Get()) // 输出: Counter: 1000

    // 测试 RWMutex
    cache := &SafeCache{data: make(map[string]string)}
    cache.Set("name", "张三")
    val, _ := cache.Get("name")
    fmt.Println("Cache:", val) // 输出: Cache: 张三
}

package main

import (
    "fmt"
    "sync"
)

// ==================== SafeMap：线程安全的 map ====================
type SafeMap struct {
    mu   sync.RWMutex
    data map[string]any
}

func NewSafeMap() *SafeMap {
    return &SafeMap{
        data: make(map[string]any),
    }
}

// LoadOrStore：如果 key 存在就返回已有值，不存在就存入新值
// 这是一个非常常见的方法，Go 标准库 sync.Map 也有同名方法
func (m *SafeMap) LoadOrStore(key string, val any) (actual any, loaded bool) {
    // ==================== 第一次检查：加读锁 ====================
    // 先用读锁快速检查一下 key 是否已存在
    m.mu.RLock()
    if v, ok := m.data[key]; ok {
        // key 已存在，直接返回已有值
        m.mu.RUnlock()
        return v, true // loaded=true 表示是加载的已有值
    }
    m.mu.RUnlock()

    // ==================== 第二次检查：加写锁 ====================
    // 为什么要再检查一遍？
    // 因为从释放读锁到获取写锁之间，可能有其他 goroutine 已经写入了这个 key
    // 这就是"double-check"（双重检查）模式
    m.mu.Lock()
    defer m.mu.Unlock()

    // 再次检查 key 是否存在
    if v, ok := m.data[key]; ok {
        // 在获取写锁期间，其他 goroutine 已经写入了，返回已有值
        return v, true
    }

    // key 确实不存在，存入新值
    m.data[key] = val
    return val, false // loaded=false 表示是新存入的值
}

func (m *SafeMap) Load(key string) (any, bool) {
    m.mu.RLock()
    defer m.mu.RUnlock()
    v, ok := m.data[key]
    return v, ok
}

func (m *SafeMap) Delete(key string) {
    m.mu.Lock()
    defer m.mu.Unlock()
    delete(m.data, key)
}

func main() {
    m := NewSafeMap()
    var wg sync.WaitGroup

    // 100 个 goroutine 同时尝试写入同一个 key
    for i := 0; i < 100; i++ {
        wg.Add(1)
        go func(i int) {
            defer wg.Done()
            // LoadOrStore 保证只有一个 goroutine 真正写入
            val, loaded := m.LoadOrStore("key", i)
            if loaded {
                fmt.Printf("goroutine %d: key 已存在，值为 %v\n", i, val)
            } else {
                fmt.Printf("goroutine %d: 成功写入值 %d\n", i, i)
            }
        }(i)
    }
    wg.Wait()
}

// ==================== double-check 模式的通用模板 ====================

func (m *SafeMap) GetOrCompute(key string, compute func() any) any {
    // 第一次检查：读锁（快路径）
    // 大部分情况下 key 已经存在，读锁就够了，不需要加写锁
    m.mu.RLock()
    if v, ok := m.data[key]; ok {
        m.mu.RUnlock()
        return v // 快速返回，不需要加写锁
    }
    m.mu.RUnlock()

    // 第二次检查：写锁（慢路径）
    // 只有 key 不存在时才走到这里
    m.mu.Lock()
    defer m.mu.Unlock()

    // 必须再检查一次！
    // 因为在 RUnlock 和 Lock 之间，可能有其他 goroutine 已经写入了
    if v, ok := m.data[key]; ok {
        return v // 其他 goroutine 已经写入了，直接用
    }

    // 确实不存在，计算新值并写入
    val := compute()
    m.data[key] = val
    return val
}

// ==================== 为什么不能只检查一次？ ====================
// 假设只检查一次（只加写锁）：
//
// func (m *SafeMap) GetOrCompute(key string) any {
//     m.mu.Lock()
//     defer m.mu.Unlock()
//     if v, ok := m.data[key]; ok {
//         return v
//     }
//     m.data[key] = compute() // 计算可能很耗时
//     return m.data[key]
// }
//
// 问题：每次调用都要加写锁，即使 key 大部分时候已经存在
// 写锁是排他的，会阻塞所有读操作，性能很差
//
// double-check 的优势：
// - key 已存在时（常见情况），只用读锁，不阻塞其他读操作
// - key 不存在时（少见情况），才加写锁

// ==================== Mutex 的内部结构 ====================
type Mutex struct {
    state int32 // 锁状态：记录锁是否被持有、是否有等待者、处于哪种模式
    sema  uint32 // 信号量：用于挂起和唤醒 goroutine
}

// ==================== 锁实现的一般模板 ====================
// 大部分语言的锁都遵循这个模板：
//
// 加锁：
//   1. 快路径：CAS 尝试加锁（自旋），成功就直接返回
//   2. 慢路径：加入等待队列，被挂起，等待被唤醒
//
// 解锁：
//   1. CAS 修改状态
//   2. 如果有等待者，唤醒队列头的 goroutine

// ==================== 正常模式 ====================
// 当一个 goroutine 被唤醒后，它不会立刻获得锁
// 而是和新来的 goroutine 竞争锁
//
// 为什么？因为被唤醒的 goroutine 需要重新调度（可能不在 CPU 上了）
// 而新来的 goroutine 正在 CPU 上运行，大概率能抢到锁
// 这样做的好处：减少 goroutine 调度开销，提高吞吐量
//
// 缺点：可能导致队列中的 goroutine 长时间拿不到锁（饥饿）

// ==================== 饥饿模式 ====================
// 如果一个 goroutine 等待锁的时间超过 1ms
// Mutex 会切换到饥饿模式
//
// 在饥饿模式下：
// - 锁直接交给队列头的 goroutine，新来的不能抢
// - 保证公平性，先到先得
//
// 退出饥饿模式的条件：
// - 队列中只剩最后一个 goroutine
// - 或者 goroutine 的等待时间小于 1ms

// 解锁相对简单
// 1. atomic 操作修改 state（解除锁定状态）
// 2. 如果发现有等待者，进入慢路径
// 3. 慢路径：根据当前模式（正常/饥饿），唤醒对应的 goroutine

package main

import (
    "fmt"
    "sync"
)

func main() {
    var mu sync.Mutex

    // ==================== 注意事项一：锁不可重入 ====================
    // Go 的 Mutex 不支持重入
    // 所谓"重入"是指：同一个 goroutine 可以多次获取同一把锁
    // Java 的 ReentrantLock 支持重入，但 Go 不支持

    // ❌ 这会导致死锁！
    // funcA 获取了锁，调用 funcB，funcB 又要获取同一把锁
    // funcB 会永远阻塞，因为 funcA 还没释放锁
    //
    // funcA() {
    //     mu.Lock()
    //     defer mu.Unlock()
    //     funcB() // funcB 里面也要 mu.Lock() → 死锁！
    // }

    // ✅ 解决方案：拆分出不需要加锁的内部方法
    // funcA() {
    //     mu.Lock()
    //     defer mu.Unlock()
    //     funcBInternal() // 不加锁的版本
    // }
    // funcB() {
    //     mu.Lock()
    //     defer mu.Unlock()
    //     funcBInternal()
    // }

    // ==================== 注意事项二：用 defer 解锁 ====================
    // 如果在加锁和解锁之间发生了 panic，锁就永远不会被释放
    // defer 可以确保即使 panic，锁也会被释放

    // ❌ 危险：如果 doSomething() panic 了，锁就不会释放
    // mu.Lock()
    // doSomething() // 可能 panic
    // mu.Unlock()

    // ✅ 安全：defer 确保无论如何都会解锁
    mu.Lock()
    defer mu.Unlock()
    fmt.Println("安全地做了一些事情")

    // ==================== 注意事项三：RWMutex 适合读多写少 ====================
    // 如果写操作很多，RWMutex 反而比 Mutex 慢
    // 因为 RWMutex 维护读者计数有额外开销
    // 而且写锁要等所有读者释放，可能导致写延迟
}

package main

import (
    "fmt"
    "sync"
)

// ==================== 单例模式 ====================
type Database struct {
    name string
}

var (
    dbInstance *Database
    dbOnce     sync.Once // 确保 dbInstance 只被初始化一次
)

// GetDB 获取数据库单例
// 无论调用多少次，Database 只会被创建一次
func GetDB() *Database {
    // Do 方法接收一个函数，确保这个函数只执行一次
    // 即使 100 个 goroutine 同时调用 GetDB，初始化也只会执行一次
    dbOnce.Do(func() {
        fmt.Println("正在初始化数据库连接...")
        // 模拟耗时的初始化操作
        dbInstance = &Database{name: "mysql-localhost:3306"}
        fmt.Println("数据库初始化完成")
    })
    return dbInstance
}

func main() {
    var wg sync.WaitGroup

    // 10 个 goroutine 同时获取数据库实例
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(i int) {
            defer wg.Done()
            db := GetDB()
            fmt.Printf("goroutine %d 拿到数据库: %s\n", i, db.name)
        }(i)
    }
    wg.Wait()

    // 输出：
    // 正在初始化数据库连接...
    // 数据库初始化完成
    // goroutine 0 拿到数据库: mysql-localhost:3306
    // goroutine 1 拿到数据库: mysql-localhost:3306
    // ...（"正在初始化"只会打印一次）
}

// ==================== sync.Once 的简化实现 ====================

type Once struct {
    done uint32 // 0 表示未执行，1 表示已执行
    m    Mutex  // 互斥锁，确保只有一个 goroutine 执行初始化
}

func (o *Once) Do(f func()) {
    // ==================== 第一次检查：原子读（快路径）====================
    // 用原子操作代替读锁，性能更好
    // 如果已经执行过了（done == 1），直接返回，不需要加锁
    if atomic.LoadUint32(&o.done) == 0 {
        // ==================== 第二次检查：加锁后再检查（慢路径）====================
        o.doSlow(f)
    }
}

func (o *Once) doSlow(f func()) {
    o.m.Lock()
    defer o.m.Unlock()

    // 加锁后再检查一次 done
    // 因为在第一次检查和加锁之间，可能其他 goroutine 已经执行完了
    if o.done == 0 {
        // 确实没执行过，执行函数
        defer atomic.StoreUint32(&o.done, 1) // 标记为已执行
        f()
    }
    // 如果 done == 1，说明其他 goroutine 已经执行了，什么都不做
}

// ==================== 和 RWMutex 版 double-check 的对比 ====================
// RWMutex 版：
//   1. RLock → 检查 → RUnlock（读锁）
//   2. Lock → 检查 → Unlock（写锁）
//
// Once 版（用原子操作代替读锁）：
//   1. atomic.Load → 检查（无锁，极快）
//   2. Lock → 检查 → Unlock（互斥锁）
//
// 原子操作比读锁快得多，因为不需要进入 runtime 调度

package main

import (
    "fmt"
    "sync"
)

// ==================== 模拟 Beego 的初始化模式 ====================

var initOnce sync.Once

// initBeforeHTTPRun 可能在多个地方被调用
// 使用 Once 确保初始化只执行一次
func initBeforeHTTPRun() {
    initOnce.Do(func() {
        fmt.Println("1. 注册路由...")
        fmt.Println("2. 加载配置...")
        fmt.Println("3. 连接数据库...")
        fmt.Println("4. 初始化缓存...")
        fmt.Println("初始化完成！")
    })
}

func main() {
    // 模拟多处调用
    initBeforeHTTPRun() // 第一次调用，执行初始化
    initBeforeHTTPRun() // 第二次调用，什么都不做
    initBeforeHTTPRun() // 第三次调用，什么都不做
}

package main

import (
    "bytes"
    "fmt"
    "sync"
)

// ==================== sync.Pool 基本用法 ====================

func main() {
    // 创建一个 Pool
    // New 函数：当 Pool 中没有可用对象时，调用此函数创建新对象
    bufferPool := &sync.Pool{
        New: func() any {
            // 当 Pool 为空时，创建一个新的 bytes.Buffer
            // 首次大小设为 512 字节，减少扩容次数
            b := bytes.NewBuffer(make([]byte, 0, 512))
            fmt.Println("  [Pool] 创建了新的 Buffer")
            return b
        },
    }

    // ==================== Get：从 Pool 获取对象 ====================
    // 第一次 Get，Pool 为空，会调用 New 创建
    buf1 := bufferPool.Get().(*bytes.Buffer)
    buf1.WriteString("hello")
    fmt.Printf("buf1 内容: %q\n", buf1.String())

    // 使用完毕后，必须重置对象状态！
    // 否则下一个使用者会拿到带着上一次数据的"脏"对象
    buf1.Reset()

    // ==================== Put：把对象放回 Pool ====================
    // 放回去后，对象可以被其他 goroutine 复用
    bufferPool.Put(buf1)

    // 第二次 Get，Pool 中有刚才放回去的 buf1
    // 不会调用 New，直接复用
    buf2 := bufferPool.Get().(*bytes.Buffer)
    fmt.Printf("buf2 内容: %q（应该是空的，因为 Reset 了）\n", buf2.String())
    buf2.Reset()
    bufferPool.Put(buf2)

    // ==================== 重要特性：GC 会清理 Pool ====================
    // sync.Pool 中缓存的对象会在 GC 时被清理
    // 所以 Pool 不适合用来做真正的缓存（比如缓存数据库查询结果）
    // 它只适合复用临时对象
}

// ==================== sync.Pool GET 步骤 ====================
//
// 比喻：你准备买东西
//
// 步骤 1：先看自己有没有（private）
//   → 检查当前 P 的 private 字段，如果有直接返回
//   → 这是最快的路径，完全无竞争
//
// 步骤 2：没有找爸妈借（shared poolChain）
//   → 从自己的 poolChain 头部开始找
//   → HEAD 指向最近创建的 ring buffer
//   → 从队头往队尾找
//
// 步骤 3：还没有找亲戚借（从别人的 poolChain 偷一个）
//   → 遍历其他 P 的 poolChain
//   → 从别人的队尾偷（偷的过程是全局并发的）
//   → 偷是最后一个需要竞争的步骤
//
// 步骤 4：再没有薅网贷（victim 缓刑）
//   → 找 victim 里的对象（上一轮 GC 前的 locals）
//   → victim 的对象即将被回收，能用就用
//
// 步骤 5：再没有就下海了（创建一个新的）
//   → 调用 New 函数创建新对象

// ==================== sync.Pool PUT 步骤 ====================
//
// 步骤 1：private 没放东西？直接放 private
//   → 最快路径，无竞争
//
// 步骤 2：否则，准备放 poolChain
//   → 步骤 2a：如果 HEAD 还没创建，创建一个 HEAD
//     → 创建一个容量为 8 的 ring buffer，把数据放进去
//
//   → 步骤 2b：如果 HEAD 的 ring buffer 没满
//     → 直接放进去
//
//   → 步骤 2c：如果 HEAD 的 ring buffer 已满
//     → 创建一个新节点，容量是 HEAD 的两倍
//     → 把数据放进新的 ring buffer
//     → 新节点成为新的 HEAD

// ==================== false sharing 问题示意 ====================

// ❌ 有 false sharing 问题
// 两个 poolLocal 紧挨着，可能在同一个 CacheLine 中
// P0 修改 poolLocal[0]，P1 修改 poolLocal[1]
// 虽然它们改的是不同的字段，但会导致对方的 CacheLine 失效
type poolLocalWithoutPad struct {
    private any    // 8 字节
    shared  chain  // 8 字节
}
// 总共 16 字节，4 个 poolLocal 刚好在一个 64 字节的 CacheLine 中

// ✅ Go 的做法：用 pad 字段补齐到 128 字节的整数倍
type poolLocal struct {
    private any    // P 私有对象
    shared  poolChain
    pad     [128]byte  // 填充字段，确保每个 poolLocal 独占 CacheLine
}
// 每个 poolLocal 占 128 字节，确保不会和其他 poolLocal 共享 CacheLine

package main

import (
    "fmt"
    "sync"
)

// ==================== bytebufferpool 核心思路的简化版 ====================

// 带容量控制的 Buffer Pool
type ByteBufferPool struct {
    // defaultSize：新创建的 buffer 的默认大小
    // 太小 → 频繁扩容（影响性能）
    // 太大 → 浪费内存
    defaultSize int

    // maxSize：超过这个大小的 buffer 不放回 Pool
    // 防止大 buffer 占用太多内存
    maxSize int

    pool sync.Pool
}

func NewByteBufferPool() *ByteBufferPool {
    return &ByteBufferPool{
        defaultSize: 512, // 默认 512 字节
        maxSize:     65536, // 最大 64KB，超过就不放回去
        pool: sync.Pool{
            New: func() any {
                // 创建一个 defaultSize 大小的 buffer
                b := make([]byte, 0, 512)
                return &b
            },
        },
    }
}

func (p *ByteBufferPool) Get() []byte {
    return *(p.pool.Get().(*[]byte))
}

func (p *ByteBufferPool) Put(b []byte) {
    // 超过 maxSize 的 buffer 不放回去，让它被 GC 回收
    // 这样防止 Pool 中缓存太多大 buffer，占用过多内存
    if cap(b) > p.maxSize {
        return // 不放回去，让 GC 回收
    }
    b = b[:0] // 重置长度，保留容量
    p.pool.Put(&b)
}

// ==================== bytebufferpool 的校准机制 ====================
// bytebufferpool 的一个亮点是"校准机制"：
// 它会统计不同大小的 buffer 被使用的次数
// 比如 0-64 bytes 的 buffer 被使用了 1000 次
//      64-128 bytes 的 buffer 被使用了 500 次
//
// 根据统计结果动态调整 defaultSize 和 maxSize：
// - defaultSize 设为使用最频繁的大小，减少扩容
// - maxSize 设为合理上限，避免大 buffer 浪费内存
//
// 这就是所谓的"分组统计使用次数 + 动态校准"

func main() {
    pool := NewByteBufferPool()

    // 获取 buffer
    buf := pool.Get()
    buf = append(buf, "hello world"...)
    fmt.Printf("使用 buffer: %q, 容量: %d\n", buf, cap(buf))

    // 归还 buffer
    pool.Put(buf)

    // 再次获取，复用刚才的 buffer
    buf2 := pool.Get()
    fmt.Printf("复用 buffer: %q, 容量: %d\n", buf2, cap(buf2))
}

package main

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

func main() {
    var wg sync.WaitGroup

    // ==================== WaitGroup 三步曲 ====================

    for i := 0; i < 5; i++ {
        // 步骤一：在启动 goroutine 之前 Add(1)
        // 必须在 goroutine 外面调用，否则可能 Wait 先于 Add 执行
        wg.Add(1)

        go func(id int) {
            // 步骤二：任务完成后 Done()（等价于 Add(-1)）
            defer wg.Done()

            fmt.Printf("任务 %d 开始执行\n", id)
            time.Sleep(time.Duration(id) * 100 * time.Millisecond) // 模拟耗时
            fmt.Printf("任务 %d 完成\n", id)
        }(i)
    }

    // 步骤三：Wait() 阻塞，直到计数器归零
    wg.Wait()
    fmt.Println("所有任务完成！")
}

// ==================== WaitGroup 的内部结构 ====================

type WaitGroup struct {
    // noCopy：告诉编译器这个结构体不能被复制
    // 复制 WaitGroup 会导致计数器状态不一致
    // 这是一个很实用的技巧，Dubbo-go 的 URL 结构体也用了
    noCopy noCopy

    // state1：在 64 位系统下
    //   高 32 位：还有多少任务没完成（counter）
    //   低 32 位：有多少 goroutine 在等 Wait()（waiter count）
    state1 uint64

    // state2：信号量，用于挂起/唤醒 goroutine
    // 和 Mutex 的 sema 字段原理一样
    state2 uint32
}

// ==================== 三个方法的实现原理 ====================

// Add：counter += delta
// 本质就是对 state1 的高 32 位做原子加法
// 如果 counter 降为 0，就唤醒所有等待的 goroutine
func (wg *WaitGroup) Add(delta int) {
    // 原子操作修改 state1 的高 32 位（counter）
    // 如果 counter 变为 0，说明所有任务完成
    // 从返回值中解析出 waiter count，逐个唤醒
}

// Done：counter -= 1
// 本质就是 Add(-1)
func (wg *WaitGroup) Done() {
    wg.Add(-1)
}

// Wait：waiter += 1，然后挂起当前 goroutine
func (wg *WaitGroup) Wait() {
    // 对 state1 的低 32 位做 CAS 加 1（waiter count）
    // 然后利用 state2（信号量）和 runtime_Semacquire 挂起 goroutine
    // 当 Add 发现 counter == 0 时，会调用 runtime_Semrelease 唤醒
}

package main

import (
    "context"
    "fmt"
    "time"

    "golang.org/x/sync/errgroup"
)

func main() {
    // ==================== errgroup.WithContext ====================
    // 返回 Group 和 Context
    // 任何一个任务返回 error，Context 会被取消
    g, ctx := errgroup.WithContext(context.Background())

    // 任务 1：成功
    g.Go(func() error {
        time.Sleep(1 * time.Second)
        fmt.Println("任务 1 完成")
        return nil
    })

    // 任务 2：失败
    g.Go(func() error {
        time.Sleep(500 * time.Millisecond)
        fmt.Println("任务 2 出错了")
        return fmt.Errorf("任务 2 失败")
        // 此时 ctx 被取消
    })

    // 任务 3：监听 ctx，提前退出
    g.Go(func() error {
        select {
        case <-ctx.Done():
            fmt.Println("任务 3 因其他任务失败而取消")
            return ctx.Err()
        case <-time.After(3 * time.Second):
            fmt.Println("任务 3 完成")
            return nil
        }
    })

    // Wait 返回第一个出错的 error
    if err := g.Wait(); err != nil {
        fmt.Println("结果:", err)
    }
    // 输出：
    // 任务 2 出错了
    // 任务 3 因其他任务失败而取消
    // 任务 1 完成
    // 结果: 任务 2 失败
}

package main

import (
    "fmt"
    "time"
)

func main() {
    // ==================== 1. 不带缓冲的 channel ====================
    // make(chan T) 创建不带缓冲的 channel
    // 特点：发送和接收必须同时准备好，否则阻塞
    // 就像面对面交接：你必须等对方伸手了，才能把东西递过去

    unbufCh := make(chan int)

    go func() {
        fmt.Println("发送者：准备发送数据...")
        unbufCh <- 42 // 阻塞，直到有接收者
        fmt.Println("发送者：数据被接收了")
    }()

    time.Sleep(500 * time.Millisecond) // 让发送者先阻塞
    fmt.Println("接收者：准备接收数据...")
    val := <-unbufCh // 接收数据，发送者不再阻塞
    fmt.Println("接收者：收到", val)
    time.Sleep(100 * time.Millisecond)

    // ==================== 2. 带缓冲的 channel ====================
    // make(chan T, N) 创建带 N 个缓冲的 channel
    // 特点：缓冲没满时发送不阻塞，缓冲没空时接收不阻塞
    // 就像快递柜：柜子有空格就能放，柜子满了就得等有人取走

    bufCh := make(chan int, 3) // 缓冲大小为 3

    // 可以连续发送 3 个数据而不阻塞
    bufCh <- 1
    bufCh <- 2
    bufCh <- 3
    fmt.Println("放了 3 个数据到缓冲 channel")

    // 再放第 4 个会阻塞（缓冲已满）
    // bufCh <- 4 // 这行会死锁！因为没有接收者

    // 接收数据
    fmt.Println("取出:", <-bufCh) // 1（先进先出）
    fmt.Println("取出:", <-bufCh) // 2
    fmt.Println("取出:", <-bufCh) // 3
}

package main

import (
    "fmt"
    "time"
)

// ==================== 思路一：当作队列传递数据 ====================
func producerConsumer() {
    // 生产者-消费者模式
    ch := make(chan string, 5)

    // 生产者 goroutine
    go func() {
        fruits := []string{"苹果", "香蕉", "橙子"}
        for _, fruit := range fruits {
            ch <- fruit // 发送数据
            fmt.Printf("生产: %s\n", fruit)
        }
        close(ch) // 发送完毕后关闭 channel
    }()

    // 消费者 goroutine
    for fruit := range ch { // range 遍历 channel，直到 channel 关闭
        fmt.Printf("消费: %s\n", fruit)
        time.Sleep(100 * time.Millisecond) // 模拟处理
    }
}

// ==================== 思路二：当作令牌控制并发数 ====================
func concurrencyLimiter() {
    // 利用 channel 的阻塞特性控制 goroutine 数量
    // 就像停车场：只有 3 个车位，满了就得等

    sem := make(chan struct{}, 3) // 最多 3 个并发

    for i := 0; i < 10; i++ {
        // 从信号 channel 获取"令牌"
        // 如果已有 3 个 goroutine 在运行，这里会阻塞
        sem <- struct{}{}

        go func(id int) {
            defer func() { <-sem }() // 完成后归还"令牌"

            fmt.Printf("goroutine %d 开始\n", id)
            time.Sleep(500 * time.Millisecond)
            fmt.Printf("goroutine %d 结束\n", id)
        }(i)
    }

    // 等待所有 goroutine 完成
    // 想要获取 3 个令牌，意味着等所有 goroutine 都归还
    for i := 0; i < 3; i++ {
        sem <- struct{}{}
    }
}

func main() {
    fmt.Println("=== 生产者-消费者 ===")
    producerConsumer()
    time.Sleep(500 * time.Millisecond)

    fmt.Println("\n=== 并发限制器 ===")
    concurrencyLimiter()
}

package main

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

// ==================== 简单的发布-订阅 ====================

type Broker struct {
    subscribers map[chan string]struct{} // 所有订阅者
    mu          sync.RWMutex
}

func NewBroker() *Broker {
    return &Broker{
        subscribers: make(map[chan string]struct{}),
    }
}

// Subscribe 订阅消息
// 每个订阅者获得自己的 channel
func (b *Broker) Subscribe() chan string {
    ch := make(chan string, 10) // 每个订阅者缓冲 10 条消息

    b.mu.Lock()
    b.subscribers[ch] = struct{}{}
    b.mu.Unlock()

    return ch
}

// Unsubscribe 取消订阅
func (b *Broker) Unsubscribe(ch chan string) {
    b.mu.Lock()
    delete(b.subscribers, ch)
    b.mu.Unlock()
    close(ch) // 关闭 channel，通知订阅者退出
}

// Publish 发布消息
// 注意：这里是一条消息只能被一个 goroutine 消费（channel 的特性）
// 所以需要遍历所有订阅者，给每个订阅者都发一份
func (b *Broker) Publish(msg string) {
    b.mu.RLock()
    defer b.mu.RUnlock()

    for ch := range b.subscribers {
        // 给每个订阅者的 channel 发送消息
        // 注意：这里用 select + default 避免阻塞
        // 如果订阅者的 channel 满了，就跳过（不阻塞发布者）
        select {
        case ch <- msg:
        default:
            fmt.Println("订阅者 channel 已满，丢弃消息")
        }
    }
}

func main() {
    broker := NewBroker()
    var wg sync.WaitGroup

    // 3 个订阅者
    for i := 0; i < 3; i++ {
        wg.Add(1)
        ch := broker.Subscribe()
        go func(id int, ch chan string) {
            defer wg.Done()
            for msg := range ch {
                fmt.Printf("订阅者 %d 收到: %s\n", id, msg)
            }
            fmt.Printf("订阅者 %d 退出\n", id)
        }(i, ch)
    }

    // 发布 5 条消息
    for i := 0; i < 5; i++ {
        broker.Publish(fmt.Sprintf("消息 %d", i))
    }

    time.Sleep(500 * time.Millisecond)
    // 取消所有订阅（关闭所有 channel）
    broker.mu.RLock()
    for ch := range broker.subscribers {
        close(ch)
    }
    broker.mu.RUnlock()

    wg.Wait()
}

package main

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

// ==================== 消息队列（支持消费组）====================

// Consumer 消费者接口
type Consumer interface {
    Consume(msg string)
}

// MessageQueue 消息队列
type MessageQueue struct {
    topicConsumers map[string][]chan string // 每个 topic 对应多个消费者的 channel
    mu             sync.RWMutex
}

func NewMessageQueue() *MessageQueue {
    return &MessageQueue{
        topicConsumers: make(map[string][]chan string),
    }
}

// Subscribe 订阅 topic
// 方案一：每个消费者订阅时创建一个子 channel
// 这样 Publish 时需要把消息复制到每个消费者的 channel
func (q *MessageQueue) Subscribe(topic string, bufSize int) chan string {
    q.mu.Lock()
    defer q.mu.Unlock()

    ch := make(chan string, bufSize)
    q.topicConsumers[topic] = append(q.topicConsumers[topic], ch)
    return ch
}

// Publish 发布消息到 topic
// 将消息广播给所有订阅了该 topic 的消费者
func (q *MessageQueue) Publish(topic string, msg string) {
    q.mu.RLock()
    defer q.mu.RUnlock()

    // 遍历所有订阅者，给每个人的 channel 都发一份
    for _, ch := range q.topicConsumers[topic] {
        select {
        case ch <- msg:
        default:
            // channel 满了，丢弃消息
        }
    }
}

func main() {
    mq := NewMessageQueue()
    var wg sync.WaitGroup

    // 3 个消费者订阅 "order" topic
    for i := 0; i < 3; i++ {
        wg.Add(1)
        ch := mq.Subscribe("order", 10)
        go func(id int, ch chan string) {
            defer wg.Done()
            for msg := range ch {
                fmt.Printf("消费者 %d 处理订单: %s\n", id, msg)
            }
        }(i, ch)
    }

    // 发布 5 条消息
    // 每条消息都会被 3 个消费者各收到一次（广播）
    for i := 0; i < 5; i++ {
        mq.Publish("order", fmt.Sprintf("订单#%d", i))
    }

    // 等待处理完成
    // 这里用 WaitGroup 简化处理，实际应该有更优雅的关闭机制
    wg2 := sync.WaitGroup{}
    wg2.Add(1)
    go func() {
        defer wg2.Done()
        time.Sleep(time.Second)
        // 关闭所有 channel
        mq.mu.Lock()
        for _, chs := range mq.topicConsumers {
            for _, ch := range chs {
                close(ch)
            }
        }
        mq.mu.Unlock()
    }()

    wg.Wait()
}

package main

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

// ==================== 任务池 ====================

// Task 表示一个任务
type Task func()

// TaskPool 任务池
type TaskPool struct {
    tasks    chan Task      // 任务队列
    wg       sync.WaitGroup // 等待所有任务完成
    workers  int            // worker 数量
}

// NewTaskPool 创建任务池
// workers：同时运行的 goroutine 数量
// queueSize：任务队列大小，满了之后 Submit 会阻塞
func NewTaskPool(workers, queueSize int) *TaskPool {
    pool := &TaskPool{
        tasks:   make(chan Task, queueSize),
        workers: workers,
    }

    // 启动指定数量的 worker goroutine
    for i := 0; i < workers; i++ {
        pool.wg.Add(1)
        go pool.worker(i)
    }

    return pool
}

// worker：不断从任务队列取任务执行
func (p *TaskPool) worker(id int) {
    defer p.wg.Done()
    for task := range p.tasks {
        // 执行任务
        task()
    }
}

// Submit 提交任务
// 如果任务队列满了，这里会阻塞（等待有空闲位置）
// 如果不想阻塞，可以用 select + default
func (p *TaskPool) Submit(task Task) {
    p.wg.Add(1)
    go func() {
        defer p.wg.Done()
        p.tasks <- task
    }()
}

// Shutdown 关闭任务池
func (p *TaskPool) Shutdown() {
    close(p.tasks) // 关闭任务队列，worker 的 range 循环会退出
    p.wg.Wait()    // 等待所有 worker 完成
}

func main() {
    // 创建一个有 3 个 worker 的任务池
    pool := NewTaskPool(3, 10)

    var taskWg sync.WaitGroup

    // 提交 10 个任务
    for i := 0; i < 10; i++ {
        taskWg.Add(1)
        taskID := i
        pool.Submit(func() {
            defer taskWg.Done()
            fmt.Printf("执行任务 %d\n", taskID)
            // 模拟耗时操作
            time.Sleep(100 * time.Millisecond)
            fmt.Printf("任务 %d 完成\n", taskID)
        })
    }

    taskWg.Wait()
    pool.Shutdown()
    fmt.Println("所有任务完成")
}

package main

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

func main() {
    // ==================== 泄露场景一：只发送不接收 ====================
    func() {
        ch := make(chan int)

        go func() {
            // 这个 goroutine 会永远阻塞
            // 因为没有人从 ch 接收数据
            ch <- 42
            fmt.Println("这行永远不会打印")
        }()

        // 函数返回后，ch 没有引用了，但 goroutine 还在阻塞
        // goroutine 泄露！
    }()

    // ==================== 泄露场景二：只接收不发送 ====================
    func() {
        ch := make(chan int)

        go func() {
            // 这个 goroutine 会永远阻塞
            // 因为没有人往 ch 发送数据
            val := <-ch
            fmt.Println("这行永远不会打印:", val)
        }()

        // goroutine 泄露！
    }()

    // ==================== 泄露场景三：读写 nil channel ====================
    func() {
        var ch chan int // nil channel

        go func() {
            // 往 nil channel 发送数据会永远阻塞
            ch <- 42
        }()

        // goroutine 泄露！
    }()

    // ==================== 正确做法：用 select + context 避免泄露 ====================
    func() {
        ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
        defer cancel()

        ch := make(chan int, 1)

        go func() {
            select {
            case ch <- 42:
                fmt.Println("发送成功")
            case <-ctx.Done():
                // 超时或取消，goroutine 正常退出，不会泄露
                fmt.Println("发送超时，退出 goroutine")
            }
        }()

        time.Sleep(200 * time.Millisecond)
    }()

    time.Sleep(500 * time.Millisecond)
    fmt.Printf("当前 goroutine 数量: %d\n", runtime.NumGoroutine())
}

package main

import "fmt"

// ==================== 内存逃逸问题 ====================
//
// Go 的内存分配有两个地方：
// - 栈（Stack）：速度快，不需要 GC，函数返回时自动回收
// - 堆（Heap）：需要 GC 管理，有性能开销
//
// 如果用 channel 发送指针类型的数据，必然会导致内存逃逸
// 因为编译器无法确定指针最终会被哪个 goroutine 接收
// 为了安全，只能分配到堆上

func main() {
    ch := make(chan *int, 1)

    // 这里的 val 会被分配到堆上（逃逸）
    // 因为它通过 channel 传递给了另一个（可能的）goroutine
    val := 42
    ch <- &val

    received := <-ch
    fmt.Println(*received) // 42
}

// ==================== 如何减少逃逸？ ====================
// 1. 尽量发送值类型而非指针类型
// 2. 如果必须发送指针，考虑使用 sync.Pool 复用对象
// 3. 不要过度优化，大部分场景下逃逸的性能影响可以忽略

// ==================== hchan（channel 的内部结构）====================

type hchan struct {
    // ==================== 数据存储 ====================
    // buf 是一个 ring buffer（环形缓冲区），用于存储缓冲数据
    // 只有带缓冲的 channel 才有 buf
    // 无缓冲 channel 的 buf 容量为 0
    buf      unsafe.Pointer // 指向环形缓冲区
    dataqsiz uint           // 环形缓冲区的大小

    // ==================== 等待队列 ====================
    // sendq：发送者等待队列（双向链表）
    //   当缓冲满了（或无缓冲且没接收者），发送者被挂到这里
    // recvq：接收者等待队列（双向链表）
    //   当缓冲空了（或无缓冲且没发送者），接收者被挂到这里
    sendq    waitq  // 等待发送的 goroutine 队列
    recvq    waitq  // 等待接收的 goroutine 队列

    // ==================== 其他字段 ====================
    closed   uint32 // channel 是否已关闭
    elemtype *_type // 元素类型
    lock     mutex  // 保护 hchan 的锁
}

// waitq 是一个双向链表
type waitq struct {
    first *sudog // 队头
    last  *sudog // 队尾
}

// ==================== 发送数据的步骤 ====================
//
// 步骤 1：如果是 nil channel → 直接阻塞（永远阻塞）
//   var ch chan int
//   ch <- 1  // 永远阻塞
//
// 步骤 2：如果有阻塞的接收者（recvq 不为空）
//   → 直接把数据交给接收者，不需要经过缓冲区
//   → 即使缓冲没满，也优先直接交付
//   → 唤醒接收者 goroutine
//
// 步骤 3：如果缓冲没满
//   → 把数据放入缓冲区（buf）
//   → 更新 sendx 索引（ring buffer 的写位置）
//
// 步骤 4：缓冲满了（或无缓冲且没接收者）
//   → 当前 goroutine 被包装成 sudog
//   → 加入 sendq 等待队列
//   → 调用 runtime 挂起 goroutine
//
// 步骤 5：被接收者唤醒
//   → 做一些清理工作
//   → 返回

// ==================== 关键细节 ====================
// 在步骤 2 中，即使缓冲没满，也会优先直接交给接收者
// 这是因为：如果有人等着接收，就没必要绕一圈走缓冲区

// ==================== 接收数据的步骤 ====================
//
// 步骤 1：如果是 nil channel → 直接阻塞
//
// 步骤 2：如果有阻塞的发送者（sendq 不为空）
//   → 直接从发送者手里拿数据
//   → 即使缓冲有数据，也优先拿发送者的
//   → 唤醒发送者 goroutine
//   → （如果是无缓冲 channel，这一步就是正常流程）
//
// 步骤 3：如果缓冲有数据
//   → 从缓冲区读取数据
//   → 更新 recvx 索引（ring buffer 的读位置）
//
// 步骤 4：缓冲空了（或无缓冲且没发送者）
//   → 当前 goroutine 被包装成 sudog
//   → 加入 recvq 等待队列
//   → 调用 runtime 挂起 goroutine
//
// 步骤 5：被发送者唤醒
//   → 做一些清理工作
//   → 返回

// ==================== 从已关闭的 channel 接收 ====================
// 从已关闭的 channel 接收数据不会 panic：
// - 如果缓冲区还有数据，继续返回数据
// - 如果缓冲区空了，返回零值，ok 返回 false
//   val, ok := <-closedCh  // val 是零值，ok 是 false

package main

import (
    "context"
    "fmt"
    "log"
    "net/http"
    "os"
    "os/signal"
    "sync"
    "syscall"
    "time"

    "golang.org/x/sync/errgroup"
)

// ==================== 1. 本地缓存（write-back 模式）====================

// LocalCache 本地缓存
// write-back 模式：数据先写到缓存，key 过期时才持久化到数据库
// 所以关闭时必须把缓存中的数据刷新到数据库
type LocalCache struct {
    mu   sync.RWMutex
    data map[string]string
}

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

func (c *LocalCache) Set(key, val string) {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.data[key] = val
}

// Flush 将缓存中的数据刷新到数据库
// 在优雅退出时调用
func (c *LocalCache) Flush() error {
    c.mu.RLock()
    defer c.mu.RUnlock()

    fmt.Printf("[Cache] 正在将 %d 条数据刷新到数据库...\n", len(c.data))
    // 模拟写入数据库
    for k, v := range c.data {
        fmt.Printf("[Cache] 写入数据库: %s = %s\n", k, v)
    }
    time.Sleep(500 * time.Millisecond) // 模拟耗时
    fmt.Println("[Cache] 刷新完成")
    return nil
}

// ==================== 2. 优雅退出的服务器管理器 ====================

// GracefulServer 优雅退出的服务器管理器
type GracefulServer struct {
    servers   []*http.Server  // 需要管理的服务器列表
    cache     *LocalCache     // 需要刷新的缓存
    callbacks []func() error  // 退出回调列表
}

func NewGracefulServer(cache *LocalCache) *GracefulServer {
    return &GracefulServer{
        cache: cache,
    }
}

// AddServer 添加服务器
func (g *GracefulServer) AddServer(addr string, handler http.Handler) {
    g.servers = append(g.servers, &http.Server{
        Addr:    addr,
        Handler: handler,
    })
}

// OnShutdown 注册退出回调
func (g *GracefulServer) OnShutdown(fn func() error) {
    g.callbacks = append(g.callbacks, fn)
}

// Start 启动所有服务器并监听退出信号
func (g *GracefulServer) Start() error {
    // ==================== 第一步：用 errgroup 启动所有服务器 ====================
    // errgroup + context 实现"一个启动失败，全部退出"
    eg, egCtx := errgroup.WithContext(context.Background())

    for _, srv := range g.servers {
        srv := srv // 避免闭包捕获问题
        eg.Go(func() error {
            // ListenAndServe 会一直阻塞，直到出错才返回
            err := srv.ListenAndServe()
            if err != nil && err != http.ErrServerClosed {
                // 只有非正常关闭的错误才返回
                return err
            }
            return nil
        })
    }

    // ==================== 第二步：监听系统信号 ====================
    // 在单独的 goroutine 中监听 SIGINT (Ctrl+C) 和 SIGTERM
    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)

    // 等待信号或服务器异常退出
    select {
    case sig := <-sigCh:
        fmt.Printf("\n[Shutdown] 收到信号: %v\n", sig)
    case <-egCtx.Done():
        // 某个服务器启动失败或异常退出
        fmt.Println("[Shutdown] 服务器异常退出")
    }

    // ==================== 第三步：优雅关闭服务器 ====================
    // 使用一个新的 context（脱离启动时的 context 控制）
    // 设置超时时间，防止关闭过程卡住
    shutdownCtx, shutdownCancel := context.WithTimeout(
        context.Background(),
        10*time.Second, // 给 10 秒时间完成优雅退出
    )
    defer shutdownCancel()

    // 使用 WaitGroup 等待所有服务器关闭完成
    var wg sync.WaitGroup
    for _, srv := range g.servers {
        wg.Add(1)
        srv := srv
        go func() {
            defer wg.Done()
            fmt.Printf("[Shutdown] 正在关闭服务器 %s...\n", srv.Addr)
            // Shutdown 会拒绝新请求，并等待已接收的请求处理完成
            if err := srv.Shutdown(shutdownCtx); err != nil {
                log.Printf("[Shutdown] 关闭服务器 %s 失败: %v\n", srv.Addr, err)
            } else {
                fmt.Printf("[Shutdown] 服务器 %s 已关闭\n", srv.Addr)
            }
        }()
    }
    wg.Wait()

    // ==================== 第四步：执行退出回调 ====================
    // 例如：将缓存数据刷新到数据库
    for _, cb := range g.callbacks {
        if err := cb(); err != nil {
            log.Printf("[Shutdown] 回调执行失败: %v\n", err)
        }
    }

    // 等待 errgroup 结束（此时所有服务器都已关闭，eg 也会返回）
    _ = eg.Wait()

    fmt.Println("[Shutdown] 优雅退出完成")
    return nil
}

// ==================== 3. main 函数 ====================

func main() {
    // 创建本地缓存
    cache := NewLocalCache()

    // 写入一些测试数据
    cache.Set("user:1", "张三")
    cache.Set("user:2", "李四")
    cache.Set("user:3", "王五")

    // 业务路由（8080 端口）
    businessMux := http.NewServeMux()
    businessMux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
        // 模拟业务处理
        time.Sleep(200 * time.Millisecond)
        w.Write([]byte("业务处理完成"))
    })
    businessMux.HandleFunc("/set", func(w http.ResponseWriter, r *http.Request) {
        key := r.URL.Query().Get("key")
        val := r.URL.Query().Get("val")
        cache.Set(key, val)
        w.Write([]byte("设置成功"))
    })

    // 管理路由（8081 端口）
    adminMux := http.NewServeMux()
    adminMux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
        w.Write([]byte("OK"))
    })

    // 创建优雅退出管理器
    gs := NewGracefulServer(cache)

    // 添加两个服务器
    gs.AddServer(":8080", businessMux)
    gs.AddServer(":8081", adminMux)

    // 注册退出回调：刷新缓存到数据库
    gs.OnShutdown(func() error {
        return cache.Flush()
    })

    // 启动服务器（会阻塞，直到收到退出信号）
    if err := gs.Start(); err != nil {
        log.Fatal(err)
    }
}