go语言Channel

by zhuzhenPosted on

4.1 什么是CSP？

CSP（Communicating Sequential Processes，通信顺序进程）并发编程模型，它的核心思想是：通过通信共享内存，而不是通过共享内存来通信。Go 语言的Goroutine 和 Channel机制，就是 CSP 的经典实现，具有以下特点：

避免共享内存：协程（Goroutine）不直接修改变量，而是通过 Channel 通信

天然同步：Channel 的发送/接收自带同步机制，无需手动加锁

易于组合：Channel 可以嵌套使用，构建复杂并发模式（如管道、超时控制）。

案例请见：【go-Channel并发模式-管道 Pipeline】、【go-Channel并发模式-超时控制】

4.2 Channel的底层实现原理是怎样的？

Channel的底层是一个名为hchan的结构体，核心包含几个关键组件：

环形缓冲区：有缓冲channel内部维护一个固定大小的环形队列，用buf指针指向缓冲区，sendx和recvx分别记录发送和接收的位置索引。这样设计能高效利用内存，避免数据搬移。

两个等待队列sendq和recvq：用来管理阻塞的goroutine。sendq存储因channel满而阻塞的发送者，recvq存储因channel空而阻塞的接收者。这些队列用双向链表实现，当条件满足时会唤醒对应的goroutine。

互斥锁：hchan内部有个mutex，所有的发送、接收操作都需要先获取锁，用来保证并发安全。虽然看起来可能影响性能，但Go的调度器做了优化，大多数情况下锁竞争并不激烈。

分析：

hchan定义如下：

type hchan struct {
        // 当前数组里现存元素个数
        qcount   uint
        // chan 底层循环数组的长度
        dataqsiz uint
        // 指向底层循环数组的指针
        // 环形数组首地址
        buf      unsafe.Pointer
        // chan 中元素大小
        elemsize uint16
        // chan 是否被关闭的标志
        closed   uint32
        // chan 中元素类型
        elemtype *_type // element type
        // 已发送元素在循环数组中的索引
        sendx    uint   // send index
        // 已接收元素在循环数组中的索引
        recvx    uint   // receive index
        // 等待接收的 goroutine 队列
        recvq    waitq  // list of recv waiters
        // 等待发送的 goroutine 队列
        sendq    waitq  // list of send waiters

        // 保护 hchan 中所有字段
        lock mutex
}

4.3 向channel发送数据的过程是怎样的？

向channel发送数据的整个过程都会在mutex保护下进行，保证并发安全。会经历几个关键步骤：

首先是检查是否有等待的接收者。如果recvq队列不为空，说明有goroutine在等待接收数据，这时会直接把数据传递给等待的接收者，跳过缓冲区，这是最高效的路径。同时会唤醒对应的goroutine继续执行。

// func chansend() 函数
if sg := c.recvq.dequeue(); sg != nil {
    // Found a waiting receiver. We pass the value we want to send
    // directly to the receiver, bypassing the channel buffer (if any).
    send(c, sg, ep, func() { unlock(&c.lock) }, 3)
    return true
}

如果没有等待接收者，就尝试写入缓冲区。检查缓冲区是否还有空间，如果qcount < dataqsiz，就把数据复制到buf[sendx]位置，然后更新sendx索引和qcount计数。这是无缓冲或缓冲区未满时的正常流径。

// func chansend() 函数
if c.qcount < c.dataqsiz {
    // Space is available in the channel buffer. Enqueue the element to send.
    qp := chanbuf(c, c.sendx)
    if raceenabled {
        racenotify(c, c.sendx, nil)
    }
    typedmemmove(c.elemtype, qp, ep) // 这里！直接拷贝数据进channel缓冲区
    c.sendx++
    if c.sendx == c.dataqsiz {
        c.sendx = 0
    }
    c.qcount++
    unlock(&c.lock)
    return true
}

当缓冲区满了就需要阻塞等待。创建一个sudog结构体包装当前goroutine和要发送的数据，加入到sendq等待队列中，然后调用gopark让当前goroutine进入阻塞状态，让出CPU给其他goroutine。

// func chansend() 函数
c.sendq.enqueue(mysg)
gopark(chanparkcommit, unsafe.Pointer(&c.lock), reason, traceBlockChanSend, 2)

被唤醒后继续执行。当有接收者从 channel 读取数据后，会从 sendq 取出等待的发送者，接收协程完成双方的数据拷贝，再唤醒发送 goroutine；被唤醒的 goroutine 只做清理收尾，然后继续执行。

后续来了接收协程执行 <-ch（chanrecv）,chanrecv 从 sendq 取出我们这个 sudog（mysg）,send () 内部执行 typedmemmove，完成数据拷贝（真正的发送动作），设置 mysg.success = true,=，调用 goready () 唤醒这个阻塞的发送 G。

因为阻塞发送的设计是：数据拷贝由对面接收方帮忙完成。 发送 goroutine 只是把自己打包 sudog，登记好待发送数据地址，排队休眠。等消费者过来消费的时候，由消费者的执行流去拷贝数据，然后唤醒生产者。生产者被叫醒的时候，活已经干完了，只需要收尾。

还有个特殊情况是向已关闭的channel发送数据会直接panic。这是Go语言的设计原则，防止向已关闭的通道写入数据。

分析：

package main

import (
    "fmt"
    "time"
)

func goroutineA(a <-chan int) {
    val := <-a
    fmt.Println("goroutine A received data: ", val)
    return
}

func goroutineB(b <-chan int) {
    val := <-b
    fmt.Println("goroutine B received data: ", val)
    return
}

func main() {
    ch := make(chan int)
    go goroutineA(ch)
    go goroutineB(ch)
    ch <- 3
    time.Sleep(time.Second)

    ch1 := make(chan struct{})
}

在第 17 行，主协程向 ch 发送了一个元素 3，来看下接下来会发生什么。

sender 发现 ch 的 recvq 里有 receiver 在等待着接收，就会出队一个 sudog，把 recvq 里 first 指针的 sudo “推举”出来了，并将其加入到 P 的可运行 goroutine 队列中。然后，sender 把发送元素拷贝到 sudog 的 elem 地址处，最后会调用 goready 将 G1 唤醒，状态变为 runnable。

当调度器光顾 G1 时，将 G1 变成 running 状态，执行 goroutineA 接下来的代码。G 表示其他可能有的 goroutine。

这里其实涉及到一个协程写另一个协程栈的操作。有两个 receiver 在 channel 的一边虎视眈眈地等着，这时 channel 另一边来了一个 sender 准备向 channel 发送数据，为了高效，用不着通过 channel 的 buf “中转”一次，直接从源地址把数据 copy 到目的地址就可以了，效率高啊！

上图是一个示意图，3 会被拷贝到 G1 栈上的某个位置，也就是 val 的地址处，保存在 elem 字段。

4.4 从Channel读取数据的过程是怎样的？

从channel读取数据也有几个关键步骤：

首先检查是否有等待的发送者。如果sendq队列不为空，说明有goroutine在等待发送数据。对于无缓冲channel，会直接从发送者那里接收数据；对于有缓冲channel，会先从缓冲区取数据，然后把等待发送者的数据放入缓冲区，这样保持FIFO顺序。

如果没有等待发送者，尝试从缓冲区读取。检查qcount > 0，如果缓冲区有数据，就从buf[recvx]位置取出数据，然后更新recvx索引和qcount计数。这是缓冲区有数据时的正常路径。

缓冲区为空时需要阻塞等待。创建sudog结构体包装当前goroutine，加入到recvq等待队列，调用gopark进入阻塞状态。当有发送者写入数据时会被唤醒继续执行。

从已关闭channel读取有特殊处理。如果channel已关闭且缓冲区为空，会返回零值和false标志；如果缓冲区还有数据，可以正常读取直到清空。这就是为什么v, ok := <-ch中的ok能判断channel状态的原因。

4.5 从一个已关闭Channel仍能读出数据吗？

从一个有缓冲的 channel 里读数据，当 channel 被关闭，依然能读出有效值。只有当返回的 ok 为 false 时，读出的数据才是无效的。

示例：

func main() {
        ch := make(chan int, 5)
        ch <- 18
        close(ch)
        x, ok := <-ch
        if ok {
                fmt.Println("received: ", x)
        }

        x, ok = <-ch
        if !ok {
                fmt.Println("channel closed, data invalid.")
        }
}

// 结果输出
// received:  18
// channel closed, data invalid.

先创建了一个有缓冲的 channel，向其发送一个元素，然后关闭此 channel。之后两次尝试从 channel 中读取数据，第一次仍然能正常读出值。第二次返回的 ok 为 false，说明 channel 已关闭，且通道里没有数据。

4.6 Channel在什么情况下会引起内存泄漏？

Channel引起内存泄漏最常见的是引起goroutine泄漏从而导致的间接内存泄漏，当goroutine阻塞在channel操作上永远无法退出时，goroutine本身和它引用的所有变量都无法被GC回收。比如一个goroutine在等待接收数据，但发送者已经退出了，这个接收者就会永远阻塞下去。或者select语句使用不当，在没有default分支的select中，如果所有case都无法执行，goroutine会永远阻塞。出现内存泄漏。

goroutine 被 gopark 阻塞时，GC 认为这个 G 还存活，goroutine 本身的 G 结构体、G 的栈内存保留，栈里面引用的所有对象，GC 都不能回收。这就是goroutine 泄漏，属于间接内存泄漏，不是 channel 结构体本身泄漏。

注意：只有通道没有 close 才会永久阻塞；如果发送方 close (ch)，接收方可以读到通道关闭，不会阻塞。

4.7 关闭Channel会产生异常吗？

试图重复关闭一个 channel、关闭 nil 值 channel，运行时 panic；关闭接收方向的 channel，编译错误，不属于运行 panic。

// 重复关闭 channel（运行时 panic）
ch := make(chan int)
close(ch)
close(ch) // panic: close of closed channel

// 关闭 nil channel（运行时 panic）
var ch chan int // 声明但没有make，ch=nil
close(ch)       //  panic: close of nil channel


// 关闭只有接收方向的 channel：编译报错，不是 panic！
func f(ch <-chan int) {
    close(ch) // 编译直接报错：invalid operation: cannot close receive-only channel
}

close 本质是给通道发一个关闭信号，这个动作属于写操作，接收方向的通道类型直接禁止写相关操作，编译器直接拦截，程序根本跑不到运行阶段。

4.8 往一个关闭的Channel写入数据会发生什么？

往已关闭的channel写入数据会直接panic。

向已关闭的channel发送数据时，runtime会检测到channel的closed标志位已经设置，立即抛出”send on closed channel”的panic。这个检查是在获取 channel 内部 mutex 锁之后执行的，所以不会有任何数据写入的尝试，直接就panic了。

ch := make(chan int, 2)
ch <- 10
close(ch) // 通道关闭，hchan.closed = 1
ch <- 20  // 往已关闭通道发送，panic：send on closed channel
fmt.Println("不会走到这里")
// 结果输出
// panic: send on closed channel

4.9 什么是select？

select是Go语言专门为channel操作设计的多路复用控制结构，类似于网络编程中的select系统调用。

核心作用是同时监听多个channel操作。当有多个channel都可能有数据收发时，select能够选择其中一个可执行的case进行操作，而不是按顺序逐个尝试。比如同时监听数据输入、超时信号、取消信号等。

4.10 select的执行机制是怎样的？

select的执行机制是随机选择。如果多个case同时满足条件，Go会随机选择一个执行，这避免了饥饿问题。如果没有case能执行就会执行default，如果没有default，当前goroutine会阻塞等待。

select {
case data := <-ch1:
    // 处理ch1的数据
case ch2 <- value:
    // 向ch2发送数据  
case <-timeout:
    // 超时处理
default:
    // 所有channel都不可用时执行
}

4.11 select的实现原理是怎样的？

Go语言实现select时，定义了一个数据结构scase表示每个case语句(包含default)。scase结构包含channel指针、操作类型等信息。select操作的整个过程通过selectgo函数在runtime层面实现。

Go运行时会将所有case进行随机排序，这是为了避免饥饿问题。然后执行两轮扫描策略：第一轮直接检查每个channel是否可读写，如果找到就绪的立即执行；如果都没就绪，第二轮就把当前goroutine加入到所有channel的发送或接收队列中，然后调用gopark进入睡眠状态，使当前goroutine让出CPU。

当某个channel变为可操作时，调度器会唤醒对应的goroutine，此时需要从其他channel的等待队列中清理掉这个goroutine，然后执行对应的case分支。

其核心原理是：case随机化 + 双重循环检测

分析：

scase结构定义：

type scase struct {
    c    *hchan   // channel指针
    elem unsafe.Pointer  // 数据元素指针，用于存放发送/接收的数据
    kind uint16   // case类型：caseNil、caseRecv、caseSend、caseDefault
    pc   uintptr  // 程序计数器，用于调试
    releasetime int64  // 释放时间，用于竞态检测
}

在默认的情况下，select 语句会在编译阶段经过如下过程的处理：

将所有的 case 转换成包含 Channel 以及类型等信息的 scase 结构体；

调用运行时函数 selectgo 获取被选择的scase 结构体索引，如果当前的 scase 是一个接收数据的操作，还会返回一个指示当前case 是否是接收的布尔值；

通过 for 循环生成一组 if 语句，在语句中判断自己是不是被选中的 case。

4.12 阻塞发送和非阻塞发送的区别是什么？

chansend 的参数 block 决定：当发送暂时无法完成的时候，goroutine 是等着，还是直接放弃

block = true：阻塞发送（普通 ch <- x）

如果现在没法发送（缓冲区满，并且没有等待的接收协程）。goroutine 调用 gopark 挂起，进入 sendq 队列等待，一直卡在这，直到可以发送。 成功发送后函数返回 true。

ch := make(chan int,1)
ch <- 1 //缓冲区满
ch <- 2 //阻塞发送，G挂起，等待有人取走1之后才能发送2

block = false：非阻塞发送（select + case ch <- x）

select {
case ch <- 2:
    fmt.Println("发送成功")
default:
    fmt.Println("发送失败，不等待")
}

如果当下没法发送，不挂起 G，直接返回 false，执行 default 分支，不会等待。