第10章协程间的数据传递

课程目录第10章协程间的数据传递

10、协程间的数据传递：通道（Channel）设计与实现

协程跑起来了，任务切出去了，调度器也转得挺顺。但有个问题一直绕不开——协程之间怎么传数据？

全局变量？加锁？信号量？嗯，这些都能用，但用起来总觉得别扭。我在做嵌入式网络协议栈的时候，协程A负责收包，协程B负责解析，协程C负责转发。三个协程之间要频繁交换数据。用全局队列加锁？死锁排查能让你怀疑人生。用管道？那是进程级的，太重了。

后来我悟了：协程之间传递数据，最优雅的方式就是通道（Channel）。说白了，就是一个带缓冲的、支持多生产者多消费者的数据管道。协程往里写，协程往外读，读写双方都不用关心对方是谁。

10.1 通道的核心设计思想

通道的本质是什么？我个人的理解是：一个线程安全的、有容量限制的环形缓冲区，加上阻塞/唤醒机制。

你想想看，协程A往通道里写数据，如果通道满了，协程A就挂起等待。协程B从通道里读数据，如果通道空了，协程B就挂起等待。这不就是生产者-消费者模型吗？

但这里有个关键点：挂起和唤醒必须由调度器来完成。不能像线程那样用条件变量，因为协程是用户态调度的，挂起就是yield出去，唤醒就是重新加入就绪队列。

通道的三个核心要素：

缓冲区：存储数据的环形队列，有固定容量
等待队列：记录哪些协程在等待读/写
锁：保护缓冲区操作的原子性（注意，这里锁的粒度要极细）

10.2 通道的数据结构设计

先看代码，这是我项目中实际用过的通道结构体。嗯，这里要注意，通道的缓冲区大小必须是2的幂，为什么？往下看。

// 通道结构体
typedef struct channel {
    void **buffer;          // 环形缓冲区，存储数据指针
    int capacity;           // 缓冲区容量（2的幂）
    int mask;               // 容量-1，用于快速取模
    int head;               // 读指针
    int tail;               // 写指针
    int closed;             // 关闭标志
    
    // 等待队列
    coroutine_queue_t *recv_waiters;  // 等待读的协程队列
    coroutine_queue_t *send_waiters;  // 等待写的协程队列
    
    // 锁
    spinlock_t lock;        // 自旋锁，保护通道内部状态
} channel_t;

为什么容量是2的幂？因为可以用位运算代替取模运算。head & mask 比 head % capacity 快一个数量级。我在一个高频交易系统中用过这个技巧，每秒几百万次通道操作，省下来的CPU时间相当可观。

10.3 通道的创建与销毁

创建通道很简单，分配内存，初始化各个字段。但我曾经踩过一个坑：缓冲区用malloc分配，但协程环境里malloc不是线程安全的。后来我改成了在创建时一次性分配，并且用原子操作保护。

channel_t* channel_create(int capacity) {
    // 确保容量是2的幂
    if (capacity & (capacity - 1)) {
        // 向上取整到最近的2的幂
        int n = 1;
        while (n < capacity) n <<= 1;
        capacity = n;
    }
    
    channel_t *ch = (channel_t*)malloc(sizeof(channel_t));
    ch->buffer = (void**)malloc(sizeof(void*) * capacity);
    ch->capacity = capacity;
    ch->mask = capacity - 1;
    ch->head = 0;
    ch->tail = 0;
    ch->closed = 0;
    ch->recv_waiters = coroutine_queue_create();
    ch->send_waiters = coroutine_queue_create();
    spinlock_init(&ch->lock);
    return ch;
}

注意：通道销毁时，必须确保没有协程在等待。否则那些挂起的协程会永远睡下去。我建议的做法是：先关闭通道，然后唤醒所有等待协程，让它们返回错误，最后再释放内存。

10.4 通道的读写操作——核心逻辑

读写操作是通道的灵魂。我把它拆成四个步骤：

加锁：保护内部状态
检查条件：缓冲区是否可读/可写
执行操作：拷贝数据，更新指针
唤醒等待者：如果有协程在等，就唤醒一个

先看写操作：

int channel_send(channel_t *ch, void *data) {
    spinlock_lock(&ch->lock);
    
    // 检查通道是否关闭
    if (ch->closed) {
        spinlock_unlock(&ch->lock);
        return -1;  // 通道已关闭
    }
    
    // 缓冲区满了，挂起当前协程
    while (ch->tail - ch->head == ch->capacity) {
        // 把当前协程加入发送等待队列
        coroutine_t *self = coroutine_self();
        coroutine_queue_push(ch->send_waiters, self);
        spinlock_unlock(&ch->lock);
        
        // 挂起当前协程，让出CPU
        coroutine_yield();
        
        // 被唤醒后重新加锁检查
        spinlock_lock(&ch->lock);
        
        // 如果通道被关闭，退出
        if (ch->closed) {
            spinlock_unlock(&ch->lock);
            return -1;
        }
    }
    
    // 写入数据
    ch->buffer[ch->tail & ch->mask] = data;
    ch->tail++;
    
    // 如果有协程在等待读，唤醒一个
    if (!coroutine_queue_empty(ch->recv_waiters)) {
        coroutine_t *waiter = coroutine_queue_pop(ch->recv_waiters);
        coroutine_resume(waiter);  // 把等待者加入就绪队列
    }
    
    spinlock_unlock(&ch->lock);
    return 0;
}

读操作是对称的：

void* channel_recv(channel_t *ch) {
    spinlock_lock(&ch->lock);
    
    // 缓冲区空了，挂起当前协程
    while (ch->tail == ch->head) {
        // 如果通道已关闭且没有数据，返回NULL
        if (ch->closed) {
            spinlock_unlock(&ch->lock);
            return NULL;
        }
        
        coroutine_t *self = coroutine_self();
        coroutine_queue_push(ch->recv_waiters, self);
        spinlock_unlock(&ch->lock);
        
        coroutine_yield();
        
        spinlock_lock(&ch->lock);
        if (ch->closed && ch->tail == ch->head) {
            spinlock_unlock(&ch->lock);
            return NULL;
        }
    }
    
    // 读取数据
    void *data = ch->buffer[ch->head & ch->mask];
    ch->head++;
    
    // 唤醒一个等待写的协程
    if (!coroutine_queue_empty(ch->send_waiters)) {
        coroutine_t *waiter = coroutine_queue_pop(ch->send_waiters);
        coroutine_resume(waiter);
    }
    
    spinlock_unlock(&ch->lock);
    return data;
}

个人经验：这里有个容易被忽略的细节——唤醒等待者必须在释放锁之前。为什么？因为如果先释放锁再唤醒，被唤醒的协程可能抢不到锁，又得挂起，造成不必要的上下文切换。我在一个项目中因为这个顺序问题，性能下降了30%。

10.5 通道的关闭与多路复用

通道关闭是个敏感操作。我建议的做法是：

设置 closed = 1
唤醒所有等待读的协程，让它们返回 NULL
唤醒所有等待写的协程，让它们返回 -1
之后的所有读写操作都立即返回错误

另外，实际项目中经常需要同时等待多个通道。比如一个协程既要处理网络数据，又要处理用户输入。这时候就需要 channel_select 机制。实现思路是：

把当前协程同时注册到多个通道的等待队列中
哪个通道先就绪，就唤醒协程
协程被唤醒后，检查是哪个通道就绪了

这个实现比较复杂，但核心思想就是一个协程可以同时等待多个事件源。

10.6 通道的完整流程图

下面这张图展示了通道的完整工作流程，从创建到读写，再到关闭：

10.7 避坑指南与性能优化

通道用起来爽，但坑也不少。我把自己踩过的坑列出来：

我曾经踩过的坑：

死锁：协程A持有通道1的锁，等待通道2；协程B持有通道2的锁，等待通道1。解决方案：统一锁顺序，或者用trylock。
惊群效应：唤醒所有等待者时，只有一个能拿到数据，其他又得挂起。解决方案：只唤醒一个，或者用条件变量+广播。
内存泄漏：通道里存的是指针，如果协程崩溃了，没人释放数据。解决方案：使用智能指针，或者在通道关闭时清理残留数据。
优先级反转：高优先级协程等待低优先级协程释放通道。解决方案：优先级继承，或者用多个通道分级。

性能优化方面，我建议：

批量操作：一次读写多个元素，减少锁竞争
无锁队列：对于单生产者单消费者场景，可以用无锁环形队列，性能提升5-10倍
缓存行对齐：把head和tail放在不同的缓存行，避免伪共享
批量唤醒：如果一次写入了多个数据，可以唤醒多个读协程

10.8 总结

通道是协程间通信的基石。它把复杂的数据传递问题，抽象成了一个简单的读写接口。你不需要关心对方是谁，不需要关心数据怎么同步，只需要往通道里扔数据，或者从通道里取数据。

我个人觉得，通道的设计哲学就是「不要通过共享内存来通信，而要通过通信来共享内存」。这句话在协程世界里尤其适用。

嗯，通道的实现看起来简单，但真正用好它，需要你对调度器、锁、等待队列都有深刻的理解。希望今天的分享能帮你少走一些弯路。

核心要点回顾：

通道 = 环形缓冲区 + 等待队列 + 锁
容量必须是2的幂，用位运算加速
读写操作要对称，唤醒等待者要在释放锁之前
关闭通道时要唤醒所有等待者
多路复用用 select 机制

公众号:蓝海资料掘金营，微信deep3321

📚 课程目录

10、协程间的数据传递：通道（Channel）设计与实现

10.1 通道的核心设计思想

10.2 通道的数据结构设计

10.3 通道的创建与销毁

10.4 通道的读写操作——核心逻辑

10.5 通道的关闭与多路复用

10.6 通道的完整流程图

10.7 避坑指南与性能优化

10.8 总结