go中channel通信的底层实现

目录

• channel通信底层实现
• 同步通道(Sync channels)
• 异步通道(Async channels)
• 零大小元素的异步通道(Async channels with zero-sized elements)
• 同步通道内部
• 异步通道内部
• 零大小元素的异步通道
• close操作
• select操作

学习go语言的goroutine时，了解到goroutine的设计思想是线程之间通信不依赖共享内存，避免使用锁导致的死锁问题，starvation问题。锁在无竞争下确实轻量，但随着 goroutine/线程数增加，很容易出现竞争。channel 提供了一种更直观、更安全的通信方式，避免程序员手动管理锁。使用channel可以减少人为锁竞争和并发问题。

在学习过程中想深究一下channel的底层原理，所以写下这篇文章记录学习过程。

channel是go中主要的通信和同步核心原语，所以必须要做到快和可拓展，否则整个并发模型都会受影响。可拓展是说当 goroutine 数量增加时，channel 的性能不会迅速劣化，可以在大规模并发下保持稳定。所以go在设计channel时主要侧重于：

1. 让单线程（无竞争）的 channel 操作更快，比如只有一个 goroutine 在写，一个在读，没有其他竞争，最好可以降低到函数调用的开销。
2. 让有竞争的缓冲 channel（生产者/消费者模型）更快，比如多个 goroutine 同时写 / 读一个带缓冲的 channel。
3. 让非阻塞失败操作更快，当 goroutine 想做一个尝试读/写而发现条件不满足时，应该能 立刻返回，只做最少的检查，而不是浪费 CPU 在锁竞争或调度上。

goruntime内部把channel分为了三类

同步通道就是我们平常写的无缓冲channel，即make(chan int)。这类通道没有缓冲区，发送必须等接收方准备好，接收也必须等发送方准备好。发送方把数据直接交给接收方，不需要额外存储。这就相当于两个goroutine当场握手，适用在需要严格同步的场景。

异步通道就是带缓冲区的channel：make(chan int, N)。内部实现是一个环形队列。在异步通道中发送和接收可以解耦，发送方可以先把数据放进缓冲区，不用等接收方。接收方从缓冲区取数据，如果没有就阻塞。但是如果缓冲区满了发送方同样会阻塞，缓冲区空了接收方也会阻塞。在内部实现中即使接收方已经在等待，数据也不一定直接传给它，而是先进入缓冲区。等接收方竞争获取数据时，如果没抢到，就还得继续阻塞。

这个是channel的特殊情况：chan struct{}，这里的元素大小是0 字节，所以本质上就是一个计数信号量。不需要额外缓冲存储数据，常常用作并发控制、限流、任务完成通知。

sem := make(chan struct{}, 3) // 信号量，最多允许3个并发
// goroutine 获取令牌
sem <- struct{}{}
// goroutine 释放令牌
<-sem

同步channel的结构体包含以下数据：

struct Hchan {
    Lock;
    bool closed;
    SudoG* sendq;  // 等待发送的 goroutine 队列
    SudoG* recvq;  // 等待接收的 goroutine 队列
};

其中Lock保证并发安全，closed标记通道是否已经关闭，sendq / recvq是链表队列，保存阻塞的发送方/接收方。

使用同步通道的发送操作时，会先给互斥量加锁，保证后续只有自己处于临界区。如果当前没有接收方，就要把自己挂起，进入等待队列。如果有接收方在等待，那么就直接配对，把数据交给接收方，这样发送方不需要阻塞。

以下是syncchansend代码解读：

bool syncchansend(Hchan *c, T val, bool block) {
    if(c->closed)  // fast-path
    	panic(“closed”);
    if(!block && c->recvq == nil)  // fast-path
    	return false;
    lock(c);
    if(c->closed) {
        unlock(c);
        panic(“closed”);
    }
    if(sg = removewaiter(&c->recvq)) {
        // Have a blocked receiver, communicate with it.
        unlock(c);
        sg->val = val;
        sg->completed = true;
        unblock(sg->g);
        return true;
    }
    if(!block) {
        unlock(c);
        return false;
    }
    // Block and wait for a pair.
    sg->g = g;
    sg->val = val;
    addwaiter(&c->sendq, sg);
    unlock(c);
    block();
    if(!sg->completed)
    	panic(“closed”);  // unblocked by close
    // Unblocked by a recv.
    return true;
}

函数参数为c：目标通道，val：要发送的数据，block：是否允许阻塞（即普通 send vs 非阻塞 send）。第一步是检查是否可以fast failure，为什么要有“fast failure`呢？因为我们在编写如下代码时：

select {
case x := <-ch:    // 如果能取就取
    fmt.Println(x)
default:           // 如果取不到就直接失败
    fmt.Println("no data")
}

这里的<-ch不会像普通的同步channel一样一直阻塞在这里，而是如果通道里面有数据，就立即返回，如果通道里面没数据，不会阻塞，直接走到default分支。如果没有fast failure的话，此时goruntime还会傻傻地去加锁 -> 检查队列 -> 发现没数据 -> 解锁 -> 返回失败，其实完全没有必要，因为我们只需要检查队列是否为空。上面源码的逻辑就是：1. 如果通道已经关闭，立即报错，不需要加锁。2. 非阻塞发送而且当前没有接收方在等着，就直接返回失败，不阻塞，也不用加锁。

lock(c);
if (c->closed) {
    unlock(c);
    panic("closed");
}

之后的逻辑就是可以发送的情况，所以我们需要加锁保护临界区，加锁之后还需要检查通道是否关闭，避免之前在执行完判断之后，获取锁之前的某一个时间点通道被关闭，不检查的话就可能会对关闭的channel进行操作，向关闭的channel执行写操作的话就会直接panic。

if(sg = removewaiter(&c->recvq)) {
    // Have a blocked receiver, communicate with it.
    unlock(c);
    sg->val = val;
    sg->completed = true;
    unblock(sg->g);
    return true;
}

这段逻辑就是找到一个已经被阻塞的接收者，如果有 goroutine 已经在 recvq 等待，那么直接配对。配对完之后就直接将要发送的数据交给接收方，发送方不阻塞，接收方也被唤醒，通信完成。

但是如果没有找到已经被阻塞的接收者，就会执行下面的逻辑：

if (!block) {
    unlock(c);
    return false;
}

是非阻塞 send的话，并且没人接收，就立即返回失败。

// Block and wait for a pair.
sg->g = g;
sg->val = val;
addwaiter(&c->sendq, sg);
unlock(c);
block();

这里就是没有接收方但是允许阻塞的情况，这里我们解释一下sg是什么。

这里的 sg 指的是sudog类型的指针。

struct sudog {
    g* g;          // 等待的 goroutine
    element val;   // 要发送或接收的值
    sudog* next;   // 链表指针，用来挂到 sendq/recvq
    sudog* prev;
    ...
    bool completed;
};

• val → 发送方要传的数据，或者接收方接收的数据存放位置。
• next/prev → 把 sudog 串起来，形成 channel 的 sendq 和 recvq 队列。
• completed → 表示这个 goroutine 的操作是否已经完成。
• g* g → 指向一个正在等待的 goroutine。

在上述channel代码中sg->g = g;表示把当前 goroutine 记录到 sudog 里，sg->val = val;把要发送的值放到 sudog 里，addwaiter(&c->sendq, sg);把 sudog 加入 channel 的发送等待队列。block();把发送方挂起，等到有接收方时就可以被唤醒。这里的 sg 是一个 sudog 实例，它代表 “当前这个 goroutine 正在等待往 channel 里发送 val”。

当有接收方出现时，runtime 会从 sendq 拿一个等待的接收者 sudog，然后把这个 val 交给它，并唤醒 sg->g（这个 goroutine）。

if(!sg->completed)
	panic(“closed”);  // unblocked by close
// Unblocked by a recv.
return true;

被唤醒时要区分，如果是recv 唤醒→ 通信成功。如果是close 唤醒→ 抛 panic。

上面就是同步channel发送数据的细节，接收操作也是类似，都是先尝试快速路径，没法完成就进入慢路径阻塞。核心点都是sudog队列，goroutine 阻塞唤醒和数据hand-off，区别在于send 队列挂 send，recv 队列挂 recv。这里给出gpt生成的类似代码作为参考：

// 阻塞/非阻塞接收
bool synchchanrecv(Hchan *c, T* out, bool block) {
    // fast-path: channel 已关闭且队列为空
    if (c->closed && c->sendq == nullptr) {
        *out = zero_value();  // 返回零值
        return false;         // ok = false
    }
    // fast-path: 非阻塞且没有发送方
    if (!block && c->sendq == nullptr) {
        return false;
    }
    lock(c);  // 进入慢路径，保护 sendq/recvq
    if (c->closed && c->sendq == nullptr) {
        unlock(c);
        *out = zero_value();
        return false;
    }
    // 检查是否有等待的发送方
    sudog* sg;
    if ((sg = removewaiter(&c->sendq)) != nullptr) {
        unlock(c);
        // 直接从发送方取数据
        *out = sg->val;
        sg->completed = true;
        unblock(sg->g);  // 唤醒发送方
        return true;
    }
    // 没有发送方
    if (!block) {
        unlock(c);
        return false;  // 非阻塞失败
    }
    // 阻塞自己，加入 recvq
    sg = new_sudog();
    sg->g = g;        // 当前 goroutine
    sg->val_ptr = out; // 接收的数据存储地址
    addwaiter(&c->recvq, sg);
    unlock(c);
    block();  // 挂起当前 goroutine，等待发送方唤醒

    // 被唤醒后判断原因
    if (!sg->completed) {
        // 被 channel close 唤醒
        *out = zero_value();
        return false;
    }
    // 被发送方唤醒，接收成功
    return true;
}

异步通道就是带缓冲区的channel，在进行发送或者接收操作时，如果只是简单地在环形缓冲区中放入或者取出数据，那么goruntime就不需要操作等待队列，对于环形缓冲区goruntime采用原子指令读写数据，此时是无锁的，效率很高。但是一旦缓冲区满了时发送方需要阻塞，或者缓冲区空了时接收方需要阻塞，这时需要操作等待队列，就必须加锁，因为等待队列需要受互斥锁保护，避免并发修改出错。

这里同样提供了快速失败路径，对于非阻塞失败操作（例如尝试 send 到一个已满的通道，或者尝试 recv 从一个空通道），直接检测失败并返回，不进入复杂的加锁逻辑。

下面是异步通道中的核心数据结构：

struct Hchan {
    uint32 cap;   // channel capacity
    Elem*  buf;   // ring buffer of size cap

    // send and receive positions,
    // low 32 bits represent position in the buffer,
    // high 32 bits represent the current “lap” over the ring buffer
    uint64 sendx;
    uint64 recvx;
};

• cap表示通道的容量，也就是缓冲区最多可以存放多少元素。make(chan int, N)中的N就是cap。
• buf指向一个环形缓冲区，大小就是cap，用来存放用户真正写入的数据。
• sendx/recvx分别表示发送位置和接收位置。两个都是64位数，使用时需要分为高32位和低32位看。低32位表示在缓冲区中的下标位置，高32位表示当前是第几圈（lap，绕缓冲区的次数）。这样设计可以区分同一个下标位置上的数据，属于第几次写/读。避免了“ABA问题”（同一个位置被覆盖后，无法判断是不是旧数据）。

struct Elem {
    // current lap,
    // the element is ready for writing on laps 0, 2, 4, ...
    // for reading -- on laps 1, 3, 5, ...
    uint32 lap;
    T      val;  // user data
};

• lap是每个缓冲区元素内额外保存的一个圈数标记。如果是偶数圈的话说明位置是空的，可以写；奇数圈说明这个位置有数据可以读。这就保证了：生产者（send）和消费者（recv）在竞争时，能明确知道这个槽位当前是可写还是 可读，不会混淆。
• val就是用户写入的数据。

如果是非阻塞的异步通道的话，多个发送者 goroutine可能同时往 channel 里写数据。它们需要争抢一个“写入位置”，这是通过对sendx做CAS操作来完成的，谁 CAS 成功，谁就获得了在 sendx 指向的位置写入数据的权利。失败的发送者会重试，直到抢到一个空槽位。

发送方写入数据以后，要让接收方知道这个槽位已经写满。这不是通过锁，而是通过每个元素的 lap 变量来完成：如果 lap 是偶数 → 表示该槽位目前是 空的，只能写。如果 lap 是奇数 → 表示该槽位目前是 满的，可以读。发送者把数据写进去时，把 lap 改成奇数，通知接收者“数据可读”。接收者读完数据后，把 lap 改成偶数，通知发送者“槽位可写”。

接下来我们看看异步通道非阻塞发送的核心代码：

bool asyncchansend_nonblock(Hchan* c, T val) {
    uint32 pos, lap, elap;
    uint64 x, newx;
    Elem *e;

    for(;;) {
        x = atomicload64(&c->sendx);
        pos = (uint32)x;
        lap = (uint32)(x >> 32);
        e = &c->buf[pos];
        elap = atomicload32(&e->lap);
        if(lap == elap) {
            // The element is ready for writing on this lap.
            // Try to claim the right to write to this element.
            if(pos + 1 < c->cap)
            	newx = x + 1;  // just increase the pos
            else
            	newx = (uint64)(lap + 2) << 32;
            if(!cas64(&c->sendx, x, newx))
            	continue;  // lose the race, retry
            // We own the element, do non-atomic write.
            e->val = val;
            // Make the element available for reading.
            atomicstore32(&e->lap, elap + 1);
            return true;
        } else if((int32)(lap - elap) > 0) {
            // The element is not yet read on the previous lap,
            // the chan is full.
            return false;
        } else {
            // The element has already been written on this lap,
            // this means that c->sendx has been changed as well,
            // retry.
        }
    }
}

x = atomicload64(&c->sendx);
pos = (uint32)x;
lap = (uint32)(x >> 32);
e = &c->buf[pos];
elap = atomicload32(&e->lap);

首先原子读取当前发送位置，计算出写位置在环形缓冲区中的下标和当前圈数，原子读取 e->lap，因为其他线程（发送者/接收者）会修改它。

if(lap == elap) {
    // The element is ready for writing on this lap.
    // Try to claim the right to write to this element.
    if(pos + 1 < c->cap)
    	newx = x + 1;  // just increase the pos
    else
    	newx = (uint64)(lap + 2) << 32;
    if(!cas64(&c->sendx, x, newx))
    	continue;  // lose the race, retry
    // We own the element, do non-atomic write.
    e->val = val;
    // Make the element available for reading.
    atomicstore32(&e->lap, elap + 1);
    return true;
}

如果当前sendx的lap和缓冲区元素的lap一致的话，说明这个槽位正处于“可写状态”。因为一开始是0，如果当前sendx的lap也是0的话表示当前槽位为空，可以写。所以发送者就会尝试用CAS抢这个位置写入数据。

if (pos + 1 < c->cap)说明当前不是缓冲区的最后一个槽位，下一个位置仍然在环形缓冲区内部，所以只需要把 pos 增加 1。else分支就是说明已经写到环形缓冲区的尾部，再往前就需要回到开头。这时候不能仅仅把 pos 设为 0，还要把 lap 增加 2，来交替读写状态，进入新的写循环。

if(!cas64(&c->sendx, x, newx))
	continue;  // lose the race, retry
// We own the element, do non-atomic write.
e->val = val;
// Make the element available for reading.
atomicstore32(&e->lap, elap + 1);
return true;

这段逻辑就是无锁操作的关键部分，CAS（Compare-And-Swap）尝试把 sendx 从 x 更新到 newx。如果成功 → 当前线程赢得了这个插槽的所有权。如果失败 → 说明别的线程同时在竞争同一个位置，这个线程就要 continue 重新尝试。如果走到了后面的e->val = val;说明该线程已经确认自己“独占”了这个队列位置，可以安全地写入数据。注意这里用的是普通赋值（非原子），因为写入数据的唯一写入者已经通过 CAS 抢占成功，不存在写冲突。之后atomicstore32(&e->lap, elap + 1);是发布这个元素，让消费者能看到它已经可用了。

else if((int32)(lap - elap) > 0) {
    // The element is not yet read on the previous lap,
    // the chan is full.
    return false;
}

lap - elap > 0 意味着：这个槽位中的旧数据还没有被读走，这个槽位在上一轮循环里写过数据，但还没被读走 → 队列满了，不能再写。由于是非阻塞的，所以直接返回false。

else {
    // The element has already been written on this lap,
    // this means that c->sendx has been changed as well,
    // retry.
}

如果走到当前这个else分支，说明当前lap < elap，也就是说这个elap被其他的写者更新了，导致第一次看到的lap和缓冲区元素的elap不一样，这就体现了多个发送者竞争环形缓冲区时失败的情况，此时我们不能再往这个槽写，不然会覆盖别人的数据，所以只可以重新读取最新的sendx并再试一次。

上述就是异步通道非阻塞读的流程，接收（Recv）操作与发送（Send）操作完全对称，唯一的区别是接收操作从 lap = 1 开始，并且是读取元素而不是写入。

接下来是异步阻塞通道的发送和接收操作，通道结构会被扩展，增加一个互斥锁（mutex）以及发送/接收的等待队列（send/recv waiter queues）。加等待队列是为了实现阻塞语义，缓冲区满时，send 必须阻塞，直到有接收者取走数据；缓冲区空时，recv 必须阻塞，直到有发送者写入数据。

struct Hchan {
    …
    Lock;
    SudoG* sendq;
    SudoG* recvq;
};

Lock是互斥锁，用来保护channel内部的共享状态，比如说等待队列的入队或者出队。sendq/recvq就是发送方和接收方的等待队列，里面存放等待发送或者接收数据的goroutine。

异步队列在阻塞发送时，goroutine会先尝试进行一次非阻塞发送。如果成功，他会检查是否有接收方在等待，如果有，就唤醒其中一个。如果非阻塞发送失败（通道已满），它就会加锁，将自己加入发送等待队列，然后再次检查缓冲区是否仍然是满的。如果缓冲区仍然满，goroutine 就会阻塞；如果缓冲区已经不满，goroutine 会把自己从等待队列移除，解锁，然后重试。再次检测是为了防止就在它准备阻塞的那一瞬间，另一个 goroutine B 可能正好执行了一个 recv，使得缓冲区变得“不满”。

阻塞接收的过程完全相同，只不过把“send”换成“recv”，把“recv”换成“send”。

阻塞算法的主要难点是要确保不会发生死锁（比如发送方在一个非满的通道上无限期阻塞，或者接收方在一个非空的通道上无限期阻塞）。通过这种“检查 → 入队 → 再检查”的方式，我们能够保证以下情况之一一定会发生：

1. 发送方看到有接收方在等待并唤醒它；
2. 接收方看到缓冲区里有元素并消费它；
3. 1 和 2 同时发生（这种情况下通过互斥锁来解决竞争）；

但绝不会出现这种情况：发送方没看到接收方在等待，接收方也没看到缓冲区有元素，然后两边都无限期阻塞。

接下来看核心代码实现：

void asyncchansend(Hchan* c, T val) {
    for(;;) {
        if(asyncchansend_nonblock(c, val)) {
            // Send succeeded, see if we need to unblock a receiver.
            if(c->recvq != nil) {
                lock(c);
                sg = removewaiter(&c->recvq);
                unlock(c);
                if(sg != nil)
                	unblock(sg->g);
            }
            return;
        } else {
            // The channel is full.
            lock(c);
            sg->g = g;
            addwaiter(&c->sendq, sg);
            if(notfull(c)) {
                removewaiter(&c->sendq, sg);
                unlock(c);
                continue;
            }
            unlock(c);
            block();
            // Retry send.
        }
    }
}

首先调用 asyncchansend_nonblock 尝试直接写入数据，如果成功，说明缓冲区有空间或者直接匹配到了等待的接收者。如果通道里有接收者在等待 (c->recvq != nil)，那么我们应该唤醒接收者避免长时间阻塞，所以我们加锁，安全地从recvq队列里面取出一个等待接收的 goroutine（removewaiter），再解锁，如果真的取到一个等待者，就唤醒它。如果没有在等待的接收者但是非阻塞还是写入成功了，说明写入缓冲区中，这时直接返回就好。

// The channel is full.
lock(c);
sg->g = g;
addwaiter(&c->sendq, sg);
if(notfull(c)) {
    removewaiter(&c->sendq, sg);
    unlock(c);
    continue;
}
unlock(c);
block();
// Retry send.

else分支下是非阻塞发送失败的情况，根据上面非阻塞发送的代码可知，缓冲区已经满了，当前goroutine不能继续发送，就需要将自己添加到sendq发送等待队列。之后notfull(c)再做了一次检查，因为在刚刚排队的过程中，可能已经有接收者拿走了数据，缓冲区不再满。如果发现通道现在不满了，就把自己从sendq发送等待队列中移除，回到循环开头 (continue)，再试一次发送。但是如果如果通道依然满，那就调用block挂起当前 goroutine，等待别人来唤醒。被唤醒之后，继续循环，重新尝试发送。

最后我们介绍零大小元素的异步通道内部实现，零大小的异步 channel 在整体上和非零大小的异步 channel 类似：

• 在非阻塞情况下，操作是无锁的。
• 等待队列仍然由互斥锁保护。
• 非阻塞失败的操作会快速返回。

不同之处有：

• Hchan 里只有一个计数器，而不是发送/接收位置和环形缓冲区；这个计数器表示 channel 中的元素数量。
• 非阻塞的发送/接收通过 CAS 循环来更新计数器。
• 是否“满”或“空”的判断仅仅是检查计数器的值。

对于channel的close操作，我们调用close时，要先加锁，然后设置 closed 标志，接着唤醒所有的等待者。异步的 send/recv 操作在进入阻塞之前都会检查 closed 标志。这样就能保证与异步 send/recv 阻塞时相同的语义。要么1.close看到了有等待着，2.等待者看到了closed标志被设置，3. 1和2同时发生时通过互斥锁解决。

select 操作不会一次性锁住所有相关的 channel，而是对每个 channel 分别进行细粒度的操作。select操作包含以下四个阶段：

1. 打乱所有相关的 channel 顺序，以提供伪随机的保证（后续所有阶段都基于这个打乱后的列表）。
2. 逐个检查所有 channel，看看是否有可以立即通信的（比如缓冲不空可以读，缓冲不满可以写，或者有 goroutine 在对面排队）。如果有，就执行通信并退出。这样可以让不会阻塞的 select 更快、更具扩展性，因为它不需要排序和一次性锁住所有互斥量。而且如果第一个 channel 已经就绪了，select 甚至不需要检查后续的 channel。
3. 为在所有 channel 上阻塞做准备，要先把自己挂到所有相关 channel 的等待队列里（sendq/recvq），这样任何一个 channel 一旦 ready，就能唤醒这个 select。
4. 阻塞。当某个 channel ready 时，会把你唤醒。被唤醒之后，并不是直接执行，而是回到阶段 1 再检查一次，确认到底是哪个 channel ready 了。

对于阶段3：为在所有 channel 上阻塞做准备。它的过程与异步 send/recv 的阻塞方式几乎相同，首先锁住 channel 的互斥量，把自己加入 send/recv 的等待队列，之后再次检查该 channel 是否已经可以通信，如果已经可以通信，那么就把自己从所有等待队列中移除，并回到阶段 2；如果仍然不行，就继续处理下一个 channel。如果此时 channel 已经可以通信，就要把自己从所有等待队列中移除，并回到阶段 2。

但是有一个问题，select 可能同时在多个 channel 的队列里，必须确保只被唤醒一次。即select监听多个管道，有多个管道几乎同时来数据，必须确保只被唤醒一次。我们采取如下解决方式：每个 select 操作都有一个全局状态字 sg。当别的 goroutine 尝试唤醒它时，会做：CAS(statep, nil, sg)，如果成功：说明这个 goroutine 赢得了“唤醒权利”，它负责完成通信并唤醒 select。如果失败：说明另一个 goroutine 已经抢先唤醒了，那么当前 goroutine 就忽略这个 waiter。这样可以避免同步 channel 多方同时匹配到一个 select，保证只发生一次通信。

下面来看select操作的核心代码：

Scase *select(Select *sel) {
    // Phase 1.
    randomize channel order;
    for(;;) {
        // Phase 2.
        foreach(Scase *cas in sel) {
            if(chansend/recv_nonblock(cas->c, ...))
            	return cas;
        }
        // Phase 3.
        selectstate = nil;
        foreach(Scase *cas in sel) {
            lock(cas->c);
            cas->sg->g = g;
            cas->sg->selectstatep = &selectstate;
            addwaiter(&cas->c->sendq/recvq, cas->sg);
            if(isready(cas->c)) {
                unlock(c);
                goto ready;
            }
            unlock(cas->c);
        }
        // Phase 4.
        block();
        ready:
        CAS(&selectstate, nil, 1);
        foreach(Scase *cas in sel) {
            lock(cas->c);
            removewaiter(&cas->c->sendq/recvq, cas->sg);
            unlock(cas->c);
        }
        // If we were unblocked by a sync chan operation,
        // the communication has completed.
        if(selectstate > 1)
        return selectstate;  // denotes the completed case
    }
}

在进入select时就要将所有的channel排序，避免其受程序编写时顺序的干扰，更加公平，保证当多个 case 同时就绪时不会偏向前面的 case，之后的所有循环都使用这个打乱后的顺序。之后就是阶段2的代码，对每个 case 做非阻塞尝试(对无缓冲 channel 意味着有对端在等；对有缓冲的 channel 意味着缓冲非空或非满；对关闭 channel 特殊处理)。若任一 chansend/recv_nonblock 成功，则马上完成该通信并返回选中的 Scase。这避免了不必要的加锁和入队，是常见的低延迟路径。

在接下来的阶段三中，selectstate 是一个共享/可被外部 CAS 的状态字，为nil时表示尚未有人取得“胜利权”。之后遍历select中所有的channel，先lock(cas->c)，这是在对某个 channel 的内部队列操作时要持有该 channel 的 mutex，保证对队列的并发修改安全。

cas->sg->g = g;
cas->sg->selectstatep = &selectstate;

这里的cas 表示 select 中的某个 case，cas->sg 是这个 case 对应的 sudog（synchronous goroutine 的缩写）结构。这是 Go runtime 内部用来把 goroutine 挂到 channel 等待队列里的一个小对象。这第一句的意思是：把当前 goroutine g 存进 sudog 里，当别的 goroutine 往这个 channel 里发送/接收时，就可以通过 sudog->g 找到应该被唤醒的 goroutine。后面这句就是保证多个 channel 只会有一个真的成功唤醒 select。因为一个 select 可能同时往多个 channel 的等待队列里加了 sudog。假设有两个 channel 几乎同时就绪，两个 goroutine 都想来唤醒你。这里用了一个全局状态字 selectstate，所有 case 都共享它。在尝试唤醒你时，唤醒方会执行 CAS(selectstatep, nil, sg)：如果成功，说明自己“抢到”了唤醒权，可以继续通信。如果失败，说明别人已经先唤醒了你，自己就放弃。这样就避免了 “被两个 channel 同时唤醒” 的问题。

addwaiter(&cas->c->sendq/recvq, cas->sg);
if(isready(cas->c)) {
    unlock(c);
    goto ready;
}
unlock(cas->c);

第一句把当前 case 对应的 sudog (cas->sg) 挂到 channel 的等待队列里，这样别的 goroutine 在操作这个 channel 时，就能找到你，完成通信并唤醒你。然后检查通道是否立即可用，即使你刚刚把自己挂到了队列里，也要马上再检查一次：因为在你挂进去的这个瞬间，可能已经有别的 goroutine 进入 channel 操作，把它变成 ready 了。如果不检查，就可能白白阻塞自己，而其实通信机会已经存在。

如果 isready 发现 channel 已经可通信：立刻释放锁 unlock(c)，跳转到 ready 标签进入唤醒清理逻辑。但是如果 isready 没通过，那说明这个 channel 真的还不可通信。那么就正常解锁，继续去处理下一个 case。

for循环之外的这个block就是挂起当前goroutine，因为当前goroutine调用的select是阻塞的，让当前 goroutine 停止运行并交给调度器调度，直到被其他 goroutine 唤醒。保证了 channel 的阻塞语义：当条件不满足时，goroutine 就会真正“停住”，不会白白占用 CPU。

ready:
CAS(&selectstate, nil, 1);
foreach(Scase *cas in sel) {
    lock(cas->c);
    removewaiter(&cas->c->sendq/recvq, cas->sg);
    unlock(cas->c);
}

这段代码就是讲select被唤醒之后的清理和状态处理阶段，首先尝试通过 CAS 将 selectstate 从 nil 设置为 1，如果没有其他 goroutine 先唤醒并完成通信（即 selectstate == nil），select 自己就“认领”通信权。如果 CAS 失败（selectstate 已经被写成其他值，例如某个外部同步 channel 的 sg），说明通信已经被其他 goroutine 完成，select 自己不再做通信。

之后的for循环就是清理等待队列，把 select 自己在其它 channel 的等待节点从队列中移除。因为select 可能同时挂在多个 channel 的队列中，只要某个 channel 已经唤醒了 select，其他 channel 队列里的节点就不再需要。

到此这篇关于go中channel通信的底层实现的文章就介绍到这了,更多相关go channel通信内容请搜索风君子博客以前的文章或继续浏览下面的相关文章希望大家以后多多支持风君子博客！

您可能感兴趣的文章:

• golang channel多协程通信常用方法底层原理全面解析
• Go通道channel通过通信共享内存
• django使用channels实现通信的示例