从生产者-消费者模型到线程池：手把手用pthread实现你的第一个Linux C并发框架

从生产者-消费者模型到线程池手把手用pthread实现你的第一个Linux C并发框架在Linux系统编程中多线程开发是提升程序性能的重要手段。但直接使用原生线程API往往面临资源管理复杂、性能不稳定等问题。本文将带你从经典的生产者-消费者模型出发逐步构建一个功能完整的线程池框架解决实际开发中的并发任务调度难题。1. 生产者-消费者模型基础实现生产者-消费者模型是多线程编程的经典范式特别适合处理异步任务。我们先实现一个基础版本#include pthread.h #include semaphore.h #define QUEUE_SIZE 10 typedef struct { int buffer[QUEUE_SIZE]; int in; int out; sem_t empty; sem_t full; pthread_mutex_t mutex; } Queue; void queue_init(Queue *q) { q-in q-out 0; sem_init(q-empty, 0, QUEUE_SIZE); sem_init(q-full, 0, 0); pthread_mutex_init(q-mutex, NULL); } void enqueue(Queue *q, int item) { sem_wait(q-empty); pthread_mutex_lock(q-mutex); q-buffer[q-in] item; q-in (q-in 1) % QUEUE_SIZE; pthread_mutex_unlock(q-mutex); sem_post(q-full); } int dequeue(Queue *q) { sem_wait(q-full); pthread_mutex_lock(q-mutex); int item q-buffer[q-out]; q-out (q-out 1) % QUEUE_SIZE; pthread_mutex_unlock(q-mutex); sem_post(q-empty); return item; }这个实现有几个关键点需要注意环形缓冲区通过模运算实现循环队列避免频繁内存分配双信号量控制empty控制生产者full控制消费者互斥锁保护确保对缓冲区的操作是原子的提示在实际项目中建议将buffer改为动态数组通过realloc实现队列扩容2. 任务抽象与线程池设计将生产者-消费者模型升级为线程池首先需要抽象任务概念typedef void (*task_func)(void *); typedef struct { task_func function; void *arg; } Task; typedef struct { Queue task_queue; pthread_t *threads; int thread_count; int shutdown; } ThreadPool;线程池的核心参数对比如下参数说明推荐值thread_count工作线程数量CPU核心数×2queue_size任务队列长度100-1000stack_size线程栈大小默认(2MB)线程池初始化函数需要特别注意错误处理int thread_pool_init(ThreadPool *pool, int thread_count) { if(pthread_mutex_init(pool-lock, NULL) ! 0) { return -1; } pool-threads (pthread_t*)malloc(thread_count * sizeof(pthread_t)); if(!pool-threads) { pthread_mutex_destroy(pool-lock); return -1; } queue_init(pool-task_queue); pool-shutdown 0; for(int i 0; i thread_count; i) { if(pthread_create(pool-threads[i], NULL, worker_thread, pool) ! 0) { // 清理已创建的线程 pool-shutdown 1; for(int j 0; j i; j) { pthread_join(pool-threads[j], NULL); } free(pool-threads); pthread_mutex_destroy(pool-lock); return -1; } } return 0; }3. 工作线程与任务调度工作线程是线程池的核心执行单元其典型实现如下void *worker_thread(void *arg) { ThreadPool *pool (ThreadPool*)arg; while(1) { Task task; if(queue_dequeue(pool-task_queue, task) 0) { if(pool-shutdown) break; task.function(task.arg); } } return NULL; }任务提交接口需要考虑多种使用场景int thread_pool_submit(ThreadPool *pool, task_func func, void *arg, int timeout_ms) { if(pool-shutdown) return -1; Task task {func, arg}; struct timespec ts; if(timeout_ms 0) { clock_gettime(CLOCK_REALTIME, ts); ts.tv_nsec (timeout_ms % 1000) * 1000000; ts.tv_sec timeout_ms / 1000 ts.tv_nsec / 1000000000; ts.tv_nsec % 1000000000; } if(timeout_ms 0) { return queue_enqueue(pool-task_queue, task); } else { return queue_timed_enqueue(pool-task_queue, task, ts); } }4. 高级特性与性能优化4.1 动态线程调整智能线程池可以根据负载动态调整线程数量void *monitor_thread(void *arg) { ThreadPool *pool (ThreadPool*)arg; while(!pool-shutdown) { sleep(5); // 每5秒检查一次 int queue_size queue_size(pool-task_queue); int active_threads get_active_thread_count(pool); if(queue_size active_threads * 2 active_threads pool-max_threads) { // 增加线程 pthread_t thread; pthread_create(thread, NULL, worker_thread, pool); pthread_detach(thread); } else if(queue_size active_threads / 2 active_threads pool-min_threads) { // 减少线程 send_exit_signal_to_thread(pool); } } return NULL; }4.2 任务优先级支持通过优先队列实现任务优先级typedef struct { Task *tasks; int capacity; int size; } PriorityQueue; void priority_queue_init(PriorityQueue *pq, int cap) { pq-tasks malloc(cap * sizeof(Task)); pq-capacity cap; pq-size 0; } void priority_queue_push(PriorityQueue *pq, Task task, int prio) { if(pq-size pq-capacity) { // 扩容逻辑 } // 根据prio插入到合适位置 // ...优先级队列实现... }4.3 性能优化技巧线程局部存储为每个工作线程维护独立的任务缓存__thread Task local_task_cache[LOCAL_CACHE_SIZE];批量任务提交减少锁竞争int submit_batch(ThreadPool *pool, Task *tasks, int count) { pthread_mutex_lock(pool-lock); // 批量添加任务 pthread_mutex_unlock(pool-lock); }无锁队列在高并发场景下考虑无锁实现typedef struct { Task *buffer; volatile int head; volatile int tail; } LockFreeQueue;5. 完整实现与测试案例下面是一个完整的日志处理系统示例// 日志任务结构 typedef struct { char *message; time_t timestamp; } LogTask; void process_log(void *arg) { LogTask *log (LogTask*)arg; printf([%.24s] %s\n, ctime(log-timestamp), log-message); free(log-message); free(log); } int main() { ThreadPool pool; thread_pool_init(pool, 4); for(int i 0; i 100; i) { LogTask *task malloc(sizeof(LogTask)); task-timestamp time(NULL); asprintf(task-message, Log message %d, i); thread_pool_submit(pool, process_log, task, 0); } // 等待所有任务完成 while(queue_size(pool.task_queue) 0) { usleep(100000); } thread_pool_shutdown(pool); return 0; }测试时需要注意的几个指标吞吐量每秒处理的任务数延迟任务从提交到执行的时间CPU利用率线程池对计算资源的利用效率内存占用随着任务增长的内存消耗6. 常见问题排查指南6.1 死锁问题线程池中常见的死锁场景任务间相互等待A任务等待B任务的输出而B任务在队列中未被调度资源竞争多个任务竞争同一把锁线程饥饿高优先级任务独占线程资源排查工具gdbthread apply all bt查看所有线程堆栈valgrind检测锁的使用情况pstack实时查看线程状态6.2 性能瓶颈典型性能问题及解决方案现象可能原因解决方案CPU利用率低任务I/O密集增加线程数吞吐量上不去锁竞争激烈减小锁粒度或使用无锁结构延迟波动大任务执行时间不均实现任务优先级6.3 内存管理线程池中的内存注意事项任务参数内存确保任务参数在任务执行期间有效线程栈大小对于递归或大数据量任务调整栈大小pthread_attr_t attr; pthread_attr_setstacksize(attr, 8*1024*1024); // 8MB7. 生产环境最佳实践在实际项目中应用线程池时有几个经验值得分享监控集成为线程池添加Prometheus监控指标// 示例监控指标 pthread_mutex_lock(stats_lock); stats.queue_size current_queue_size(); stats.active_threads get_active_count(); pthread_mutex_unlock(stats_lock);优雅退出实现分级关闭策略void thread_pool_graceful_shutdown(ThreadPool *pool, int timeout_sec) { pool-shutdown 1; // 第一阶段停止接受新任务 pthread_cond_broadcast(pool-cond); // 第二阶段等待正在执行的任务完成 for(int i 0; i timeout_sec; i) { if(get_active_count(pool) 0) break; sleep(1); } // 第三阶段强制终止 for(int i 0; i pool-thread_count; i) { pthread_cancel(pool-threads[i]); } }与epoll结合处理I/O密集型任务void *io_worker(void *arg) { int epoll_fd epoll_create1(0); // ...epoll事件循环... while(!pool-shutdown) { int n epoll_wait(epoll_fd, events, MAX_EVENTS, 100); for(int i 0; i n; i) { Task *task (Task*)events[i].data.ptr; task-function(task-arg); } } }任务超时处理避免长时间阻塞int pthread_tryjoin_np(pthread_t thread, void **retval, time_t timeout);在实现一个电商平台的订单处理系统时我们使用线程池处理支付回调发现当第三方支付接口响应慢时会导致线程池被占满。最终的解决方案是为支付回调设置单独的小型线程池实现任务超时自动取消添加熔断机制当超时率超过阈值时自动降级