第11讲:linux下多线程编程实例

#include<stdio.h>#include<stdlib.h>#include<pthread.h>#include<unistd.h>#define NUM_THREADS 5// 互斥锁保护共享资源pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;int shared_counter = 0;// 线程函数void* thread_function(void* arg){    int thread_id = *(int*)arg;    // 加锁保护临界区    pthread_mutex_lock(&mutex);    shared_counter++;    printf("Thread %d: counter = %d\n", thread_id, shared_counter);    pthread_mutex_unlock(&mutex);    // 模拟工作    sleep(1);    printf("Thread %d: finished\n", thread_id);    return NULL;}intmain(){    pthread_t threads[NUM_THREADS];    int thread_ids[NUM_THREADS];    // 创建线程    for (int i = 0; i < NUM_THREADS; i++) {        thread_ids[i] = i;        if (pthread_create(&threads[i], NULL, thread_function, &thread_ids[i]) != 0) {            perror("pthread_create");            exit(1);        }        printf("Main: created thread %d\n", i);    }    // 等待所有线程完成    for (int i = 0; i < NUM_THREADS; i++) {        pthread_join(threads[i], NULL);        printf("Main: thread %d joined\n", i);    }    pthread_mutex_destroy(&mutex);    printf("Final counter: %d\n", shared_counter);    return 0;}

gcc -o thread_basic thread_basic.c -pthread

#include<stdio.h>#include<stdlib.h>#include<pthread.h>#include<unistd.h>#define BUFFER_SIZE 10#define NUM_PRODUCERS 3#define NUM_CONSUMERS 2#define ITEMS_PER_PRODUCER 5// 缓冲区结构typedef struct {    int buffer[BUFFER_SIZE];    int in;    int out;    int count;    pthread_mutex_t mutex;    pthread_cond_t not_full;    pthread_cond_t not_empty;} Buffer;Buffer shared_buffer = {    .in = 0,    .out = 0,    .count = 0,    .mutex = PTHREAD_MUTEX_INITIALIZER,    .not_full = PTHREAD_COND_INITIALIZER,    .not_empty = PTHREAD_COND_INITIALIZER};// 生产者线程void* producer(void* arg){    int producer_id = *(int*)arg;    for (int i = 0; i < ITEMS_PER_PRODUCER; i++) {        pthread_mutex_lock(&shared_buffer.mutex);        // 等待缓冲区非满        while (shared_buffer.count == BUFFER_SIZE) {            pthread_cond_wait(&shared_buffer.not_full, &shared_buffer.mutex);        }        // 生产数据        int item = producer_id * 100 + i;        shared_buffer.buffer[shared_buffer.in] = item;        shared_buffer.in = (shared_buffer.in + 1) % BUFFER_SIZE;        shared_buffer.count++;        printf("Producer %d: produced item %d (buffer: %d/%d)\n",                producer_id, item, shared_buffer.count, BUFFER_SIZE);        // 通知消费者        pthread_cond_signal(&shared_buffer.not_empty);        pthread_mutex_unlock(&shared_buffer.mutex);        usleep(rand() % 500000);  // 随机延迟    }    return NULL;}// 消费者线程void* consumer(void* arg){    int consumer_id = *(int*)arg;    while (1) {        pthread_mutex_lock(&shared_buffer.mutex);        // 等待缓冲区非空        while (shared_buffer.count == 0) {            pthread_cond_wait(&shared_buffer.not_empty, &shared_buffer.mutex);        }        // 消费数据        int item = shared_buffer.buffer[shared_buffer.out];        shared_buffer.out = (shared_buffer.out + 1) % BUFFER_SIZE;        shared_buffer.count--;        printf("Consumer %d: consumed item %d (buffer: %d/%d)\n",                consumer_id, item, shared_buffer.count, BUFFER_SIZE);        // 通知生产者        pthread_cond_signal(&shared_buffer.not_full);        pthread_mutex_unlock(&shared_buffer.mutex);        usleep(rand() % 800000);  // 随机延迟    }    return NULL;}intmain(){    pthread_t producers[NUM_PRODUCERS];    pthread_t consumers[NUM_CONSUMERS];    int producer_ids[NUM_PRODUCERS];    int consumer_ids[NUM_CONSUMERS];    srand(time(NULL));    // 创建生产者线程    for (int i = 0; i < NUM_PRODUCERS; i++) {        producer_ids[i] = i + 1;        pthread_create(&producers[i], NULL, producer, &producer_ids[i]);    }    // 创建消费者线程    for (int i = 0; i < NUM_CONSUMERS; i++) {        consumer_ids[i] = i + 1;        pthread_create(&consumers[i], NULL, consumer, &consumer_ids[i]);    }    // 等待生产者完成    for (int i = 0; i < NUM_PRODUCERS; i++) {        pthread_join(producers[i], NULL);    }    // 等待缓冲区清空后取消消费者    sleep(2);    for (int i = 0; i < NUM_CONSUMERS; i++) {        pthread_cancel(consumers[i]);    }    printf("All producers finished\n");    return 0;}

#include<stdio.h>#include<stdlib.h>#include<pthread.h>#include<unistd.h>#define THREAD_POOL_SIZE 4#define TASK_QUEUE_SIZE 10// 任务结构typedef struct {    void (*function)(void*);    void* arg;} Task;// 线程池结构typedef struct {    Task task_queue[TASK_QUEUE_SIZE];    int queue_front;    int queue_rear;    int queue_count;    int active;    pthread_mutex_t mutex;    pthread_cond_t not_empty;    pthread_cond_t not_full;    pthread_t threads[THREAD_POOL_SIZE];} ThreadPool;ThreadPool pool = {    .queue_front = 0,    .queue_rear = 0,    .queue_count = 0,    .active = 1,    .mutex = PTHREAD_MUTEX_INITIALIZER,    .not_empty = PTHREAD_COND_INITIALIZER,    .not_full = PTHREAD_COND_INITIALIZER};// 示例任务函数voidsample_task(void* arg){    int task_id = *(int*)arg;    printf("Thread %lu: executing task %d\n", pthread_self(), task_id);    sleep(1);  // 模拟工作    printf("Thread %lu: completed task %d\n", pthread_self(), task_id);}// 工作线程函数void* worker_thread(void* arg){    while (1) {        pthread_mutex_lock(&pool.mutex);        // 等待任务        while (pool.queue_count == 0 && pool.active) {            pthread_cond_wait(&pool.not_empty, &pool.mutex);        }        if (!pool.active && pool.queue_count == 0) {            pthread_mutex_unlock(&pool.mutex);            break;        }        // 获取任务        Task task = pool.task_queue[pool.queue_front];        pool.queue_front = (pool.queue_front + 1) % TASK_QUEUE_SIZE;        pool.queue_count--;        pthread_cond_signal(&pool.not_full);        pthread_mutex_unlock(&pool.mutex);        // 执行任务        task.function(task.arg);    }    return NULL;}// 提交任务到线程池voidsubmit_task(void (*function)(void*), void* arg){    pthread_mutex_lock(&pool.mutex);    while (pool.queue_count == TASK_QUEUE_SIZE) {        pthread_cond_wait(&pool.not_full, &pool.mutex);    }    pool.task_queue[pool.queue_rear].function = function;    pool.task_queue[pool.queue_rear].arg = arg;    pool.queue_rear = (pool.queue_rear + 1) % TASK_QUEUE_SIZE;    pool.queue_count++;    pthread_cond_signal(&pool.not_empty);    pthread_mutex_unlock(&pool.mutex);}// 初始化线程池voidinit_thread_pool(){    for (int i = 0; i < THREAD_POOL_SIZE; i++) {        pthread_create(&pool.threads[i], NULL, worker_thread, NULL);    }}// 销毁线程池voiddestroy_thread_pool(){    pthread_mutex_lock(&pool.mutex);    pool.active = 0;    pthread_cond_broadcast(&pool.not_empty);    pthread_mutex_unlock(&pool.mutex);    for (int i = 0; i < THREAD_POOL_SIZE; i++) {        pthread_join(pool.threads[i], NULL);    }    pthread_mutex_destroy(&pool.mutex);    pthread_cond_destroy(&pool.not_empty);    pthread_cond_destroy(&pool.not_full);}intmain(){    printf("Starting thread pool with %d threads\n", THREAD_POOL_SIZE);    init_thread_pool();    // 提交15个任务    int task_ids[15];    for (int i = 0; i < 15; i++) {        task_ids[i] = i + 1;        submit_task(sample_task, &task_ids[i]);        printf("Submitted task %d\n", i + 1);    }    // 等待所有任务完成    sleep(5);    printf("Destroying thread pool\n");    destroy_thread_pool();    return 0;}

#include<stdio.h>#include<stdlib.h>#include<pthread.h>#include<unistd.h>#define NUM_READERS 5#define NUM_WRITERS 2// 共享数据结构typedef struct {    int data;    int read_count;    pthread_mutex_t mutex;    pthread_cond_t can_write;    pthread_cond_t can_read;    int writing;} SharedData;SharedData shared = {    .data = 0,    .read_count = 0,    .mutex = PTHREAD_MUTEX_INITIALIZER,    .can_write = PTHREAD_COND_INITIALIZER,    .can_read = PTHREAD_COND_INITIALIZER,    .writing = 0};// 读者线程void* reader(void* arg){    int reader_id = *(int*)arg;    while (1) {        pthread_mutex_lock(&shared.mutex);        // 等待写者完成        while (shared.writing) {            pthread_cond_wait(&shared.can_read, &shared.mutex);        }        shared.read_count++;        pthread_mutex_unlock(&shared.mutex);        // 读操作        printf("Reader %d: read data = %d (readers: %d)\n",                reader_id, shared.data, shared.read_count);        pthread_mutex_lock(&shared.mutex);        shared.read_count--;        // 如果没有读者，通知写者        if (shared.read_count == 0) {            pthread_cond_signal(&shared.can_write);        }        pthread_mutex_unlock(&shared.mutex);        sleep(1);    }    return NULL;}// 写者线程void* writer(void* arg){    int writer_id = *(int*)arg;    while (1) {        pthread_mutex_lock(&shared.mutex);        // 等待所有读者完成        while (shared.read_count > 0) {            pthread_cond_wait(&shared.can_write, &shared.mutex);        }        shared.writing = 1;        // 写操作        shared.data++;        printf("Writer %d: wrote data = %d\n", writer_id, shared.data);        shared.writing = 0;        // 通知等待的读者        pthread_cond_broadcast(&shared.can_read);        pthread_mutex_unlock(&shared.mutex);        sleep(2);    }    return NULL;}intmain(){    pthread_t readers[NUM_READERS];    pthread_t writers[NUM_WRITERS];    int reader_ids[NUM_READERS];    int writer_ids[NUM_WRITERS];    // 创建读者线程    for (int i = 0; i < NUM_READERS; i++) {        reader_ids[i] = i + 1;        pthread_create(&readers[i], NULL, reader, &reader_ids[i]);    }    // 创建写者线程    for (int i = 0; i < NUM_WRITERS; i++) {        writer_ids[i] = i + 1;        pthread_create(&writers[i], NULL, writer, &writer_ids[i]);    }    // 运行一段时间    sleep(10);    // 取消线程    for (int i = 0; i < NUM_READERS; i++) {        pthread_cancel(readers[i]);    }    for (int i = 0; i < NUM_WRITERS; i++) {        pthread_cancel(writers[i]);    }    return 0;}

# 基础示例gcc -o thread_basic thread_basic.c -pthread./thread_basic# 生产者消费者gcc -o producer_consumer producer_consumer.c -pthread./producer_consumer# 线程池gcc -o thread_pool thread_pool.c -pthread./thread_pool# 读写锁gcc -o read_write read_write.c -pthread./read_write