HikariTish

1
// CSP 风格：通过 channel 传递任务
2
tasks := make(chan Task, 10)
3
go worker(tasks)
4
tasks <- Task{...}
5

6
// 共享内存风格：通过 Mutex 保护状态
7
var (
8
    mu    sync.Mutex
9
    state map[string]int
10
)

1
ch := make(chan int)        // dataqsiz = 0
2
go func() { ch <- 1 }()    // 发送方阻塞，直到接收方到来
3
v := <-ch                   // 接收方到达，完成"会合"

1
ch := make(chan int, 3)    // dataqsiz = 3
2
ch <- 1                     // 直接入队，不阻塞
3
ch <- 2
4
ch <- 3
5
ch <- 4                     // 缓冲区满，阻塞

1
var ch chan int             // ch == nil
2
ch <- 1                     // 永久阻塞
3
<-ch                        // 永久阻塞
4
close(ch)                   // panic: close of nil channel

1
hchan
2
┌─────────────────────────────────────────┐
3
│ qcount=2, dataqsiz=4                    │
4
│ buf ──────────┐                         │
5
│ sendx=3       │                         │
6
│ recvx=1       │                         │
7
│ closed=0      │                         │
8
│ recvq ──┐     │                         │
9
│ sendq   │     │                         │
10
└─────────┼─────┼─────────────────────────┘
11
          │     │
12
          │     ↓
13
          │  环形缓冲区 buf[4]
14
          │  ┌────┬────┬────┬────┐
15
          │  │    │ V₁ │ V₂ │    │
16
          │  └────┴────┴────┴────┘
17
          │   [0]  [1]  [2]  [3]
18
          │        ↑          ↑
19
          │       recvx     sendx
20
          ↓
21
     接收方等待队列 recvq
22
     ┌─────┐  ┌─────┐
23
     │sudog│→│sudog│→nil
24
     │  G₁ │  │  G₂ │
25
     └─────┘  └─────┘

1
// src/runtime/chan.go（简化版）
2

3
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
4
    // ① nil channel：阻塞模式下永久挂起
5
    if c == nil {
6
        if !block {
7
            return false
8
        }
9
        gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
10
        throw("unreachable")
11
    }
12

13
    lock(&c.lock)
14

15
    // ② 已关闭：触发 panic
16
    if c.closed != 0 {
17
        unlock(&c.lock)
18
        panic(plainError("send on closed channel"))
19
    }
20

21
    // ③ 有等待的接收方：直接交接数据，跳过缓冲区
22
    if sg := c.recvq.dequeue(); sg != nil {
23
        send(c, sg, ep, func() { unlock(&c.lock) }, 3)
24
        return true
25
    }
26

27
    // ④ 缓冲区未满：写入缓冲区
28
    if c.qcount < c.dataqsiz {
29
        qp := chanbuf(c, c.sendx)
30
        typedmemmove(c.elemtype, qp, ep)
31
        c.sendx++
32
        if c.sendx == c.dataqsiz {
33
            c.sendx = 0          // 环形回绕
34
        }
35
        c.qcount++
36
        unlock(&c.lock)
37
        return true
38
    }
39

40
    // ⑤ 缓冲区满：阻塞当前 goroutine
41
    gp := getg()
42
    mysg := acquireSudog()
43
    mysg.elem = ep
44
    mysg.g = gp
45
    mysg.c = c
46
    c.sendq.enqueue(mysg)
47
    gopark(chanparkcommit, unsafe.Pointer(&c.lock),
48
        waitReasonChanSend, traceEvGoBlockSend, 2)
49

50
    // 被唤醒后的清理工作
51
    if mysg.success {
52
        releaseSudog(mysg)
53
        return true
54
    }
55
    // success == false 说明被唤醒时 channel 已关闭
56
    panic(plainError("send on closed channel"))
57
}

1
chansend(ch, v)
2
    │
3
    ├── ch == nil ? ──→ 永久 park（或非阻塞返回 false）
4
    │
5
    ├── 加锁 c.lock
6
    │
7
    ├── ch.closed ? ──→ panic("send on closed channel")
8
    │
9
    ├── recvq 非空? ──→ send() 直接交接 → 唤醒接收方 → 返回
10
    │
11
    ├── 缓冲区未满? ──→ 写入 buf[sendx]，sendx++，qcount++ → 返回
12
    │
13
    └── 缓冲区已满
14
          │
15
          ├── 包装为 sudog，挂入 sendq
16
          ├── gopark()，G → _Gwaiting
17
          │
18
          └── (被唤醒后)
19
                ├── success → 数据已被接收方取走 → 返回
20
                └── !success → channel 被关闭 → panic

1
// src/runtime/chan.go（简化版）
2

3
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
4
    // ① nil channel：阻塞模式下永久挂起
5
    if c == nil {
6
        if !block {
7
            return
8
        }
9
        gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)
10
        throw("unreachable")
11
    }
12

13
    lock(&c.lock)
14

15
    // ② channel 已关闭且缓冲区为空：返回零值 + ok=false
16
    if c.closed != 0 && c.qcount == 0 {
17
        unlock(&c.lock)
18
        if ep != nil {
19
            typedmemclr(c.elemtype, ep)   // 写入零值
20
        }
21
        return true, false
22
    }
23

24
    // ③ 有等待的发送方
25
    if sg := c.sendq.dequeue(); sg != nil {
26
        recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
27
        return true, true
28
    }
29

30
    // ④ 缓冲区有数据：从 buf[recvx] 读取
31
    if c.qcount > 0 {
32
        qp := chanbuf(c, c.recvx)
33
        if ep != nil {
34
            typedmemmove(c.elemtype, ep, qp)
35
        }
36
        typedmemclr(c.elemtype, qp)
37
        c.recvx++
38
        if c.recvx == c.dataqsiz {
39
            c.recvx = 0          // 环形回绕
40
        }
41
        c.qcount--
42
        unlock(&c.lock)
43
        return true, true
44
    }
45

46
    // ⑤ 缓冲区空且无发送方：阻塞当前 goroutine
47
    gp := getg()
48
    mysg := acquireSudog()
49
    mysg.elem = ep
50
    mysg.g = gp
51
    mysg.c = c
52
    c.recvq.enqueue(mysg)
53
    gopark(chanparkcommit, unsafe.Pointer(&c.lock),
54
        waitReasonChanReceive, traceEvGoBlockRecv, 2)
55

56
    // 被唤醒后
57
    success := mysg.success
58
    releaseSudog(mysg)
59
    return true, success
60
}

1
chanrecv(ch)
2
    │
3
    ├── ch == nil ? ──→ 永久 park（或非阻塞返回）
4
    │
5
    ├── 加锁 c.lock
6
    │
7
    ├── ch.closed && qcount == 0 ? ──→ 返回 (零值, false)
8
    │
9
    ├── sendq 非空?
10
    │     ├── 无缓冲：直接从发送方拷贝
11
    │     └── 有缓冲：取缓冲区头，把发送方数据填入尾部
12
    │     → 唤醒发送方 → 返回 (值, true)
13
    │
14
    ├── 缓冲区有数据? ──→ 从 buf[recvx] 读取，recvx++，qcount-- → 返回
15
    │
16
    └── 缓冲区空且无发送方
17
          │
18
          ├── 包装为 sudog，挂入 recvq
19
          ├── gopark()，G → _Gwaiting
20
          │
21
          └── (被唤醒后)
22
                ├── success → 收到发送方数据 → 返回 (值, true)
23
                └── !success → channel 被关闭 → 返回 (零值, false)

1
ch := make(chan int, 2)
2
ch <- 1
3
ch <- 2
4
close(ch)
5

6
v1, ok1 := <-ch    // v1=1, ok1=true（缓冲区还有数据）
7
v2, ok2 := <-ch    // v2=2, ok2=true
8
v3, ok3 := <-ch    // v3=0, ok3=false（已关闭且缓冲区为空，返回零值）

1
select {
2
case v := <-ch1:
3
    // 从 ch1 接收
4
case ch2 <- x:
5
    // 向 ch2 发送
6
case <-time.After(time.Second):
7
    // 超时
8
default:
9
    // 所有 case 都未就绪
10
}

1
// 动态启用/禁用某个 case
2
func process(in <-chan int, done <-chan struct{}) {
3
    for {
4
        select {
5
        case v, ok := <-in:
6
            if !ok {
7
                in = nil           // 关闭后将其置 nil，禁用该 case
8
                continue
9
            }
10
            handle(v)
11
        case <-done:
12
            return
13
        }
14
    }
15
}

1
func gen(nums ...int) <-chan int {
2
    out := make(chan int)
3
    go func() {
4
        defer close(out)
5
        for _, n := range nums {
6
            out <- n
7
        }
8
    }()
9
    return out
10
}
11

12
func sq(in <-chan int) <-chan int {
13
    out := make(chan int)
14
    go func() {
15
        defer close(out)
16
        for n := range in {
17
            out <- n * n
18
        }
19
    }()
20
    return out
21
}
22

23
// 使用：gen → sq → main
24
for v := range sq(gen(1, 2, 3, 4)) {
25
    fmt.Println(v)   // 1, 4, 9, 16
26
}

1
// Fan-out：一个 channel 喂多个 worker
2
func fanOut(in <-chan int, n int) []<-chan int {
3
    outs := make([]<-chan int, n)
4
    for i := 0; i < n; i++ {
5
        outs[i] = sq(in)   // 多个 worker 共享同一个输入
6
    }
7
    return outs
8
}
9

10
// Fan-in：多个 channel 合并到一个
11
func fanIn(chans ...<-chan int) <-chan int {
12
    out := make(chan int)
13
    var wg sync.WaitGroup
14
    for _, c := range chans {
15
        wg.Add(1)
16
        go func(c <-chan int) {
17
            defer wg.Done()
18
            for v := range c {
19
                out <- v
20
            }
21
        }(c)
22
    }
23
    go func() {
24
        wg.Wait()
25
        close(out)
26
    }()
27
    return out
28
}

1
func worker(done <-chan struct{}) {
2
    for {
3
        select {
4
        case <-done:
5
            return            // 收到取消信号，退出
6
        default:
7
            // 正常工作
8
        }
9
    }
10
}
11

12
done := make(chan struct{})
13
go worker(done)
14
// ...
15
close(done)                   // 广播取消（所有接收方都会收到零值）

1
select {
2
case res := <-ch:
3
    handle(res)
4
case <-time.After(2 * time.Second):
5
    fmt.Println("timeout")
6
}

1
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
2
defer cancel()
3

4
select {
5
case res := <-ch:
6
    handle(res)
7
case <-ctx.Done():
8
    fmt.Println(ctx.Err())
9
}

1
// 用容量为 N 的 channel 限制并发数
2
sem := make(chan struct{}, 3)   // 最多 3 个并发
3

4
for _, task := range tasks {
5
    sem <- struct{}{}            // 获取"许可"
6
    go func(t Task) {
7
        defer func() { <-sem }() // 释放"许可"
8
        process(t)
9
    }(task)
10
}

1
原子操作 < Mutex < Channel
2
（开销从低到高）

1
var data int
2
done := make(chan struct{})
3

4
go func() {
5
    data = 42                    // ① 写入数据
6
    close(done)                  // ② 关闭 channel
7
}()
8

9
<-done                           // ③ 接收到零值
10
fmt.Println(data)                // ④ 一定能读到 42（① happens-before ④）

1
// ✓ 总是由发送方关闭
2
func producer(out chan<- int) {
3
    defer close(out)             // 确保关闭
4
    for i := 0; i < 10; i++ {
5
        out <- i
6
    }
7
}
8

9
// ✓ 用方向限定参数类型，让编译器帮你检查
10
func consumer(in <-chan int) {   // 只读
11
    for v := range in {
12
        fmt.Println(v)
13
    }
14
}
15

16
// ✓ 用 sync.Once 防止重复关闭
17
type SafeChan struct {
18
    ch   chan int
19
    once sync.Once
20
}
21

22
func (s *SafeChan) Close() {
23
    s.once.Do(func() { close(s.ch) })
24
}
25

26
// ✓ 使用 context 而非裸 done channel
27
func worker(ctx context.Context) {
28
    for {
29
        select {
30
        case <-ctx.Done():
31
            return
32
        case <-time.After(time.Second):
33
            doWork()
34
        }
35
    }
36
}

1
// 使用 runtime.NumGoroutine() 在测试前后对比
2
before := runtime.NumGoroutine()
3
runMyCode()
4
runtime.GC()
5
time.Sleep(100 * time.Millisecond)
6
after := runtime.NumGoroutine()
7
if after > before {
8
    t.Errorf("goroutine leak: %d -> %d", before, after)
9
}

1
v, ok := <-ch
2
if !ok {
3
    // channel 已关闭且缓冲区为空
4
}

1
┌─────────────────────────────────────────────────────────────┐
2
│                      Channel 总览                            │
3
├─────────────────────────────────────────────────────────────┤
4
│                                                             │
5
│              ┌──────────── hchan ────────────┐              │
6
│              │ qcount, dataqsiz              │              │
7
│              │ sendx, recvx, closed          │              │
8
│              │ lock                          │              │
9
│              └─────┬───────┬───────┬─────────┘              │
10
│                    │       │       │                        │
11
│         ┌──────────┘       │       └──────────┐             │
12
│         ↓                  ↓                  ↓             │
13
│    环形缓冲区          sendq (waitq)      recvq (waitq)     │
14
│    ┌─┬─┬─┬─┐         ┌──────┐            ┌──────┐           │
15
│    │V│V│ │ │         │sudog │→...        │sudog │→...       │
16
│    └─┴─┴─┴─┘         │  G   │            │  G   │           │
17
│     ↑   ↑            └──────┘            └──────┘           │
18
│   recvx sendx        阻塞的发送方         阻塞的接收方        │
19
│                                                             │
20
│   send 优化：sendq 中有接收方时，直接拷贝到接收方栈           │
21
│   close 广播：唤醒所有 recvq 和 sendq 中的 goroutine          │
22
│                                                             │
23
└─────────────────────────────────────────────────────────────┘