第11章：线程（Threads）

一、导读

1.1 本章学习目标

第11章引入线程的概念——同一进程内的独立执行流。与进程不同，线程共享进程的地址空间和资源（文件描述符、信号处理等），但拥有独立的栈和寄存器上下文。线程是比进程更轻量级的并发单元，创建和切换的开销更小。多线程编程是提高程序并发性和响应性的关键手段，但也带来了同步和竞态条件的挑战。

本章的核心学习目标包括：

理解线程的概念：线程与进程的区别、线程共享和私有的资源
掌握POSIX线程的创建和管理：pthread_create、pthread_join、pthread_exit
学会使用互斥锁：保护临界区，防止多个线程同时访问共享数据
掌握条件变量：线程间通信，配合互斥锁实现等待/通知模式
理解读写锁：允许并发读但独占写，适合读多写少的场景
了解线程的安全性问题：可重入函数、线程局部存储、死锁

1.2 为什么本章如此重要

现代计算机几乎都是多核处理器，单线程程序无法充分利用硬件的并行能力。多线程编程成为提高程序性能的关键手段：

Web服务器：每个客户端连接由一个线程处理
数据库系统：并发查询由多个线程并行执行
图形界面：UI线程和工作线程分离，保持界面响应
科学计算：大规模计算分解为多个线程并行处理

然而，多线程编程也带来了复杂性：数据竞争、死锁、活锁、饥饿等问题。理解线程同步机制（互斥锁、条件变量、读写锁）是编写正确、高效的多线程程序的前提。

1.3 核心问题引导

在阅读本章时，请带着以下问题思考：

线程和进程的本质区别是什么？线程共享哪些资源，私有哪些资源？

为什么多线程比多进程更轻量？轻量在哪里？

什么是数据竞争？如何检测和避免？

互斥锁解决了什么问题？为什么需要unlock？

条件变量的wait为什么要配合互斥锁使用？为什么用while而不是if检查条件？

什么是死锁？如何避免死锁？

二、核心概念详解

2.1 线程的概念

线程（thread）是进程内的独立执行流。一个进程可以包含多个线程，它们共享进程的地址空间，但拥有独立的栈、寄存器状态和线程局部存储。

2.1.1 线程共享的资源

共享资源	说明
地址空间	代码段、数据段、堆
文件描述符表	每个线程可以操作相同的文件
信号处理器	信号处理函数对进程内所有线程生效
进程ID、父进程ID	线程共享进程标识
当前工作目录	所有线程共享
用户ID、组ID	线程共享

2.1.2 线程私有的资源

私有资源	说明
线程ID	每个线程有唯一标识
栈	每个线程有独立的栈空间
寄存器状态	程序计数器、栈指针等
信号掩码	每个线程可以阻塞不同的信号
errno	每个线程有自己的errno副本
线程局部存储（TSD）	线程私有的全局变量

#include <stdio.h>
#include <pthread.h>
#include <unistd.h>

void *thread_func(void *arg) {
    printf("Thread: TID=%lu, PID=%d\n", 
           (unsigned long)pthread_self(), getpid());
    return NULL;
}

int main() {
    printf("Main: TID=%lu, PID=%d\n", 
           (unsigned long)pthread_self(), getpid());
    
    pthread_t tid;
    pthread_create(&tid, NULL, thread_func, NULL);
    pthread_join(tid, NULL);
    
    return 0;
}

2.2 线程创建

#include <pthread.h>

int pthread_create(pthread_t *thread, const pthread_attr_t *attr,
                   void *(*start_routine)(void *), void *arg);

thread：输出参数，返回新线程的ID
attr：线程属性（NULL使用默认属性）
start_routine：线程函数
arg：传递给线程函数的参数

#include <stdio.h>
#include <pthread.h>
#include <stdlib.h>

void *thread_func(void *arg) {
    int id = *(int *)arg;
    printf("Thread %d started\n", id);
    sleep(1);
    printf("Thread %d done\n", id);
    return NULL;
}

int main() {
    pthread_t threads[5];
    int ids[5];
    
    for (int i = 0; i < 5; i++) {
        ids[i] = i + 1;
        pthread_create(&threads[i], NULL, thread_func, &ids[i]);
    }
    
    for (int i = 0; i < 5; i++) {
        pthread_join(threads[i], NULL);
    }
    
    printf("All threads done\n");
    return 0;
}

2.3 线程终止

线程有三种终止方式：

函数返回：线程函数正常返回

pthread_exit：显式终止线程

被其他线程取消：pthread_cancel

#include <pthread.h>
#include <stdio.h>

// 方式1：函数返回
void *thread1(void *arg) {
    printf("Thread 1 returning\n");
    return (void *)42;  // 返回值可以被pthread_join获取
}

// 方式2：pthread_exit
void *thread2(void *arg) {
    printf("Thread 2 calling pthread_exit\n");
    pthread_exit((void *)100);
}

int main() {
    pthread_t t1, t2;
    void *retval;
    
    pthread_create(&t1, NULL, thread1, NULL);
    pthread_create(&t2, NULL, thread2, NULL);
    
    pthread_join(t1, &retval);
    printf("Thread 1 returned: %ld\n", (long)retval);
    
    pthread_join(t2, &retval);
    printf("Thread 2 returned: %ld\n", (long)retval);
    
    return 0;
}

注意：main函数返回或调用exit会终止整个进程（包括所有线程）。如果只想终止单个线程，应使用pthread_exit。

2.4 互斥锁（Mutex）

互斥锁（mutex）用于保护临界区，防止多个线程同时访问共享数据。

#include <pthread.h>

// 初始化
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
// 或动态初始化
pthread_mutex_init(&mutex, NULL);

// 加锁和解锁
int pthread_mutex_lock(pthread_mutex_t *mutex);
int pthread_mutex_unlock(pthread_mutex_t *mutex);

// 尝试加锁（非阻塞）
int pthread_mutex_trylock(pthread_mutex_t *mutex);

// 销毁
pthread_mutex_destroy(pthread_mutex_t *mutex);

#include <stdio.h>
#include <pthread.h>

int counter = 0;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

void *increment(void *arg) {
    for (int i = 0; i < 100000; i++) {
        pthread_mutex_lock(&mutex);
        counter++;  // 临界区
        pthread_mutex_unlock(&mutex);
    }
    return NULL;
}

int main() {
    pthread_t t1, t2;
    pthread_create(&t1, NULL, increment, NULL);
    pthread_create(&t2, NULL, increment, NULL);
    
    pthread_join(t1, NULL);
    pthread_join(t2, NULL);
    
    printf("Counter: %d (expected: 200000)\n", counter);
    
    pthread_mutex_destroy(&mutex);
    return 0;
}

没有互斥锁的后果：

// 错误示例：数据竞争
int counter = 0;

void *increment(void *arg) {
    for (int i = 0; i < 100000; i++) {
        counter++;  // 非原子操作！
        // 实际执行：load -> increment -> store
        // 两个线程可能同时load相同的值，导致丢失更新
    }
    return NULL;
}
// 结果可能小于200000

2.5 死锁

死锁（deadlock）是两个或多个线程互相等待对方释放资源，导致都无法继续执行。

// 死锁示例
pthread_mutex_t mutex1 = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t mutex2 = PTHREAD_MUTEX_INITIALIZER;

void *thread1(void *arg) {
    pthread_mutex_lock(&mutex1);
    sleep(1);  // 增加死锁概率
    pthread_mutex_lock(&mutex2);
    // 使用资源...
    pthread_mutex_unlock(&mutex2);
    pthread_mutex_unlock(&mutex1);
    return NULL;
}

void *thread2(void *arg) {
    pthread_mutex_lock(&mutex2);
    sleep(1);
    pthread_mutex_lock(&mutex1);
    // 使用资源...
    pthread_mutex_unlock(&mutex1);
    pthread_mutex_unlock(&mutex2);
    return NULL;
}

避免死锁的策略：

固定加锁顺序：所有线程按相同顺序获取锁

使用trylock：尝试获取锁，失败时释放已持有的锁并重试

使用超时锁：pthread_mutex_timedlock设置超时

// 固定加锁顺序避免死锁
void safe_function() {
    pthread_mutex_lock(&mutex1);  // 总是先锁mutex1
    pthread_mutex_lock(&mutex2);  // 再锁mutex2
    // 使用资源...
    pthread_mutex_unlock(&mutex2);
    pthread_mutex_unlock(&mutex1);
}

2.6 条件变量

条件变量（condition variable）用于线程间通信：一个线程等待某个条件成立，另一个线程通知条件已满足。

#include <pthread.h>

pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
pthread_cond_init(&cond, NULL);

int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);
int pthread_cond_signal(pthread_cond_t *cond);    // 唤醒一个等待线程
int pthread_cond_broadcast(pthread_cond_t *cond); // 唤醒所有等待线程
pthread_cond_destroy(pthread_cond_t *cond);

条件变量的使用模式：

#include <stdio.h>
#include <pthread.h>
#include <unistd.h>

int ready = 0;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

void *worker(void *arg) {
    // 等待条件成立
    pthread_mutex_lock(&mutex);
    while (!ready) {  // 必须用while，不能用if
        pthread_cond_wait(&cond, &mutex);
    }
    // 条件已成立，执行工作
    printf("Worker: ready=%d, doing work\n", ready);
    pthread_mutex_unlock(&mutex);
    return NULL;
}

int main() {
    pthread_t tid;
    pthread_create(&tid, NULL, worker, NULL);
    
    sleep(1);  // 模拟准备时间
    
    // 设置条件并通知
    pthread_mutex_lock(&mutex);
    ready = 1;
    pthread_cond_signal(&cond);
    pthread_mutex_unlock(&mutex);
    
    pthread_join(tid, NULL);
    printf("Main: done\n");
    
    pthread_mutex_destroy(&mutex);
    pthread_cond_destroy(&cond);
    return 0;
}

为什么用while而不是if？

虚假唤醒（spurious wakeup）：某些实现中，cond_wait可能在没有signal的情况下返回

多个等待者：signal只唤醒一个线程，但被唤醒的线程检查条件时，条件可能已被其他线程改变

2.7 读写锁

读写锁（read-write lock）允许多个线程同时读，但写操作是独占的。适合读多写少的场景。

#include <pthread.h>

pthread_rwlock_t rwlock = PTHREAD_RWLOCK_INITIALIZER;

int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);   // 读锁
int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock);   // 写锁
int pthread_rwlock_unlock(pthread_rwlock_t *rwlock);
pthread_rwlock_destroy(pthread_rwlock_t *rwlock);

#include <stdio.h>
#include <pthread.h>
#include <unistd.h>

int shared_data = 0;
pthread_rwlock_t rwlock = PTHREAD_RWLOCK_INITIALIZER;

void *reader(void *arg) {
    int id = *(int *)arg;
    for (int i = 0; i < 5; i++) {
        pthread_rwlock_rdlock(&rwlock);
        printf("Reader %d: data=%d\n", id, shared_data);
        pthread_rwlock_unlock(&rwlock);
        usleep(100000);
    }
    return NULL;
}

void *writer(void *arg) {
    int id = *(int *)arg;
    for (int i = 0; i < 3; i++) {
        pthread_rwlock_wrlock(&rwlock);
        shared_data++;
        printf("Writer %d: data=%d\n", id, shared_data);
        pthread_rwlock_unlock(&rwlock);
        usleep(200000);
    }
    return NULL;
}

int main() {
    pthread_t readers[3], writers[2];
    int ids[3] = {1, 2, 3};
    
    for (int i = 0; i < 3; i++)
        pthread_create(&readers[i], NULL, reader, &ids[i]);
    for (int i = 0; i < 2; i++)
        pthread_create(&writers[i], NULL, writer, &ids[i]);
    
    for (int i = 0; i < 3; i++)
        pthread_join(readers[i], NULL);
    for (int i = 0; i < 2; i++)
        pthread_join(writers[i], NULL);
    
    pthread_rwlock_destroy(&rwlock);
    return 0;
}

三、实战案例

3.1 生产者-消费者模型

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>

#define BUF_SIZE 10
#define NUM_PRODUCERS 2
#define NUM_CONSUMERS 3

int buffer[BUF_SIZE];
int count = 0;
int in = 0, out = 0;

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t not_full = PTHREAD_COND_INITIALIZER;
pthread_cond_t not_empty = PTHREAD_COND_INITIALIZER;

void *producer(void *arg) {
    int id = *(int *)arg;
    for (int i = 0; i < 10; i++) {
        int item = id * 100 + i;
        
        pthread_mutex_lock(&mutex);
        while (count == BUF_SIZE) {
            printf("Producer %d: buffer full, waiting...\n", id);
            pthread_cond_wait(&not_full, &mutex);
        }
        
        buffer[in] = item;
        in = (in + 1) % BUF_SIZE;
        count++;
        printf("Producer %d: produced %d (count=%d)\n", id, item, count);
        
        pthread_cond_signal(&not_empty);
        pthread_mutex_unlock(&mutex);
        
        usleep(rand() % 100000);
    }
    return NULL;
}

void *consumer(void *arg) {
    int id = *(int *)arg;
    for (int i = 0; i < 7; i++) {
        pthread_mutex_lock(&mutex);
        while (count == 0) {
            printf("Consumer %d: buffer empty, waiting...\n", id);
            pthread_cond_wait(&not_empty, &mutex);
        }
        
        int item = buffer[out];
        out = (out + 1) % BUF_SIZE;
        count--;
        printf("Consumer %d: consumed %d (count=%d)\n", id, item, count);
        
        pthread_cond_signal(&not_full);
        pthread_mutex_unlock(&mutex);
        
        usleep(rand() % 150000);
    }
    return NULL;
}

int main() {
    pthread_t producers[NUM_PRODUCERS], consumers[NUM_CONSUMERS];
    int pids[NUM_PRODUCERS], cids[NUM_CONSUMERS];
    
    for (int i = 0; i < NUM_PRODUCERS; i++) {
        pids[i] = i + 1;
        pthread_create(&producers[i], NULL, producer, &pids[i]);
    }
    
    for (int i = 0; i < NUM_CONSUMERS; i++) {
        cids[i] = i + 1;
        pthread_create(&consumers[i], NULL, consumer, &cids[i]);
    }
    
    for (int i = 0; i < NUM_PRODUCERS; i++)
        pthread_join(producers[i], NULL);
    for (int i = 0; i < NUM_CONSUMERS; i++)
        pthread_join(consumers[i], NULL);
    
    printf("All done. Final count: %d\n", count);
    
    pthread_mutex_destroy(&mutex);
    pthread_cond_destroy(&not_full);
    pthread_cond_destroy(&not_empty);
    return 0;
}

3.2 线程安全的单例模式

#include <stdio.h>
#include <pthread.h>
#include <stdlib.h>

typedef struct {
    int value;
    pthread_mutex_t mutex;
} Singleton;

static Singleton *instance = NULL;
static pthread_once_t once_control = PTHREAD_ONCE_INIT;

void init_singleton() {
    instance = malloc(sizeof(Singleton));
    instance->value = 42;
    pthread_mutex_init(&instance->mutex, NULL);
    printf("Singleton initialized\n");
}

Singleton *get_singleton() {
    pthread_once(&once_control, init_singleton);
    return instance;
}

void *thread_func(void *arg) {
    Singleton *s = get_singleton();
    printf("Thread %lu: singleton value = %d\n",
           (unsigned long)pthread_self(), s->value);
    return NULL;
}

int main() {
    pthread_t threads[5];
    for (int i = 0; i < 5; i++) {
        pthread_create(&threads[i], NULL, thread_func, NULL);
    }
    for (int i = 0; i < 5; i++) {
        pthread_join(threads[i], NULL);
    }
    
    free(instance);
    return 0;
}

3.3 并行计算

#include <stdio.h>
#include <pthread.h>
#include <stdlib.h>
#include <time.h>

#define ARRAY_SIZE 10000000
#define NUM_THREADS 4

int array[ARRAY_SIZE];

typedef struct {
    int start;
    int end;
    long sum;
} ThreadArg;

void *compute_sum(void *arg) {
    ThreadArg *ta = (ThreadArg *)arg;
    ta->sum = 0;
    for (int i = ta->start; i < ta->end; i++) {
        ta->sum += array[i];
    }
    return NULL;
}

int main() {
    // 初始化数组
    srand(time(NULL));
    for (int i = 0; i < ARRAY_SIZE; i++) {
        array[i] = rand() % 100;
    }
    
    // 串行计算
    clock_t start = clock();
    long serial_sum = 0;
    for (int i = 0; i < ARRAY_SIZE; i++) {
        serial_sum += array[i];
    }
    clock_t serial_time = clock() - start;
    
    // 并行计算
    start = clock();
    pthread_t threads[NUM_THREADS];
    ThreadArg args[NUM_THREADS];
    int chunk = ARRAY_SIZE / NUM_THREADS;
    
    for (int i = 0; i < NUM_THREADS; i++) {
        args[i].start = i * chunk;
        args[i].end = (i == NUM_THREADS - 1) ? ARRAY_SIZE : (i + 1) * chunk;
        pthread_create(&threads[i], NULL, compute_sum, &args[i]);
    }
    
    long parallel_sum = 0;
    for (int i = 0; i < NUM_THREADS; i++) {
        pthread_join(threads[i], NULL);
        parallel_sum += args[i].sum;
    }
    clock_t parallel_time = clock() - start;
    
    printf("Serial sum: %ld, time: %.3f ms\n",
           serial_sum, (double)serial_time / CLOCKS_PER_SEC * 1000);
    printf("Parallel sum: %ld, time: %.3f ms\n",
           parallel_sum, (double)parallel_time / CLOCKS_PER_SEC * 1000);
    printf("Speedup: %.2fx\n",
           (double)serial_time / parallel_time);
    
    return 0;
}

四、本章小结

第11章介绍了多线程编程的基础。核心知识点包括：

线程概念：同一进程内的独立执行流。共享地址空间、文件描述符等，私有栈、寄存器、errno等。

线程创建：pthread_create创建线程，pthread_join等待线程结束，pthread_exit终止线程。

互斥锁：保护临界区，防止数据竞争。lock/unlock成对使用，避免死锁。

条件变量：线程间通信，配合互斥锁使用。wait原子地释放锁并等待，signal/broadcast唤醒等待线程。必须用while检查条件。

读写锁：允许多个读线程并发，写线程独占。适合读多写少的场景。

死锁避免：固定加锁顺序、使用trylock、设置超时。

五、思考与练习

思考题

为什么条件变量的wait操作必须是原子的（释放锁+等待）？如果不是原子操作，会出现什么问题？

互斥锁和读写锁各自适用于什么场景？如何选择？

为什么pthread_join不能对同一个线程调用两次？

线程的信号掩码是进程级的还是线程级的？如何向特定线程发送信号？

编程练习

实现一个线程安全的队列（有界缓冲区），支持push和pop操作。

编写一个程序，使用多线程计算矩阵乘法，每个线程计算结果矩阵的一行。

实现一个读者-写者锁的简化版（不使用pthread_rwlock）。

编写一个线程池：预创建N个工作线程，从任务队列中取出任务执行。