基于POSIX标准的多线程同步机制与线程池实现
在Linux环境下，使用POSIX线程（pthread）库可以高效地实现多线程程序中的同步与通信。本文将介绍互斥锁的不同类型、条件变量和信号量的使用方式，并最终构建一个可复用的线程池框架。
互斥锁属性配置
互斥锁可通过属性对象设置其行为特性。以下代码展示了如何根据命令行参数设定互斥锁的类型：
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <string.h>
int main(int argc, char *argv[]) {
pthread_mutex_t mutex;
pthread_mutexattr_t attr;
if (argc != 2) {
fprintf(stderr, "请指定锁类型: normal, errorcheck, 或 recursive\n");
return EXIT_FAILURE;
}
pthread_mutexattr_init(&attr);
pthread_mutexattr_setpshared(&attr, PTHREAD_PROCESS_SHARED);
if (strcmp(argv[1], "errorcheck") == 0) {
pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_ERRORCHECK);
} else if (strcmp(argv[1], "recursive") == 0) {
pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE);
} // 默认为 NORMAL 类型
pthread_mutex_init(&mutex, &attr);
// 尝试重复加锁
pthread_mutex_lock(&mutex);
printf("首次加锁成功\n");
int result = pthread_mutex_lock(&mutex);
if (result != 0) {
printf("二次加锁失败，错误码: %d\n", result);
} else {
printf("二次加锁成功\n");
pthread_mutex_unlock(&mutex);
}
pthread_mutex_unlock(&mutex);
pthread_mutex_destroy(&mutex);
pthread_mutexattr_destroy(&attr);
return 0;
}
使用条件变量实现线程协作
条件变量常用于线程间的通知机制。下面示例中，主线程读取输入后唤醒工作线程处理数据：
#include <stdio.h>
#include <pthread.h>
#include <string.h>>
#include <unistd.h>
#define BUFFER_SIZE 32
char shared_buffer[BUFFER_SIZE];
pthread_mutex_t mtx = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
int data_ready = 0;
void* input_thread(void* arg) {
fgets(shared_buffer, BUFFER_SIZE, stdin);
size_t len = strlen(shared_buffer);
if (len > 0 && shared_buffer[len-1] == '\n') {
shared_buffer[len-1] = '\0';
}
pthread_mutex_lock(&mtx);
data_ready = 1;
pthread_cond_signal(&cond);
pthread_mutex_unlock(&mtx);
pthread_exit(NULL);
}
void* process_thread(void* arg) {
pthread_mutex_lock(&mtx);
while (!data_ready) {
pthread_cond_wait(&cond, &mtx);
}
printf("接收到数据: %s\n", shared_buffer);
pthread_mutex_unlock(&mtx);
pthread_exit(NULL);
}
int main() {
pthread_t t1, t2;
pthread_create(&t1, NULL, input_thread, NULL);
pthread_create(&t2, NULL, process_thread, NULL);
pthread_join(t1, NULL);
pthread_join(t2, NULL);
return 0;
}
线程池设计与实现
线程池通过预创建线程减少频繁创建销毁的开销。核心结构包含任务队列、互斥锁和条件变量：
// threadpool.h
#ifndef THREAD_POOL_H
#define THREAD_POOL_H
typedef struct Task {
void (*function)(void*);
void* argument;
struct Task* next;
} Task;
typedef struct ThreadPool {
pthread_mutex_t lock;
pthread_cond_t notify;
Task* first;
int thread_count;
int task_count;
int shutdown;
pthread_t* threads;
} ThreadPool;
void threadpool_init(ThreadPool* pool, int num_threads);
int threadpool_add_task(ThreadPool* pool, void (*func)(void*), void* arg);
int threadpool_destroy(ThreadPool* pool);
#endif
工作线程循环等待新任务：
static void* worker_routine(void* arg) {
ThreadPool* pool = (ThreadPool*)arg;
Task* current;
while (1) {
pthread_mutex_lock(&pool->lock);
while (pool->task_count == 0 && !pool->shutdown) {
pthread_cond_wait(&pool->notify, &pool->lock);
}
if (pool->shutdown) break;
current = pool->first;
if (current) {
pool->first = current->next;
pool->task_count--;
}
pthread_mutex_unlock(&pool->lock);
if (current) {
current->function(current->argument);
free(current);
}
}
pthread_mutex_unlock(&pool->lock);
pthread_exit(NULL);
return NULL;
}
主函数中初始化线程池除并提交多个任务：
void sample_task(void* param) {
int id = *(int*)param;
printf("执行任务 #%d in thread 0x%lx\n", id, pthread_self());
sleep(1);
}
int main() {
ThreadPool pool;
threadpool_init(&pool, 3);
for (int i = 0; i < 10; i++) {
int* tid = malloc(sizeof(int));
*tid = i;
threadpool_add_task(&pool, sample_task, tid);
}
sleep(5);
threadpool_destroy(&pool);
return 0;
}
编译与构建
使用Makefile管理项目依赖：
.PHONY: all clean
TARGET = app
OBJS = main.o threadpool.o
all: $(TARGET)
$(TARGET): $(OBJS)
gcc -o $@ $^ -lpthread
%.o: %.c
gcc -c -Wall $< -o $@
clean:
rm -f *.o $(TARGET)
