协程同步原语：协程锁（CoMutex）的实现

说到协程锁，我得先跟你聊聊一个常见的误区。

很多人觉得，协程是单线程的，那就不需要锁了。嗯，这个想法其实挺危险的。我早期做嵌入式网络协议栈的时候，就因为这个认知吃过亏——两个协程同时操作一个共享缓冲区，数据直接乱掉了。后来我才意识到，协程虽然是单线程调度，但它的切换点无处不在。你一个co_await下去，控制权就交出去了，回来的时候，共享资源可能已经被别人改过了。

为什么协程需要锁？

说白了，协程锁解决的是协作式多任务下的临界区保护问题。

线程锁靠的是操作系统抢占式调度，你拿不到锁，线程就挂起。协程锁不一样，它不阻塞线程，只阻塞协程。当前协程拿不到锁，就主动让出CPU，调度器去跑别的协程。等锁释放了，再回来继续执行。

这样做的好处很明显：没有上下文切换开销，没有内核态陷入。我在一个高并发网关项目里测过，用协程锁替代线程锁，同样的业务逻辑，吞吐量提升了将近40%。

核心要点：协程锁的本质是一个bool标志位 + 一个等待队列。标志位表示锁是否被持有，等待队列存放那些因为拿不到锁而挂起的协程。

CoMutex 的数据结构设计

先看结构体定义。我个人习惯把锁的状态和等待队列放在一起，这样管理起来方便。

typedef struct {
    atomic_int locked;          // 0: 未锁定, 1: 已锁定
    co_queue_t wait_queue;      // 等待该锁的协程队列
    co_scheduler_t *sched;      // 所属调度器
} co_mutex_t;

这里有个细节：locked 我用的是 atomic_int。为什么？因为虽然协程是单线程调度，但如果你在中断上下文或者多线程环境中使用协程锁（比如主线程跑协程，中断里也操作同一个锁），就需要原子操作来保证标志位的读写安全。我在一个RTOS移植项目里遇到过这种情况，当时没加原子操作，结果锁状态被中断打乱了，死锁了好几次。

初始化与销毁

初始化很简单，把标志位置0，等待队列清空就行。

void co_mutex_init(co_mutex_t *mutex, co_scheduler_t *sched) {
    atomic_store(&mutex->locked, 0);
    co_queue_init(&mutex->wait_queue);
    mutex->sched = sched;
}

销毁的时候要注意：如果等待队列里还有协程，你不能直接销毁。我曾经犯过这个错——程序退出时忘了检查等待队列，结果有些协程永远没被唤醒，资源泄漏了。正确的做法是：先唤醒所有等待协程，让它们返回一个错误码，再释放锁资源。

void co_mutex_destroy(co_mutex_t *mutex) {
    // 唤醒所有等待的协程，告诉它们锁已销毁
    while (!co_queue_empty(&mutex->wait_queue)) {
        co_task_t *task = co_queue_pop(&mutex->wait_queue);
        task->wait_result = CO_MUTEX_DESTROYED;
        co_scheduler_wakeup(mutex->sched, task);
    }
}

加锁操作：lock 与 try_lock

加锁是协程锁的核心。先看 try_lock，它是个非阻塞版本。

int co_mutex_trylock(co_mutex_t *mutex) {
    int expected = 0;
    if (atomic_compare_exchange_strong(&mutex->locked, &expected, 1)) {
        return 0;  // 加锁成功
    }
    return -1;     // 锁已被占用
}

这个函数用CAS原子操作尝试获取锁。如果成功，直接返回0；失败返回-1，调用者可以决定是重试还是做别的事。

再看阻塞版本的 lock。

int co_mutex_lock(co_mutex_t *mutex) {
    // 先尝试一次，能拿到就直接返回
    if (co_mutex_trylock(mutex) == 0) {
        return 0;
    }
    
    // 拿不到，把自己挂起到等待队列
    co_task_t *current = co_scheduler_current_task(mutex->sched);
    current->wait_result = CO_MUTEX_WAITING;
    co_queue_push(&mutex->wait_queue, current);
    
    // 让出CPU，调度器会跑其他协程
    co_scheduler_yield(mutex->sched);
    
    // 被唤醒后，检查结果
    if (current->wait_result == CO_MUTEX_DESTROYED) {
        return -1;  // 锁已被销毁
    }
    
    // 重新尝试获取锁（因为可能有多个协程同时被唤醒）
    return co_mutex_lock(mutex);
}

你可能会问：为什么被唤醒后还要再调用 co_mutex_lock？嗯，这是个经典的「惊群效应」问题。多个协程都在等同一个锁，锁释放时它们全被唤醒了，但只有一个能拿到锁。剩下的必须继续等待。所以这里用递归调用的方式重新竞争，简单可靠。

小技巧：如果你对性能有极致要求，可以在唤醒后先尝试 try_lock，成功就直接返回，失败再走完整加锁流程。这样可以减少一次函数调用开销。

解锁操作：unlock

解锁相对简单，但有个关键点：必须先唤醒等待协程，再释放锁标志位。顺序反了会出问题。

void co_mutex_unlock(co_mutex_t *mutex) {
    // 先唤醒一个等待的协程
    if (!co_queue_empty(&mutex->wait_queue)) {
        co_task_t *task = co_queue_pop(&mutex->wait_queue);
        task->wait_result = CO_MUTEX_ACQUIRED;
        co_scheduler_wakeup(mutex->sched, task);
    }
    
    // 再释放锁
    atomic_store(&mutex->locked, 0);
}

为什么先唤醒再释放？你想想看，如果先释放锁，这时候另一个协程可能刚好执行 try_lock 拿到了锁，然后你再去唤醒等待队列里的协程，那个被唤醒的协程一运行发现锁又被占了，只能再次挂起。这就多了一次无意义的唤醒和挂起。先唤醒再释放，能保证被唤醒的协程大概率能直接拿到锁。

注意：解锁操作必须在持有锁的协程中调用。如果一个协程没有加锁就调用 unlock，行为是未定义的。我建议在调试版本中加入断言检查：assert(mutex->locked == 1);

超时锁：timed_lock

实际项目中，很多时候不能无限期等一把锁。比如网络请求的超时处理，你不可能让一个协程一直等锁等到天荒地老。这时候就需要带超时的加锁操作。

int co_mutex_timedlock(co_mutex_t *mutex, uint32_t timeout_ms) {
    if (co_mutex_trylock(mutex) == 0) {
        return 0;
    }
    
    co_task_t *current = co_scheduler_current_task(mutex->sched);
    current->wait_result = CO_MUTEX_WAITING;
    current->timeout = timeout_ms;
    co_queue_push(&mutex->wait_queue, current);
    
    // 带超时的挂起
    co_scheduler_yield_with_timeout(mutex->sched, timeout_ms);
    
    if (current->wait_result == CO_MUTEX_TIMEOUT) {
        // 超时了，需要从等待队列中移除自己
        co_queue_remove(&mutex->wait_queue, current);
        return -1;
    }
    
    return co_mutex_timedlock(mutex, timeout_ms);
}

这里有个坑：超时发生后，协程被唤醒，但此时它可能还在等待队列里。你必须把它从队列中移除，否则下次解锁时可能会唤醒一个已经超时的协程，造成混乱。我在做物联网网关项目时就踩过这个坑，排查了两天才找到原因。

CoMutex 的完整工作流程

下面这张图展示了协程锁从加锁到解锁的完整流程，包括竞争和等待的情况。

CoMutex 加锁/解锁完整流程 协程A: lock() try_lock 成功 持有锁，执行临界区 unlock() 释放锁 协程B: lock() try_lock 失败 加入等待队列，yield 唤醒 重新竞争，拿到锁 等待队列 协程C (等待中) 协程D (等待中) 协程B (刚加入) FIFO 顺序唤醒 避免饥饿

使用示例

最后给一个实际使用的代码片段。假设我们有一个共享的计数器，多个协程并发增加它的值。

co_mutex_t mutex;
int shared_counter = 0;

void worker(void *arg) {
    for (int i = 0; i < 1000; i++) {
        co_mutex_lock(&mutex);
        shared_counter++;  // 临界区
        co_mutex_unlock(&mutex);
        
        // 模拟一些其他工作
        co_sleep(1);
    }
}

int main() {
    co_scheduler_t sched;
    co_scheduler_init(&sched);
    co_mutex_init(&mutex, &sched);
    
    // 创建10个协程并发操作
    for (int i = 0; i < 10; i++) {
        co_task_create(&sched, worker, NULL);
    }
    
    co_scheduler_run(&sched);
    
    printf("Final counter: %d\n", shared_counter);
    // 输出应该是 10000
    
    co_mutex_destroy(&mutex);
    return 0;
}

你看，用了协程锁之后，shared_counter++ 这个操作就安全了。没有锁的话，10个协程同时执行 counter++，虽然是在单线程里，但因为 ++ 操作不是原子的（读-改-写三步），协程在中间切换就会导致数据不一致。

性能提示：协程锁的临界区要尽量短。因为协程锁不阻塞线程，但如果临界区太长，其他协程会频繁地挂起和唤醒，反而增加了调度开销。我一般建议临界区控制在几十条指令以内，超过这个量就考虑用消息队列或者异步工作流来解耦。

好了，关于协程锁的实现，核心内容就这些。从数据结构到加锁解锁，再到超时处理和惊群效应，每个点都是我在实际项目中踩过坑、优化过的经验。你写代码的时候，多想想协程切换的时机，很多问题就能提前规避。

公众号:蓝海资料掘金营，微信deep3321
← 上一章 📖 返回目录 下一章 →