C 多线程编程中的互斥锁与条件变量：解决线程协作问题

多线程程序除了需要避免多个线程同时修改共享数据，还经常需要让线程彼此配合：消费者在队列为空时等待，生产者放入数据后唤醒消费者；工作线程等待任务到来，管理线程则在任务就绪时通知它们。这类问题既涉及数据保护，也涉及等待和唤醒。

在 POSIX C 程序中，互斥锁（mutex）和条件变量（condition variable）通常配合使用。互斥锁保护共享状态，条件变量让线程在某个状态尚未满足时休眠，并在状态可能改变时重新检查。理解这两者各自负责什么，以及等待时如何避免竞态，是写出可靠线程协作代码的关键。

互斥锁和条件变量分别解决什么问题

互斥锁：保护共享状态

当多个线程访问同一块可变数据，且至少有一个线程会修改它时，如果访问没有同步，就可能发生数据竞争。数据竞争不仅会造成逻辑错误；在 C 语言内存模型中，未同步的冲突访问还可能导致未定义行为。

互斥锁用于保证同一时刻只有一个线程进入受保护的临界区。通常做法是：线程先锁定互斥锁，再读取或修改共享状态，完成后解锁。所有访问相关状态的线程都必须遵守同一套锁定规则。

pthread_mutex_lock(&mutex);
shared_count++;
pthread_mutex_unlock(&mutex);

锁保护的对象不必只有一个变量。例如，环形队列的数组、读写位置、元素数量和关闭标志可能共同构成一份状态。若这些字段之间存在不变量，就应使用一致的同步方式保护它们，而不是只锁住其中某个字段。

条件变量：等待状态变化

条件变量本身不保存业务状态，也不是“事件已经发生”的标志。它提供的是一种等待机制：线程在持有互斥锁时检查共享状态；如果状态不满足，就通过 pthread_cond_wait 原子地释放互斥锁并进入等待；被唤醒后重新取得互斥锁，再检查状态。

例如，消费者等待队列非空，真正的判断条件是“队列中有元素”，而不是“条件变量收到过通知”。通知可能早于等待发生，也可能唤醒多个竞争者，还可能出现虚假唤醒。因此，状态必须存放在受互斥锁保护的共享变量中，条件变量只负责提醒等待者重新检查状态。

适用场景

• 生产者—消费者队列：消费者等待队列非空，生产者等待队列有空位；队列状态由互斥锁保护。
• 任务调度：工作线程等待任务列表中出现任务，调度线程添加任务后通知一个或多个工作线程。
• 阶段同步：线程等待某个阶段完成，例如初始化结束或共享资源准备就绪。
• 资源池：线程等待可用连接、缓冲区或其他资源；资源数量是受锁保护的条件状态。

条件变量适合等待“由共享状态定义的条件”，而不是用来保护数据。如果任务队列本身需要安全访问，仍然需要互斥锁，或者采用经过正确设计的无锁数据结构。条件变量也不是通用的跨进程通知方式：本文示例使用默认的进程私有属性，供同一进程内的线程协作。

核心规则：等待条件，不是等待通知

pthread_cond_wait 的典型形式必须放在循环中：

pthread_mutex_lock(&mutex);
while (!condition) {
    pthread_cond_wait(&cond, &mutex);
}
/* condition 已在持锁状态下检查 */
use_shared_state();
pthread_mutex_unlock(&mutex);

使用 while 而不是 if，原因主要有三点：

1. 可能发生虚假唤醒。 POSIX 允许等待线程在没有对应通知的情况下返回，因此返回不代表条件成立。
2. 多个线程可能争用同一状态。 一个线程被通知并重新取得锁之前，其他线程可能已经消费了资源，条件又变回不成立。
3. 通知不携带资源，也不保证一对一。 被唤醒只是获得重新检查条件的机会。

pthread_cond_wait(&cond, &mutex) 要求调用线程已经锁定 mutex。调用时，函数会原子地释放该互斥锁并等待；返回前会重新锁定它。因此线程恢复执行后，仍持有互斥锁。条件检查、等待前的解锁和进入等待之间不会留下一个可让通知丢失的普通窗口。

这里的“原子”指释放互斥锁并开始等待的同步语义，不代表整个业务操作都是原子的。调用返回后仍需检查共享状态，并在适当时解锁。

可运行示例：有界队列、生产者和消费者

下面的程序实现一个固定容量的环形队列。一个生产者依次放入 10 个整数，然后关闭队列；两个消费者取出数据，队列关闭且已排空时退出。队列的数组、读写位置、元素数量和关闭标志都由同一把互斥锁保护。

程序使用两个条件变量：not_empty 表示消费者可能有数据可取，not_full 表示生产者可能有空间可用。生产者放入元素后通知消费者，消费者取出元素后通知生产者。队列关闭时，生产者广播唤醒所有等待中的消费者，使它们能够检查关闭状态并退出。

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>

#define CAPACITY 4
#define ITEM_COUNT 10
#define CONSUMER_COUNT 2

typedef struct {
    int data[CAPACITY];
    int head;
    int tail;
    int count;
    int closed;
    pthread_mutex_t mutex;
    pthread_cond_t not_empty;
    pthread_cond_t not_full;
} queue_t;

static queue_t queue = {
    .head = 0,
    .tail = 0,
    .count = 0,
    .closed = 0,
    .mutex = PTHREAD_MUTEX_INITIALIZER,
    .not_empty = PTHREAD_COND_INITIALIZER,
    .not_full = PTHREAD_COND_INITIALIZER
};

static void check_pthread(int error, const char *what)
{
    if (error != 0) {
        fprintf(stderr, "%s failed: %d\n", what, error);
        exit(EXIT_FAILURE);
    }
}

/* 调用时必须持有 queue.mutex。 */
static void queue_put(int value)
{
    while (queue.count == CAPACITY && !queue.closed) {
        check_pthread(
            pthread_cond_wait(&queue.not_full, &queue.mutex),
            "pthread_cond_wait(not_full)");
    }

    if (queue.closed) {
        pthread_mutex_unlock(&queue.mutex);
        fprintf(stderr, "cannot put into a closed queue\n");
        exit(EXIT_FAILURE);
    }

    queue.data[queue.tail] = value;
    queue.tail = (queue.tail + 1) % CAPACITY;
    queue.count++;

    check_pthread(pthread_cond_signal(&queue.not_empty),
                  "pthread_cond_signal(not_empty)");
}

/* 调用时必须持有 queue.mutex。
   有数据时返回 1 并写入 *value；关闭且排空时返回 0。 */
static int queue_get(int *value)
{
    while (queue.count == 0 && !queue.closed) {
        check_pthread(
            pthread_cond_wait(&queue.not_empty, &queue.mutex),
            "pthread_cond_wait(not_empty)");
    }

    if (queue.count == 0) {
        return 0;
    }

    *value = queue.data[queue.head];
    queue.head = (queue.head + 1) % CAPACITY;
    queue.count--;

    check_pthread(pthread_cond_signal(&queue.not_full),
                  "pthread_cond_signal(not_full)");
    return 1;
}

static void *producer(void *arg)
{
    (void)arg;

    for (int i = 1; i <= ITEM_COUNT; i++) {
        check_pthread(pthread_mutex_lock(&queue.mutex),
                      "pthread_mutex_lock");
        queue_put(i);
        check_pthread(pthread_mutex_unlock(&queue.mutex),
                      "pthread_mutex_unlock");

        printf("produced %d\n", i);
    }

    check_pthread(pthread_mutex_lock(&queue.mutex),
                  "pthread_mutex_lock");
    queue.closed = 1;
    check_pthread(pthread_cond_broadcast(&queue.not_empty),
                  "pthread_cond_broadcast(not_empty)");
    check_pthread(pthread_cond_broadcast(&queue.not_full),
                  "pthread_cond_broadcast(not_full)");
    check_pthread(pthread_mutex_unlock(&queue.mutex),
                  "pthread_mutex_unlock");

    return NULL;
}

static void *consumer(void *arg)
{
    int id = *(int *)arg;

    for (;;) {
        int value;

        check_pthread(pthread_mutex_lock(&queue.mutex),
                      "pthread_mutex_lock");
        int got_item = queue_get(&value);
        check_pthread(pthread_mutex_unlock(&queue.mutex),
                      "pthread_mutex_unlock");

        if (!got_item) {
            break;
        }

        printf("consumer %d consumed %d\n", id, value);
    }

    printf("consumer %d exiting\n", id);
    return NULL;
}

int main(void)
{
    pthread_t producer_thread;
    pthread_t consumer_threads[CONSUMER_COUNT];
    int ids[CONSUMER_COUNT];

    for (int i = 0; i < CONSUMER_COUNT; i++) {
        ids[i] = i + 1;
        check_pthread(
            pthread_create(&consumer_threads[i], NULL, consumer, &ids[i]),
            "pthread_create(consumer)");
    }

    check_pthread(pthread_create(&producer_thread, NULL, producer, NULL),
                  "pthread_create(producer)");

    check_pthread(pthread_join(producer_thread, NULL), "pthread_join(producer)");

    for (int i = 0; i < CONSUMER_COUNT; i++) {
        check_pthread(pthread_join(consumer_threads[i], NULL),
                      "pthread_join(consumer)");
    }

    check_pthread(pthread_cond_destroy(&queue.not_empty),
                  "pthread_cond_destroy(not_empty)");
    check_pthread(pthread_cond_destroy(&queue.not_full),
                  "pthread_cond_destroy(not_full)");
    check_pthread(pthread_mutex_destroy(&queue.mutex),
                  "pthread_mutex_destroy");

    return 0;
}

将代码保存为 queue.c，在常见的 Linux 环境中可使用以下命令编译运行：

cc -std=c11 -Wall -Wextra -Wpedantic -pthread queue.c -o queue
./queue

-pthread 用于启用相应的编译和链接选项。输出顺序会因线程调度而变化，但每个整数只会被消费一次，两个消费者最终都会退出。示例为便于聚焦同步逻辑，假设线程创建成功；完整应用还应设计创建失败时的资源回收和已启动线程的退出流程。

示例中的关键细节

• 等待时使用循环。生产者在队列满且未关闭时等待；消费者在队列空且未关闭时等待。每次被唤醒后，都会重新检查共享状态。
• 条件和状态由同一把锁保护。队列计数、关闭标志以及环形队列操作都在持锁状态下读写，避免检查条件和改变状态之间产生竞态。
• 通知放在状态更新之后。生产者先写入数据并增加计数，再通知消费者；消费者先移除数据并减少计数，再通知生产者。通知不是状态更新的替代品。
• 关闭也是共享状态。队列关闭后，消费者仍会取出已存在的数据；只有在关闭且队列为空时才退出。
• 关闭时广播。可能有多个消费者正在等待数据。只唤醒一个可能让其他线程继续睡眠，因此关闭时对消费者广播。示例也广播了 not_full，以免未来扩展为多个生产者时，关闭状态使等待的生产者无法重新检查退出条件。
• 消费者拿走数据后及时解锁。示例在锁外打印，避免慢速 I/O 长时间占用队列锁，降低其他线程等待时间。

signal 与 broadcast 如何选择

pthread_cond_signal 至少会使一个正在该条件变量上等待的线程具备继续执行的机会；如果没有等待者，信号不会被保存，供未来线程“领取”。它适合状态变化后只需唤醒一个等待者的场景，例如有一个新任务进入队列时，通常只需唤醒一个消费者。

pthread_cond_broadcast 会使所有当前等待在该条件变量上的线程都具备继续执行的机会。它适用于可能影响所有等待者的状态变化，例如资源池被关闭、等待条件整体失效，或多个线程都可能因此满足条件的情况。广播不等于所有线程都会同时继续工作：线程仍需逐个重新取得互斥锁，并检查条件。

选择哪种通知方式取决于状态变化影响哪些等待者。错误地只唤醒一个线程，可能造成其他线程长期等待；无差别广播则可能导致大量线程被唤醒后又发现条件不满足，形成额外竞争。通知策略应与等待条件和状态转移一起设计。

容易忽略的边界和错误

用 if 等待，或把通知当作条件

下面的写法不可靠：

if (queue.count == 0) {
    pthread_cond_wait(&not_empty, &mutex);
}

线程返回时队列仍可能为空，或共享状态已经被另一个线程改变。必须用 while 重新判断受保护的条件。类似地，不要设置一个简单的“收到信号”变量后就假定资源可用；变量只有在锁保护下与实际资源状态一致，才可作为判断依据。

检查条件时没有持有互斥锁

若线程先在锁外检查“队列为空”，随后才调用等待，就可能在这两步之间发生状态变化和通知。线程可能错过通知并一直睡眠。正确模式是在持有互斥锁时检查条件，再通过 pthread_cond_wait 等待；该函数负责原子地释放锁并进入等待过程。

修改状态时没有遵守同一把锁

条件判断受锁保护，但某个生产者在锁外改写计数或关闭标志，仍然是不安全的。所有参与条件判断的共享变量都应有一致的同步规则。仅仅“条件变量周围有一把锁”并不能自动保护其他代码对变量的访问。

持锁执行慢操作或调用未知代码

持有互斥锁时进行文件或网络 I/O、长时间计算，或调用可能再次获取同一把锁的函数，都会延长临界区，甚至引发死锁。通常应在锁内完成必要的状态读取和更新，然后尽快解锁，再进行耗时工作。示例将打印操作放在锁外，但输出本身的行序仍不固定。

遗漏关闭或退出通知

线程可能正在条件变量上等待，而主线程直接结束进程并不等同于有序关闭线程。队列、线程池等长期运行组件应明确关闭状态，并在状态改变后唤醒相关等待者。等待线程醒来后还要检查关闭和排空条件，不能仅凭“被唤醒”就退出或继续取数据。

在不同互斥锁保护下等待同一个条件变量

对同一个条件变量进行等待时，应遵循 POSIX 对动态绑定的要求：在等待者尚未结束等待期间，不能把该条件变量与另一把互斥锁混用。工程上通常为一个条件变量指定固定的关联互斥锁，并让所有相关等待和状态检查都使用它。

忘记解锁、重复解锁或销毁仍在使用的对象

每条控制路径都要与锁定操作配对。线程还在等待、运行或可能再次访问对象时，不能销毁相关互斥锁和条件变量。销毁前应先停止新任务，改变状态并通知等待者，等待相关线程退出，再进行清理。

忽视错误码

多数 pthread 函数直接返回错误码，而不是通过返回值表示成功或失败后再依赖 errno。实际程序应按其返回值处理错误。示例用辅助函数检查部分 API 的错误，以保持主体清晰；生产环境还应避免在线程已经启动后简单调用 exit，而应设计能够安全释放资源和通知其他线程的错误处理流程。

初始化、超时等待与参数边界

示例用 PTHREAD_MUTEX_INITIALIZER 和 PTHREAD_COND_INITIALIZER 初始化静态对象。动态对象可以通过 pthread_mutex_init、pthread_cond_init 初始化，并在不再使用且没有线程持有或等待时调用相应的销毁函数。初始化属性允许调整部分行为，但应明确属性的适用平台和进程共享需求；不要把默认对象误当成跨进程同步对象。

若等待需要超时，可使用 pthread_cond_timedwait。它使用绝对时间截止点；超时返回后，调用线程仍需重新取得互斥锁，并像其他返回情况一样重新检查条件。函数返回超时并不表示业务条件一定没有同时改变，也不意味着可以绕过锁直接访问共享数据。若要减少系统时钟调整对截止时间的影响，可考虑通过条件变量属性选择单调时钟，但该接口的可用性和配置应以目标平台的 POSIX 实现为准。

不要把任意指针、未初始化对象或已经销毁的互斥锁传给相关函数。互斥锁和条件变量必须先正确初始化，并保持有效直到所有使用者都退出等待和访问。线程同步 API 不会替应用验证队列容量、索引范围或对象生命周期；这些仍是程序自身必须维护的不变量。

互斥锁与自旋锁、原子操作的取舍

互斥锁适合临界区可能较长、线程可能需要阻塞等待的常见共享状态保护。自旋锁会让等待线程持续占用处理器反复检查，通常只适合很短、且经过测量确认的场景；它不能替代条件变量来高效等待某个长期未满足的条件。

原子操作适用于某些独立计数器或简单状态转换，但当多个字段必须保持一致、或线程需要等待复杂条件时，仅把其中一个变量改成原子类型并不能自动解决协作问题。不要为了避免互斥锁而自行拼凑复杂的内存序逻辑，除非确实需要并能证明算法正确。

安全与可靠性：同步不是安全边界

互斥锁可以帮助保护进程内共享数据，降低数据竞争导致的内存破坏和状态不一致风险；它本身不是安全边界，也不能替代输入校验、访问控制或资源配额。攻击者不能仅靠“有一把锁”就失去访问权限，锁只约束遵守该协议的线程。

多线程代码还需留意拒绝服务类问题：线程可能因遗漏通知而永久等待，长时间持锁会造成其他线程饥饿或阻塞，队列容量和生产速率不匹配也可能导致资源耗尽。设置有界队列、明确关闭流程、限制任务资源消耗，并对超时与错误进行合理处理，往往比盲目增加线程更重要。

若需要线程间的公平性、优先级继承、跨进程共享等特性，应查阅目标系统对 pthread 属性和调度策略的支持与限制，不能假定所有 POSIX 系统的默认行为相同。涉及安全敏感或高可靠场景时，还应结合目标平台测试、静态分析和并发压力测试；同步原语本身不能证明整个协议没有死锁或活锁。

设计时的检查清单

• 明确共享状态和不变量，并指定由哪把互斥锁保护。
• 把条件表达为可检查的共享状态，例如“队列非空”或“队列已关闭且排空”。
• 在持锁状态下检查条件；等待时使用 while，返回后重新检查。
• 先修改状态，再按需要调用 signal 或 broadcast。
• 确定关闭、超时、错误和线程退出的完整路径。
• 缩短临界区，并确保对象不会在仍被使用时销毁。

总结来说，互斥锁负责让共享状态的访问保持一致，条件变量负责让线程在条件未满足时高效等待。可靠的线程协作并不依赖通知“恰好到来”，而依赖受锁保护的状态、循环检查的等待协议，以及明确的生命周期管理。把这三点落实到代码中，生产者—消费者、任务队列和资源池等常见问题就更容易实现，也更容易审查和维护。