案例1: 最基础的生产者消费者模型#include stdio.h // 标准输入输出printf #include pthread.h // POSIX线程库线程创建/互斥锁/条件变量 #include unistd.h // 睡眠函数sleep #include stdlib.h // 标准库内存分配/退出等此处暂未用到 #define BUF_SIZE 3 // 缓冲区大小最多存放3个数据 int g_buf[BUF_SIZE]; // 共享缓冲区存储生产/消费的数据 int g_idx 0; // 缓冲区当前已存元素个数共享资源需互斥保护 // 同步原语互斥锁两个条件变量分别协调生产者/消费者 pthread_mutex_t g_mutex; // 互斥锁保护共享资源g_buf/g_idx的排他访问 pthread_cond_t g_cond_prod; // 生产者条件变量缓冲区有空位时唤醒生产者 pthread_cond_t g_cond_cons; // 消费者条件变量缓冲区有数据时唤醒消费者 // 生产者线程函数循环往缓冲区放入5个数据 // 参数arg - 线程入参此处未使用返回值线程退出状态此处返回NULL void* producer(void* arg) { // 生产5个数据循环5次 for (int i 0; i 5; i) { // 1. 加互斥锁保护共享资源g_buf/g_idx防止多线程同时修改 // 加锁后其他线程调用pthread_mutex_lock会阻塞直到当前线程解锁 pthread_mutex_lock(g_mutex); // 2. 检查缓冲区是否满如果满了生产者需要等待消费者取数据 // 注意必须用while而非if避免「虚假唤醒」条件变量被唤醒后条件可能已不满足 // 比如多个生产者被唤醒时缓冲区可能已被其他生产者占满 while (g_idx BUF_SIZE) { // pthread_cond_wait做两件事 // a. 原子操作解锁g_mutex 阻塞当前线程等待被唤醒 // b. 被唤醒后继续从这个代码开始执行, 重新加锁g_mutex然后退出wait继续执行 pthread_cond_wait(g_cond_prod, g_mutex); } // 3. 生产数据往缓冲区放入当前数值i g_buf[g_idx] i; // g_idx先赋值再自增等价于g_buf[g_idx]i; g_idx; // 打印生产信息pthread_self()获取当前线程IDg_idx是当前缓冲区数据量 printf(生产者[%ld]放入%d缓冲区数量%d\n, pthread_self(), i, g_idx); // 4. 唤醒消费者缓冲区已有数据通知消费者可以取数据了 // pthread_cond_signal唤醒等待该条件变量的任意一个线程此处只有1个消费者 // 若有多个消费者可用pthread_cond_broadcast唤醒所有 pthread_cond_signal(g_cond_cons); // 5. 解锁互斥锁释放共享资源让其他线程消费者可以访问 pthread_mutex_unlock(g_mutex); // 模拟生产耗时比如生产一个数据需要1秒让出CPU让消费者有机会执行 sleep(1); } // 线程正常退出返回NULL return NULL; } // 消费者线程函数循环从缓冲区取出5个数据 // 参数arg - 线程入参此处未使用返回值线程退出状态此处返回NULL void* consumer(void* arg) { // 消费5个数据与生产者生产数量一致 for (int i 0; i 5; i) { // 1. 加互斥锁保护共享资源防止与生产者同时修改g_buf/g_idx pthread_mutex_lock(g_mutex); // 2. 检查缓冲区是否空如果空了消费者需要等待生产者放数据 // 同样用while避免虚假唤醒比如多个消费者被唤醒时缓冲区已空 while (g_idx 0) { // 阻塞当前线程等待生产者的唤醒信号g_cond_cons // 等待期间会自动解锁g_mutex被唤醒后重新加锁 pthread_cond_wait(g_cond_cons, g_mutex); } // 3. 消费数据从缓冲区取出最后一个数据 int val g_buf[--g_idx]; // g_idx先自减再赋值等价于g_idx--; valg_buf[g_idx]; // 打印消费信息当前线程ID 取出的数据 剩余缓冲区数量 printf(消费者[%ld]取出%d缓冲区数量%d\n, pthread_self(), val, g_idx); // 4. 唤醒生产者缓冲区有空位了通知生产者可以继续生产 pthread_cond_signal(g_cond_prod); // 5. 解锁互斥锁释放共享资源让生产者可以访问 pthread_mutex_unlock(g_mutex); // 模拟消费耗时比如消费一个数据需要1秒让出CPU让生产者有机会执行 sleep(1); } // 线程正常退出返回NULL return NULL; } int main() { // 定义线程ID分别存储生产者/消费者线程的ID pthread_t tid_prod, tid_cons; // 初始化同步原语必须先初始化才能使用 // 第二个参数为NULL使用默认属性 pthread_mutex_init(g_mutex, NULL); // 初始化互斥锁 pthread_cond_init(g_cond_prod, NULL); // 初始化生产者条件变量 pthread_cond_init(g_cond_cons, NULL); // 初始化消费者条件变量 // 创建线程 // 参数线程ID指针、线程属性NULL默认、线程函数、函数入参NULL pthread_create(tid_prod, NULL, producer, NULL); // 创建生产者线程 pthread_create(tid_cons, NULL, consumer, NULL); // 创建消费者线程 // 等待线程结束主线程阻塞直到生产者/消费者线程执行完毕 // 第二个参数为NULL不接收线程的返回值 pthread_join(tid_prod, NULL); // 等待生产者线程退出 pthread_join(tid_cons, NULL); // 等待消费者线程退出 // 销毁同步原语释放系统资源避免内存泄漏 pthread_mutex_destroy(g_mutex); // 销毁互斥锁 pthread_cond_destroy(g_cond_prod); // 销毁生产者条件变量 pthread_cond_destroy(g_cond_cons); // 销毁消费者条件变量 return 0; // 主线程正常退出 }案例2: 信号量我就用信号量最经典的**“限流”**场景来举例只有 3 个窗口的银行来了 10 个客户办理业务。在这个例子中你会看到信号量充当了“叫号机”和“柜员空闲数”。只有 3 个线程能同时运行办理业务。其他 7 个线程必须阻塞等待直到有人办完离开。信号量Semaphore-CSDN博客https://blog.csdn.net/Howrun777/article/details/157183697?sharetypeblogdetailsharerId157183697sharereferPCsharesourceHowrun777spm1011.2480.3001.8118代码实现这个代码不需要 pthread_cond也不需要 while 循环检查逻辑非常直线条。#include stdio.h #include pthread.h #include unistd.h #include stdlib.h #include semaphore.h // 必须引入信号量头文件 #define NUM_WINDOWS 3 // 只有3个柜台窗口 #define NUM_CUSTOMERS 10 // 来了10个客户 sem_t sem_windows; // 定义信号量 // 客户线程函数 void* customer(void* arg) { int id *(int*)arg; free(arg); // 释放传进来的内存 printf(--- 客户 %d 到达银行正在排队...\n, id); // [P操作] 申请资源 // 信号量 0 则减1并进入 // 信号量 0 则阻塞在这里睡觉 sem_wait(sem_windows); // --- 临界区能走到这里说明拿到窗口了 --- printf( 客户 %d 抢到了窗口正在办理业务...\n, id); sleep(2); // 模拟办理业务耗时 2秒 printf( 客户 %d 办理完毕离开窗口\n, id); // [V操作] 释放资源 // 信号量 1如果有人在排队会唤醒其中一个 sem_post(sem_windows); return NULL; } int main() { pthread_t tids[NUM_CUSTOMERS]; // 初始化信号量 // 第二个参数 0 表示线程间共享 // 第三个参数 NUM_WINDOWS (3) 是信号量的初始值表示有3个空位 sem_init(sem_windows, 0, NUM_WINDOWS); // 创建 10 个客户线程 for (int i 0; i NUM_CUSTOMERS; i) { int* p_id malloc(sizeof(int)); // 动态分配内存传参 *p_id i; pthread_create(tids[i], NULL, customer, p_id); // 稍微间隔一下让打印好看点不是必须的 usleep(100000); } // 等待所有线程结束 for (int i 0; i NUM_CUSTOMERS; i) { pthread_join(tids[i], NULL); } // 销毁信号量 sem_destroy(sem_windows); return 0; }案例3: 环形队列生产者消费者模型,环形队列 ,信号量, 且单独实现环形队列类这是一个非常好的练习能够帮助你理解如何从面向过程 (C语言)转向面向对象 (C)的思维方式。在 C 中我们通常会利用RAII (资源获取即初始化)技术来管理锁和信号量利用模板 (Template)来让队列支持任意数据类型。以下是完整的 C 实现包含三个部分Semaphore 类将 POSIX sem_t 封装成 C 类为了 RAII 管理。RingQueue 类线程安全的泛型环形队列。主程序生产者与消费者逻辑。完整代码 (C Standard 11/14/17/20 兼容)为了保证代码在 Linux/Unix 环境下与之前的 C 代码逻辑一致底层依然使用 semaphore.h但用 C 类进行了封装。#include iostream // 标准输入输出cout/endl #include vector // 动态数组用于实现环形队列的缓冲区 #include thread // C11线程库线程创建/线程ID获取 #include mutex // C11互斥锁lock_guard #include semaphore.h // POSIX信号量sem_init/sem_wait/sem_post #include unistd.h // sleep函数模拟生产/消费耗时 // --------------------------------------------------------- // 1. 信号量封装类 (RAII风格) // 核心设计利用C的构造/析构自动管理信号量生命周期避免手动销毁导致的资源泄漏 // --------------------------------------------------------- class Semaphore { private: sem_t _sem; // POSIX信号量原生对象 public: // 构造函数初始化信号量 // 参数value信号量初始值0 // sem_init参数说明 // _sem要初始化的信号量对象 // 0信号量作用域0进程内线程共享非0进程间共享 // value信号量初始值可用资源数 Semaphore(int value) { sem_init(_sem, 0, value); } // 析构函数销毁信号量RAII核心对象销毁时自动释放资源 // 避免手动调用sem_destroy遗漏导致的系统资源泄漏 ~Semaphore() { sem_destroy(_sem); } // P操作 (Wait/减操作)申请资源若无可用则阻塞 // 作用信号量值-1若值0当前线程阻塞直到有其他线程调用signal() void wait() { sem_wait(_sem); } // V操作 (Post/Signal/加操作)释放资源唤醒阻塞线程 // 作用信号量值1若有线程阻塞在wait()则唤醒其中一个 void signal() { sem_post(_sem); } }; // --------------------------------------------------------- // 2. 线程安全的环形队列类 (模板类) // 核心设计 // - 环形队列通过head/tail取模实现缓冲区循环复用 // - 同步机制信号量控制空/数据数量 互斥锁保护临界区 // - 线程安全所有对head/tail/buffer的操作都加锁保护 // --------------------------------------------------------- template typename T // 模板类支持任意类型的数据存储 class RingQueue { private: std::vectorT _buffer; // 环形缓冲区替代原生数组自动管理内存无需手动new/delete int _capacity; // 队列总容量缓冲区最大可存储元素数 int _head; // 读索引消费者使用指向下一个要取出的元素位置 int _tail; // 写索引生产者使用指向下一个要写入的元素位置 // 同步工具保证多线程访问的安全性和顺序性 std::mutex _mtx; // 互斥锁保护head/tail/_buffer的原子操作临界区 Semaphore _sem_empty; // 信号量表示缓冲区的「空闲位置数」生产者可用资源 Semaphore _sem_data; // 信号量表示缓冲区的「已存储数据数」消费者可用资源 public: // 构造函数初始化环形队列的容量和同步工具 // 参数cap队列容量必须0 RingQueue(int cap) : _capacity(cap), // 初始化队列总容量 _buffer(cap), // 初始化缓冲区大小为cap _head(0), // 读索引初始化为0 _tail(0), // 写索引初始化为0 _sem_empty(cap), // 初始空闲位置数 总容量队列空 _sem_data(0) // 初始数据数 0队列空 {} // 禁止拷贝构造和赋值运算符重载 // 原因队列包含互斥锁、信号量等不可拷贝的同步对象拷贝会导致未定义行为 RingQueue(const RingQueue) delete; RingQueue operator(const RingQueue) delete; // 入队操作 (生产者调用)将数据写入环形队列 // 参数data要写入队列的数据const 避免拷贝提高效率 void push(const T data) { // 1. P操作申请「空闲位置」资源 // 若队列已满_sem_empty值0生产者线程阻塞直到消费者取出数据释放空位 _sem_empty.wait(); // 2. 加互斥锁应对多个生产者, 保护临界区_tail/_buffer的修改 // lock_guard是RAII风格锁作用域结束时自动解锁避免手动解锁遗漏导致死锁 { // 局部作用域限制lock_guard的生命周期解锁时机更精准 std::lock_guardstd::mutex lock(_mtx); // --- 临界区开始仅允许一个线程执行 --- _buffer[_tail] data; // 将数据写入当前写索引位置 _tail (_tail 1) % _capacity; // 写索引后移取模实现「环形」到末尾则回到0 // --- 临界区结束 --- } // 出作用域lock_guard析构_mtx自动解锁 // 3. V操作释放「数据」资源通知消费者有新数据可消费 // _sem_data值1若有消费者阻塞在pop()的wait()则唤醒其中一个 _sem_data.signal(); } // 出队操作 (消费者调用)从环形队列取出数据 // 参数out_data输出参数用于接收取出的数据指针避免返回值拷贝 void pop(T* out_data) { // 1. P操作申请「数据」资源 // 若队列为空_sem_data值0消费者线程阻塞直到生产者写入数据 _sem_data.wait(); // 2. 加互斥锁应对多个消费者, 保护临界区_head/_buffer的读取 { std::lock_guardstd::mutex lock(_mtx); // --- 临界区开始 --- *out_data _buffer[_head]; // 从当前读索引位置取出数据 _head (_head 1) % _capacity; // 读索引后移取模实现环形 // --- 临界区结束 --- } // 3. V操作释放「空闲位置」资源通知生产者有新空位可写入 // _sem_empty值1若有生产者阻塞在push()的wait()则唤醒其中一个 _sem_empty.signal(); } }; // --------------------------------------------------------- // 3. 主程序逻辑生产者-消费者模型测试 // --------------------------------------------------------- // 全局的环形队列实例容量为5的整数队列生产者和消费者共享 RingQueueint g_queue(5); // 生产者线程函数无限循环生产数据并写入队列 void producer() { int i 0; // 生产的数据值从0开始递增 while (true) { // 死循环持续生产 // 打印生产信息输出线程ID和生产的数据值 std::cout 生产者 [ std::this_thread::get_id() ] 生产: i std::endl; // 将数据推入队列若队列满则自动阻塞无需手动判断 g_queue.push(i); i; // 下一个要生产的数据值 sleep(1); // 模拟生产耗时1秒生产一个数据 } } // 消费者线程函数无限循环从队列取出数据并消费 void consumer() { while (true) { // 死循环持续消费 int data; // 存储从队列取出的数据 // 从队列取出数据若队列为空则自动阻塞无需手动判断 g_queue.pop(data); // 打印消费信息输出线程ID和消费的数据值 区分消费日志 std::cout 消费者 [ std::this_thread::get_id() ] 消费: data std::endl; sleep(2); // 模拟消费耗时2秒消费一个数据比生产慢队列会逐渐填满 } } int main() { // 创建生产者线程和消费者线程 // 线程创建后自动执行对应的线程函数producer/consumer std::thread t_prod(producer); std::thread t_cons(consumer); // 等待线程结束阻塞主线程直到生产者/消费者线程退出 // 注由于线程函数是死循环此处join()永远不会返回程序会一直运行 t_prod.join(); t_cons.join(); return 0; }注意: 如果是单生产者-单消费者模型, 不需要加锁, 可以只使用信号量完成C 写法的关键点类封装 (Encapsulation)使用者不需要手动调用 sem_init 或 sem_destroy这些都在 Semaphore 和 RingQueue 的构造函数和析构函数中自动完成。使用者不需要知道内部用的是 sem_wait 还是 pthread_mutex_lock接口只有简单的 push 和 pop。RAII 锁管理 (std::lock_guard)在 C 中你必须记得手动 unlock。如果在解锁前函数 return 了或者抛出了错误就会造成死锁。在 C 中std::lock_guardstd::mutex lock(_mtx); 这行代码创建了一个对象。当这个对象超出作用域也就是遇到右大括号 }时它的析构函数会自动释放锁。这是 C 多线程最核心的安全机制。C 线程互斥锁 lock_guardhttps://blog.csdn.net/Howrun777/article/details/157184216?spm1001.2014.3001.5501案例4: SPSC 无锁队列实现 (工业级写法)这才是“不需要锁”的终极形态。我们利用 std::atomic 的 acquire / release 语义来替代锁和信号量。#include vector #include atomic #include thread #include iostream template typename T class SPSCLockFreeQueue { private: std::vectorT _buffer; int _capacity; // 使用 atomic 替代普通的 int并移除 Mutex 和 Semaphore std::atomicint _head; std::atomicint _tail; public: SPSCLockFreeQueue(int cap) : _capacity(cap 1), _buffer(cap 1), _head(0), _tail(0) {} // 注意实际容量是 cap多留一个空位用于判断队满head (tail 1) % size // 生产者调用 bool push(const T data) { int current_tail _tail.load(std::memory_order_relaxed); int next_tail (current_tail 1) % _capacity; // 检查是否已满需要读取 _head // memory_order_acquire: 保证看到消费者对 _head 的最新修改 if (next_tail _head.load(std::memory_order_acquire)) { return false; // 队列满 } _buffer[current_tail] data; // 更新 _tail // memory_order_release: 保证在此之前的写入buffer data对读取 tail 的线程可见 _tail.store(next_tail, std::memory_order_release); return true; } // 消费者调用 bool pop(T* out_data) { int current_head _head.load(std::memory_order_relaxed); // 检查是否为空需要读取 _tail // memory_order_acquire: 保证看到生产者对 _tail 和 buffer 的最新修改 if (current_head _tail.load(std::memory_order_acquire)) { return false; // 队列空 } *out_data _buffer[current_head]; // 更新 _head // memory_order_release: 通知生产者这个位置已经空出来了 _head.store((current_head 1) % _capacity, std::memory_order_release); return true; } };针对你之前的代码在单生产者-单消费者模式下可以删掉 mutex。因为 semaphore 的屏障作用已经足够保证安全。你是对的。更深层的逻辑如果你觉得 mutex 慢那么 semaphore 也很慢。最佳实践真正的 SPSC 优化方案是不使用任何阻塞工具锁或信号量而是利用 std::atomic 配合 CPU 缓存一致性协议实现真正的“零拷贝、零阻塞”。