Go 并发编程

// 启动一个 goroutine
go func() {
    fmt.Println("Hello from goroutine")
}()

// 带参数的 goroutine
go worker(id, jobs, results)

// 常见：HTTP 服务为每个请求自动启动一个 goroutine
http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
    // 此处已经在独立的 goroutine 中
})

// 仅声明（nil channel）
var ch chan int

// 创建无缓冲 channel（同步通信）
ch := make(chan int)

// 创建有缓冲 channel（异步通信）
ch := make(chan int, 10)          // 容量 10

// 只读 / 只写 channel
func produce(out chan<- int) {    // 只能发送
    out <- 42
}
func consume(in <-chan int) {     // 只能接收
    v := <-in
}

ch := make(chan int, 2)

ch <- 1                           // 发送
ch <- 2

v := <-ch                         // 接收
v, ok := <-ch                     // 接收 + 检查是否已关闭

close(ch)                         // 关闭（由发送方关闭！）

// 遍历 channel（直到 close）
for v := range ch {
    fmt.Println(v)
}

// 无缓冲：发送方阻塞直到接收方就绪（同步）
ch := make(chan int)
go func() { ch <- 42 }()         // 阻塞，直到...
v := <-ch                        // ...接收方接收

// 有缓冲：发送不阻塞（缓冲未满时）
ch := make(chan int, 2)
ch <- 1                          // 不阻塞
ch <- 2                          // 不阻塞
// ch <- 3                       // 阻塞（缓冲已满）

// 限制函数只能发送或只能接收——编译期保证
func producer(out chan<- int) {
    out <- 42
    // <-out                       // 编译错误！
}

func consumer(in <-chan int) {
    v := <-in
    // in <- 42                    // 编译错误！
}

// 双向 → 单向：自动转换
ch := make(chan int)
go producer(ch)                   // chan int → chan<- int
go consumer(ch)                   // chan int → <-chan int

select {
case v := <-ch1:
    fmt.Println("from ch1:", v)
case ch2 <- value:
    fmt.Println("sent to ch2")
case v, ok := <-ch3:
    if !ok {
        fmt.Println("ch3 closed")
    }
case <-time.After(1 * time.Second):
    fmt.Println("timeout")
default:
    fmt.Println("nothing ready")
}

// 超时控制
select {
case res := <-ch:
    handle(res)
case <-time.After(3 * time.Second):
    fmt.Println("timeout")
}

// 非阻塞发送
select {
case ch <- value:
    fmt.Println("sent")
default:
    fmt.Println("channel full, dropping")
}

// 非阻塞接收
select {
case v := <-ch:
    fmt.Println("received:", v)
default:
    fmt.Println("nothing available")
}

// 退出信号
for {
    select {
    case v := <-dataCh:
        process(v)
    case <-done:
        return
    }
}

// Fan-in（合并多个 channel）
func fanIn(ch1, ch2 <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for ch1 != nil || ch2 != nil {
            select {
            case v, ok := <-ch1:
                if !ok { ch1 = nil; continue }
                out <- v
            case v, ok := <-ch2:
                if !ok { ch2 = nil; continue }
                out <- v
            }
        }
    }()
    return out
}

var mu sync.Mutex
var counter int

func increment() {
    mu.Lock()
    defer mu.Unlock()             // defer 模式（推荐）
    counter++                     // 临界区
}

var rw sync.RWMutex
var data map[string]string

func read(key string) string {
    rw.RLock()                    // 多个 goroutine 可同时持有读锁
    defer rw.RUnlock()
    return data[key]
}

func write(key, value string) {
    rw.Lock()                     // 写锁：独占，等待所有读锁释放
    defer rw.Unlock()
    data[key] = value
}

var wg sync.WaitGroup

for i := range 5 {                // 1.22+：for range n 直接迭代次数
    wg.Go(func() {                // 1.25+：自动 Add/Done，无需手动计数
        fmt.Println("worker", i)  // 1.22+：循环变量每轮独立，可直接捕获
    })
}

wg.Wait()                         // 等待全部任务完成
fmt.Println("all done")

var wg sync.WaitGroup

for i := range 5 {
    wg.Add(1)                     // 计数 +1（必须在 go 语句之前调用）
    go func(id int) {
        defer wg.Done()           // 计数 -1
        fmt.Println("worker", id)
    }(i)
}

wg.Wait()

// 改写前
for i := 0; i < 5; i++ {
    wg.Add(1)
    go func() {
        defer wg.Done()
        fmt.Println("legacy worker", i)
    }()
}

// 执行 go fix 后（实测产物）
for i := 0; i < 5; i++ {
    wg.Go(func() {
        fmt.Println("legacy worker", i)
    })
}

var once sync.Once
var config *Config

func GetConfig() *Config {
    once.Do(func() {
        config = loadConfig()     // 只会执行一次
    })
    return config
}

var m sync.Map

// 写入
m.Store("key", "value")

// 读取
v, ok := m.Load("key")

// 读取或写入
actual, loaded := m.LoadOrStore("key", "default")

// 删除
m.Delete("key")

// 遍历
m.Range(func(key, value any) bool {
    fmt.Println(key, value)
    return true                   // 继续遍历
})

var bufPool = sync.Pool{
    New: func() any {
        return new(bytes.Buffer)
    },
}

func process(data string) {
    buf := bufPool.Get().(*bytes.Buffer)
    defer bufPool.Put(buf)        // 用完后归还

    buf.Reset()
    buf.WriteString(data)
    // ...
}

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

// 等待方
mu.Lock()
for !condition() {
    cond.Wait()                   // 释放锁 + 等待信号 + 重新获取锁
}
mu.Unlock()

// 通知方
mu.Lock()
// 改变条件
cond.Signal()                     // 唤醒一个等待方
// 或 cond.Broadcast() — 唤醒所有等待方
mu.Unlock()

// 根 context（永不取消）
ctx := context.Background()       // 入口/初始化
ctx := context.TODO()             // 不确定该用哪个时（占位符）

// 派生 context
ctx, cancel := context.WithCancel(parentCtx)           // 手动取消
ctx, cancel := context.WithCancelCause(parentCtx)      // 手动取消 + 携带原因（1.20+）
ctx, cancel := context.WithTimeout(parentCtx, 5*time.Second)  // 超时
ctx, cancel := context.WithDeadline(parentCtx, time.Now().Add(5*time.Second))  // 截止时间
ctx := context.WithValue(parentCtx, key, value)         // 携带值

func doWork(ctx context.Context) error {
    for {
        select {
        case <-ctx.Done():        // 收到取消/超时信号
            return ctx.Err()      // context.Canceled or context.DeadlineExceeded
        default:
            // 继续工作...
        }
    }
}

// 超时控制
func fetchURL(ctx context.Context, url string) error {
    ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
    defer cancel()                // 及时释放资源（重要！）

    req, _ := http.NewRequestWithContext(ctx, "GET", url, nil)
    resp, err := http.DefaultClient.Do(req)
    // ...
}

// context 传值的最佳实践
type contextKey struct{}           // 未导出类型，保证唯一

func WithUserID(ctx context.Context, id string) context.Context {
    return context.WithValue(ctx, contextKey{}, id)
}

func UserIDFromContext(ctx context.Context) (string, bool) {
    id, ok := ctx.Value(contextKey{}).(string)
    return id, ok
}

package main

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

var ErrAuthExpired = errors.New("auth expired")

func main() {
    ctx, cancel := context.WithCancelCause(context.Background())

    go func() { // 某个后台探测器：带原因取消
        time.Sleep(10 * time.Millisecond)
        cancel(ErrAuthExpired)
    }()

    <-ctx.Done()
    fmt.Println("ctx.Err():", ctx.Err())
    fmt.Println("context.Cause(ctx):", context.Cause(ctx))
    fmt.Println("errors.Is(cause, ErrAuthExpired):",
        errors.Is(context.Cause(ctx), ErrAuthExpired))
}

func producer(out chan<- int) {
    defer close(out)
    for i := range 10 {            // 1.22+：for range n
        out <- i
    }
}

func consumer(in <-chan int) {
    for v := range in {
        fmt.Println(v)
    }
}

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

// Fan-Out：一个输入 → 多个 worker 处理
func fanOut(in <-chan int, workers int) []<-chan int {
    outputs := make([]<-chan int, workers)
    for i := range workers {
        out := make(chan int)
        go func(ch chan int) {
            defer close(ch)
            for v := range in {
                ch <- v * 2         // 并行处理
            }
        }(out)
        outputs[i] = out
    }
    return outputs
}

// 串联处理：gen → square → print
func gen(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            out <- n
        }
    }()
    return out
}

func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            out <- n * n
        }
    }()
    return out
}

// 组合
for n := range square(square(gen(2, 3))) {
    fmt.Println(n)                 // 16, 81
}

func workerPool(numWorkers int, jobs <-chan Job, results chan<- Result) {
    var wg sync.WaitGroup
    for range numWorkers {
        wg.Go(func() {             // 1.25+
            for job := range jobs {
                results <- process(job)
            }
        })
    }
    go func() {
        wg.Wait()
        close(results)             // 全部 worker 退出后才关闭结果通道
    }()
}

// 使用有缓冲 channel 作为信号量，控制并发度
func processAll(items []Item, maxConcurrency int) {
    sem := make(chan struct{}, maxConcurrency)
    var wg sync.WaitGroup

    for _, item := range items {   // 1.22+：循环变量直接捕获即可
        sem <- struct{}{}          // 获取令牌（满时在这里阻塞，实现节流）
        wg.Go(func() {             // 1.25+
            defer func() { <-sem }() // 释放令牌
            process(item)
        })
    }
    wg.Wait()
}

import "sync/atomic"

var counter int64

atomic.AddInt64(&counter, 1)        // 原子递增
v := atomic.LoadInt64(&counter)     // 原子读取
atomic.StoreInt64(&counter, 0)      // 原子写入
old := atomic.SwapInt64(&counter, 10)  // 原子交换

// CAS（Compare And Swap）
if atomic.CompareAndSwapInt64(&counter, oldValue, newValue) {
    // 交换成功
}

// Go 1.19+：原子类型（推荐）
var count atomic.Int64
count.Add(1)
v := count.Load()
count.Store(0)

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

// 并发执行多个任务，任一错误即取消其他
g, ctx := errgroup.WithContext(ctx)

for _, url := range urls {         // 1.22+：直接捕获 url，无需 url := url
    g.Go(func() error {
        return fetch(ctx, url)
    })
}

if err := g.Wait(); err != nil {
    log.Fatal(err)
}

func Test(t *testing.T, f func(*testing.T))   // 在气泡里运行测试函数
func Wait()                                    // 阻塞直到气泡内其他 goroutine 全部持久阻塞
func Sleep(d time.Duration)                    // 1.27 新增：Sleep 与 Wait 的结合

func TestVirtualClock(t *testing.T) {
	realStart := time.Now()
	synctest.Test(t, func(t *testing.T) {
		start := time.Now()
		fmt.Println("bubble 起点:", start.UTC())

		ping := make(chan time.Time, 1)
		stop := make(chan struct{})
		go func() {
			defer close(ping)
			for {
				ping <- time.Now()
				select {
				case <-stop:
					return
				default:
				}
				time.Sleep(time.Hour) // 气泡内的 Sleep，只推进虚拟时钟
			}
		}()

		for range 3 {
			<-ping
			fmt.Println("收到 ping，虚拟流逝:", time.Since(start))
		}
		close(stop)
		synctest.Wait() // 等其他 goroutine 全部收尾

		t0 := time.Now()
		synctest.Sleep(10 * time.Hour) // 1.27 新增
		fmt.Println("Sleep(10h) 返回，time.Since(t0):", time.Since(t0))
		fmt.Println("虚拟总流逝:", time.Since(start))
	})
	fmt.Println("真实总耗时:", time.Since(realStart)) // 气泡外才是真实时间
}

func TestWaitDetectsDeadlock(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		ch := make(chan int)
		go func() { ch <- 1 }() // 无人接收，durable 阻塞
		synctest.Wait()         // 气泡内没人能唤醒它 → 死锁
	})
}

runtime.NumGoroutine() // 当前 goroutine 总数，适合周期性打点/告警

package main

import (
	"fmt"
	"os"
	"time"

	_ "net/http/pprof"
	"runtime/pprof"
)

func main() {
	leakCh := make(chan int)          // 无缓冲，无人接收
	go func() { leakCh <- 1 }()       // 泄漏：永久阻塞在发送上
	time.Sleep(50 * time.Millisecond)

	p := pprof.Lookup("goroutineleak")
	fmt.Println(`pprof.Lookup("goroutineleak") != nil:`, p != nil)
	p.WriteTo(os.Stderr, 1)           // 文本形式输出
}

// 错误：goroutine 永远阻塞在 channel 发送上
ch := make(chan int)
go func() { ch <- 42 }()         // 没有接收方，泄漏！

// 正确方向：有缓冲兜底、接收方配对，或带上 ctx 退出分支

close(ch)
ch <- 1                           // panic: send on closed channel

// 错误：WaitGroup 是值类型，传参会复制
func worker(wg sync.WaitGroup) {  // 复制！Done() 不影响原始的
    defer wg.Done()
}
// 正确：传指针
func worker(wg *sync.WaitGroup) {
    defer wg.Done()
}

// Go 1.22 起这段就是正确的：每轮迭代是独立的 i
for i := 0; i < 3; i++ {
    go func() { fmt.Println(i) }() // 输出 0、1、2（顺序随机）
}

var mu sync.Mutex
mu2 := mu                        // 错误：不能复制（go vet copylocks 会报）
// Mutex 包含内部状态，复制会导致死锁或无效锁
// 结构体含 Mutex 时，应始终通过指针传递

ch := make(chan int)
go func() {
    for v := range ch { ... }    // 永久阻塞直到 ch 被 close
}()
// 如果忘记 close(ch)，该 goroutine 永远阻塞

mu.Lock()
updateCounter()                   // 如果 updateCounter 也 Lock → 死锁！
mu.Unlock()

// 正确：让 updateCounter 假设锁已持有（命名上区分，如 lockHeld 后缀），或拆分函数