c++生产者消费者者模式笔记-1阻塞问题

生产者消费者模式是并发编程的核心模式之一核心是想要提高程序的运行效率。这里记录一下自己的思考使用通俗易懂的语言和以日志记录为例解读生产者消费者模式并实现生产者消费者模式。将生产者消费者模式的核心内容划分为三个问题阻塞问题、内存积压问题、cpu空转问题。这里是第一章阻塞问题。阻塞问题生产者消费者模式首先就是要解决相互阻塞的问题。在编程过程中循环是普遍存在的循环内部普遍来说是顺序执行顺序结构可以分为上下两个部分。程序上方的输出数据往往是程序下方的输入从数据的角度看程序上方就是生产过程称为生产者程序下方就是消费过程称为消费者。这种结构下生产者和消费者之间必然相互等待如果两个两者耗时较长那么就会导致程序运行效率降低这就称之为程序之间相互阻塞。实际场景日志记录系统就是一个典型的生产者消费者模式。项目运行时需要循环记录日志在循环内部程序上方是生成日志程序下方是写入日志文件。所以生产者就是生成字符串消费者是写入日志文件。代码实现程序在同一个线程中循环执行循环体中使用顺序结构生产一个字符串紧接着在磁盘中写入一个字符串。这就是同步耦合实现这样会导致生产者生产和消费者相互等待即相互阻塞。void sync_log() { std::cout同步耦合日志系统std::endl; std::ofstream log_file; std::string log_pathlog1.txt; std::chrono::high_resolution_clock::time_point start std::chrono::high_resolution_clock::now(); for(int i0;i10;i) { log_file.open(log_path,std::ios::app); //cpu内存处理数据 std::chrono::high_resolution_clock::time_point t1 std::chrono::high_resolution_clock::now(); //大量数据 std::string large_data(2048, x); // 2KB 数据 //时间记录 auto time std::chrono::system_clock::now(); auto time_t std::chrono::system_clock::to_time_t(time); std::string content_timestd::ctime(time_t); std::cout日志记录时刻content_time; std::string content[content_time] large_data; std::chrono::high_resolution_clock::time_point t2 std::chrono::high_resolution_clock::now(); //用户缓存区写日志 log_filecontentstd::endl; //写磁盘 log_file.close(); std::chrono::high_resolution_clock::time_point t3 std::chrono::high_resolution_clock::now(); std::cout处理业务耗时std::chrono::duration_caststd::chrono::microseconds(t2-t1).count()微秒std::endl; std::cout写日志耗时std::chrono::duration_caststd::chrono::microseconds(t3-t2).count()微秒std::endlstd::endl; } // 写入磁盘 std::chrono::high_resolution_clock::time_point end std::chrono::high_resolution_clock::now(); std::cout生成日志和io相互阻塞std::endl; std::cout耗时std::chrono::duration_caststd::chrono::microseconds(end-start).count()msstd::endl; } int main() { sync_log(); return 0; }运行结果同步耦合日志系统 日志记录时刻Fri May 15 10:20:42 2026 处理业务耗时2101微秒 写日志耗时141微秒 日志记录时刻Fri May 15 10:20:42 2026 处理业务耗时68微秒 写日志耗时112微秒 日志记录时刻Fri May 15 10:20:42 2026 处理业务耗时155微秒 写日志耗时87微秒 日志记录时刻Fri May 15 10:20:42 2026 处理业务耗时86微秒 写日志耗时111微秒 日志记录时刻Fri May 15 10:20:42 2026 处理业务耗时78微秒 写日志耗时100微秒 日志记录时刻Fri May 15 10:20:42 2026 处理业务耗时56微秒 写日志耗时95微秒 日志记录时刻Fri May 15 10:20:42 2026 处理业务耗时59微秒 写日志耗时68微秒 日志记录时刻Fri May 15 10:20:42 2026 处理业务耗时45微秒 写日志耗时85微秒 日志记录时刻Fri May 15 10:20:42 2026 处理业务耗时52微秒 写日志耗时88微秒 日志记录时刻Fri May 15 10:20:42 2026 处理业务耗时66微秒 写日志耗时107微秒 生成日志和io相互阻塞 耗时8398ms结果分析每一次循环都需要生成一个字符串除了在第一次循环其他每一次都需要等待写入日志完成后才能生成字符串同理写入磁盘也类似包括第一次循环每一次都需要等待生成字符串完成后才能写入磁盘。这就是生产者和消费者速度相互阻塞生成字符串是生产者写入磁盘是消费者两者相互等待导致程序整体效率低。解决阻塞生产者和消费者相互阻塞如何解决同步耦合深入分析阻塞问题的根本原因在于同步耦合即调用方必须等待被调用方完成才能继续执行的模式。在生产者消费者视角下看同步耦合来自于生产者和消费者在同一个线程、处于同一个顺序结构中上下游的位置当循环时前后者就会相互制约。异步解耦合既然阻塞时同步耦合导致的那么解决就需要异步解耦合。即调用方不需要等待被调用方完成就可以继续执行的模式。在生产者消费者视角下看就需要将生产者和消费者放在两个独立的线程中运行这样两者不相互依赖同时运行。代码实现分离生产者和消费者线程生产者只需要生成日志字符串不需要关心写入日志文件消费者只需要写入日志文件不需要关心生成日志字符串但是这样又会产生新的问题生产者生产数据消费者消费数据两者如何关联起来呢为了解决这些问题我们可以创建共享变量作为通信方式生产者和消费者能同时访问到这个变量生产者生产数据后将数据放入共享变量消费者消费数据时从共享变量中取出数据进行消费由于是多线程访问共享变量所以需要引入互斥锁当生产者生产数据时消费者无法访问共享变量当消费者消费数据时生产者无法访问共享变量避免了数据竞争保证了数据安全这样生产者和消费者就互不依赖同时运行解决了阻塞问题这样就是异步解耦合程序完成相同的功能但是效率更高。void sync_log() { std::cout同步耦合日志系统std::endl; std::ofstream log_file; std::string log_pathlog1.txt; std::chrono::high_resolution_clock::time_point start std::chrono::high_resolution_clock::now(); for(int i0;i10;i) { log_file.open(log_path,std::ios::app); //cpu内存处理数据 // std::chrono::high_resolution_clock::time_point t1 std::chrono::high_resolution_clock::now(); //大量数据 std::string large_data(2048, x); // 2KB 数据 //时间记录 auto time std::chrono::system_clock::now(); auto time_t std::chrono::system_clock::to_time_t(time); std::string content_timestd::ctime(time_t); // std::cout日志记录时刻content_time; std::string content[content_time] large_data; std::chrono::high_resolution_clock::time_point t2 std::chrono::high_resolution_clock::now(); //用户缓存区写日志 log_filecontentstd::endl; //写磁盘 log_file.close(); // std::chrono::high_resolution_clock::time_point t3 std::chrono::high_resolution_clock::now(); // std::cout处理业务耗时std::chrono::duration_caststd::chrono::microseconds(t2-t1).count()微秒std::endl; // std::cout写日志耗时std::chrono::duration_caststd::chrono::microseconds(t3-t2).count()微秒std::endlstd::endl; } // 写入磁盘 std::chrono::high_resolution_clock::time_point end std::chrono::high_resolution_clock::now(); std::cout生成日志和io相互阻塞std::endl; std::cout耗时std::chrono::duration_caststd::chrono::microseconds(end-start).count()msstd::endl; } void async_log() { std::cout异步解耦日志系统,不再相互阻塞std::endl; std::ofstream log_file; std::string log_pathlog2.txt; // 缓存 std::dequestd::string log_buffer; // 停止标志 std::atomicbool stop_flag(false); // 加锁避免数据竞争 std::mutex log_mutex; // cpu内存处理数据 auto log_data_func[log_buffer,stop_flag,log_mutex](){ int count1; std::chrono::high_resolution_clock::time_point start std::chrono::high_resolution_clock::now(); while(true) { // std::chrono::high_resolution_clock::time_point t1 std::chrono::high_resolution_clock::now(); //创建数据 std::string large_data(2048, x); // 4KB 数据 // std::this_thread::sleep_for(std::chrono::milliseconds(100)); //时间记录 auto time std::chrono::system_clock::now(); auto time_t std::chrono::system_clock::to_time_t(time); std::string content_timestd::ctime(time_t); // std::cout生成日志content_timestd::endl; std::string content[content_time] large_data; //写入缓存 std::unique_lockstd::mutex lock(log_mutex); log_buffer.push_back(content); lock.unlock(); // std::chrono::high_resolution_clock::time_point t2 std::chrono::high_resolution_clock::now(); // std::cout处理业务耗时std::chrono::duration_caststd::chrono::microseconds(t2-t1).count()微秒std::endlstd::endl; count; if(count10) { std::chrono::high_resolution_clock::time_point t_10 std::chrono::high_resolution_clock::now(); std::cout生成10条日志耗时std::chrono::duration_caststd::chrono::microseconds(t_10-start).count()msstd::endl; } if (stop_flag) { break; } }; std::cout生成日志结束std::endl; std::cout总共生成数据量count条std::endl; }; std::thread thread_log_data(log_data_func); // 日志写入磁盘 auto log_disk_func[log_buffer,log_file,log_path,stop_flag,log_mutex](){ //计时开始 std::chrono::high_resolution_clock::time_point start std::chrono::high_resolution_clock::now(); int count1; while(true) { //判断退出循环 if (count10) { break; } if(!log_buffer.empty()) { std::chrono::high_resolution_clock::time_point t1 std::chrono::high_resolution_clock::now(); //打开文件 log_file.open(log_path,std::ios::app); //写日志 std::unique_lockstd::mutex lock(log_mutex); auto datalog_buffer.front(); log_buffer.pop_front(); lock.unlock(); log_filedatastd::endl; //关闭文件 log_file.close(); std::chrono::high_resolution_clock::time_point t2 std::chrono::high_resolution_clock::now(); // std::cout写日志耗时std::chrono::duration_caststd::chrono::microseconds(t2-t1).count()微秒std::endl; count; } } //通知生产进程结束 stop_flag true; //计时结束 std::chrono::high_resolution_clock::time_point end std::chrono::high_resolution_clock::now(); std::coutio耗时std::chrono::duration_caststd::chrono::microseconds(end-start).count()msstd::endl; }; std::thread thread_log_disk(log_disk_func); thread_log_data.join(); thread_log_disk.join(); } int main() { // 同步阻塞问题 sync_log(); std::cout----------------std::endl; async_log(); return 0; }运行结果同步耦合日志系统 生成日志和io相互阻塞 耗时4978ms ---------------- 异步解耦日志系统,不再相互阻塞 生成10条日志耗时46ms io耗时3864ms 生成日志结束 总共生成数据量600条结果分析同步耦合日志系统生成日志和io相互阻塞记录10条日志到文件共耗时4978ms异步解耦日志系统生成日志和io不再相互阻塞记录10条日志到文件共耗时3864ms异步解耦合日志系统当10条日志记录结束时总共生成数据量都已经生成600条日志了同样是记录10条日志相互阻塞的方式耗时4978ms非阻塞的方式耗时3864ms非阻塞的方式耗时明显更短同样是记录10条日志相互阻塞的方式只生成了10条日志而非阻塞的方式生成了600条非阻塞的方式生成了更多的数据结论异步解耦合的方式以更短的时间生成了达到同样的记录效果并且生成了更多的数据说明异步解耦合的方式更高效