使用atomic_compare_exchange_weak实现生产者-消费者问题时重复访问的解决方法

你的代码存在多个核心问题，导致重复访问和同步失效，逐一拆解如下：

1. 互斥锁设计完全错误 #

生产者使用lock_p、消费者使用lock_c，但两者操作的是同一组共享资源（buffer、in、out、counter），这种拆分锁的方式完全起不到互斥作用——生产者和消费者可以同时进入各自的"临界区"，直接引发数据竞争。

2. 全局expected变量破坏CAS逻辑 #

expected_p和expected_c是全局变量，多个生产者/消费者线程会同时修改它们。atomic_compare_exchange_weak的expected参数是输入输出型，线程间的干扰会导致CAS判断失效，锁无法正确获取。

3. 解锁逻辑不完整且非原子 #

• 仅在部分分支解锁：生产者仅在task_log[item][0] == -1时解锁，进入else分支continue后锁会被永久持有，引发死锁；消费者同理。
• 直接赋值解锁：lock_p = false;不是原子操作，对于atomic_bool类型，必须用atomic_store进行原子赋值，否则会破坏锁的状态一致性。

4. 忙等循环存在竞态且低效 #

while(counter >= BUFSIZE)和while(counter <= 0)在锁外判断条件，多个线程可能同时通过判断，进入临界区后条件已不满足（比如多个生产者同时看到counter < BUFSIZE，但进入后缓冲区已被填满），导致错误。

5. 非原子操作的共享变量 #

next_item++不是原子操作，即使在锁内，若锁本身失效，依然会引发重复生产。

修复后的代码 #

推荐使用POSIX线程标准的pthread_mutex_t（互斥锁）和pthread_cond_t（条件变量）实现，这是生产者-消费者问题的标准高效解法，避免手动CAS实现锁的风险：

全局变量定义 #

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

#define BUFSIZE 8

// 共享缓冲区相关
int buffer[BUFSIZE];
int in = 0;
int out = 0;
int counter = 0;
int next_item = 0;

// 同步原语
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t not_full = PTHREAD_COND_INITIALIZER;   // 生产者等待：缓冲区非满
pthread_cond_t not_empty = PTHREAD_COND_INITIALIZER;  // 消费者等待：缓冲区非空

// 日志与统计
int task_log[100][2]; // 假设最多处理100个项
int produced = 0;
int consumed = 0;
_Bool alive = true;

生产者线程 #

void *producer(void *arg) {
    int i = *(int *)arg;
    int item;
    
    while(alive) {
        // 获取互斥锁
        pthread_mutex_lock(&mutex);
        
        // 等待缓冲区有空位，避免忙等
        while (counter >= BUFSIZE) {
            pthread_cond_wait(&not_full, &mutex);
            // 被唤醒时自动重新持有锁
        }
        
        // 临界区操作：生产项
        item = next_item++;
        buffer[in] = item;
        in = (in + 1) % BUFSIZE;
        counter++;  
        
        if (task_log[item][0] == -1) {
            task_log[item][0] = i;
            produced++;
            printf("<P%d,%d>", i, item); 
        } else {
            printf("ERROR: 重复生产项 %d\n", item);
            // 回滚缓冲区状态
            counter--;
            in = (in - 1 + BUFSIZE) % BUFSIZE;
        }
        
        // 解锁并通知消费者有新数据
        pthread_mutex_unlock(&mutex);
        pthread_cond_signal(&not_empty);
    }
    pthread_exit(NULL); 
}

消费者线程 #

void *consumer(void *arg)
{
    int i = *(int *)arg;
    int item;

    while (alive) {
        // 获取互斥锁
        pthread_mutex_lock(&mutex);
        
        // 等待缓冲区有数据
        while (counter <= 0) {
            pthread_cond_wait(&not_empty, &mutex);
        }
        
        // 临界区操作：消费项
        item = buffer[out];
        out = (out + 1) % BUFSIZE;    
        counter--; 
        
        if (task_log[item][0] == -1) {
            printf("ERROR: 消费未生产项 %d\n", item);
            // 回滚缓冲区状态
            counter++;
            out = (out - 1 + BUFSIZE) % BUFSIZE;
        } else if (task_log[item][1] == -1) {
            task_log[item][1] = i;
            consumed++;
            printf("<C%d,%d>\n", i, item);   
        } else {
            printf("ERROR: 重复消费项 %d\n", item);
            // 回滚缓冲区状态
            counter++;
            out = (out - 1 + BUFSIZE) % BUFSIZE;
        }
        
        // 解锁并通知生产者有空位
        pthread_mutex_unlock(&mutex);
        pthread_cond_signal(&not_full);
    }
    pthread_exit(NULL);
}

关键修复点说明 #

1. 统一互斥锁：用单个pthread_mutex_t保护所有共享资源的访问，确保同一时间只有一个线程进入临界区。
2. 条件变量替代忙等：pthread_cond_wait会释放锁并进入休眠，被唤醒时自动重新获取锁，避免CPU空转，同时解决了锁外判断的竞态问题。
3. 完整的错误回滚：生产/消费出错时，回滚counter和in/out的状态，保证缓冲区一致性。
4. 原子性与线程安全：所有共享变量的修改都在互斥锁保护下，避免数据竞争。
5. 正确的通知机制：生产完成后通知消费者，消费完成后通知生产者，保证线程间的高效协作。

内容的提问来源于stack exchange，提问作者lele