{T}

阻塞I/O + 线程模型：轻量级并发方案

在前文中，我们探讨了使用进程模型处理并发连接的方案。进程切换（context switch）的代价较高，而线程（thread）作为更轻量的执行单元，提供了另一种并发处理途径。本文将深入分析线程模型的核心概念、POSIX 线程 API，以及基于线程池的高效服务器设计。

线程基础

线程（thread）是运行于进程内部的逻辑流（logical flow），由操作系统内核调度管理。每个线程拥有独立的执行上下文（context），包括：

  • 线程 ID（thread ID / tid）：唯一标识线程
  • 栈空间（stack）：存储局部变量和调用链
  • 程序计数器（program counter）：记录执行位置
  • 寄存器组（registers）：存储当前计算状态

同一进程内的所有线程共享进程的虚拟地址空间，包括代码段、数据段、堆内存及共享库。这种共享机制使得线程间通信效率远高于进程间通信（IPC）。

每个进程启动时自动创建一个主线程（main thread），主线程可进一步创建子线程，形成对等线程（peer thread）关系。

线程与进程的比较

特性进程线程
创建开销高（需复制地址空间）低（共享地址空间）
上下文切换高（需切换地址空间）低（仅切换寄存器/栈）
内存隔离完全隔离共享地址空间
通信机制IPC（管道、共享内存等）直接访问共享变量
崩溃影响独立，不影响其他进程可能导致整个进程崩溃

POSIX 线程模型

POSIX 线程（pthread）是现代 UNIX 系统的线程编程标准接口，定义了约 60 个线程管理函数。以下通过示例程序介绍核心 API。

基础示例

c
int another_shared = 0;

void *thread_run(void *arg) {
    int *calculator = (int *) arg;
    printf("hello, world, tid == %lu\n", (unsigned long)pthread_self());
    for (int i = 0; i < 1000; i++) {
        *calculator += 1;
        another_shared += 1;
    }
    return NULL;
}

int main(int c, char **v) {
    int calculator = 0;

    pthread_t tid1;
    pthread_t tid2;

    pthread_create(&tid1, NULL, thread_run, &calculator);
    pthread_create(&tid2, NULL, thread_run, &calculator);

    pthread_join(tid1, NULL);
    pthread_join(tid2, NULL);

    printf("calculator is %d\n", calculator);
    printf("another_shared is %d\n", another_shared);
    return 0;
}

程序执行结果：

code
$ ./thread-helloworld
hello, world, tid == 125607936
hello, world, tid == 126144512
calculator is 2000
another_shared is 2000

注意：上述程序存在数据竞争（data race）风险。两个线程并发修改共享变量 calculator 和 another_shared，未使用同步机制。此处结果正确纯属巧合，实际运行可能产生不一致的结果。

核心 API 详解

创建线程

c
int pthread_create(pthread_t *tid, const pthread_attr_t *attr,
                   void *(*func)(void *), void *arg);
// 返回值：成功返回 0，失败返回正的 Exxx 错误码
  • tid：输出参数，返回新线程的标识符
  • attr：线程属性（优先级、栈大小等），NULL 使用默认值
  • func：线程入口函数，签名为 void *(*func)(void *)
  • arg：传递给入口函数的参数指针

线程内部可通过 pthread_self() 获取自身 tid：

c
pthread_t pthread_self(void);

终止线程

c
void pthread_exit(void *status);

调用线程立即终止，status 指向的值可被 pthread_join() 获取。若主线程调用 pthread_exit()，进程将等待所有线程终止后才退出。

c
int pthread_cancel(pthread_t tid);

请求终止指定线程，目标线程可在取消点（cancellation point）检查并处理取消请求。

回收线程资源

c
int pthread_join(pthread_t tid, void **thread_return);
// 返回值：成功返回 0，失败返回正的 Exxx 错误码

调用线程阻塞，直至目标线程终止。thread_return 可获取目标线程的退出状态。该函数不会强制终止目标线程，仅等待其自然结束。

分离线程

c
int pthread_detach(pthread_t tid);

将线程标记为分离状态（detached state）。分离线程终止后自动回收资源，无需其他线程调用 pthread_join()。在高并发服务器中，每个连接由独立线程处理时，通常在子线程入口处调用 pthread_detach(pthread_self()) 实现自动资源回收。

每连接一线程模型

架构设计

图表渲染中…
图表渲染中…

程序实现

c
#include "lib/common.h"

extern void loop_echo(int);

void *thread_run(void *arg) {
    pthread_detach(pthread_self());
    int fd = (int)(intptr_t) arg;
    loop_echo(fd);
    return NULL;
}

int main(int c, char **v) {
    int listener_fd = tcp_server_listen(SERV_PORT);
    pthread_t tid;

    while (1) {
        struct sockaddr_storage ss;
        socklen_t slen = sizeof(ss);
        int fd = accept(listener_fd, (struct sockaddr *) &ss, &slen);
        if (fd < 0) {
            error(1, errno, "accept failed");
        } else {
            pthread_create(&tid, NULL, &thread_run, (void *)(intptr_t) fd);
        }
    }

    return 0;
}

实现要点：

  • 文件描述符通过 (void *)(intptr_t) 进行类型转换传递，确保指针与整型间安全转换
  • 子线程入口处调用 pthread_detach() 实现自动资源回收
  • loop_echo() 函数执行阻塞 I/O 读写，处理客户端数据

loop_echo 实现

c
char rot13_char(char c) {
    if ((c >= 'a' && c <= 'm') || (c >= 'A' && c <= 'M'))
        return c + 13;
    else if ((c >= 'n' && c <= 'z') || (c >= 'N' && c <= 'Z'))
        return c - 13;
    else
        return c;
}

void loop_echo(int fd) {
    char outbuf[MAX_LINE + 1];
    size_t outbuf_used = 0;
    ssize_t result;
    while (1) {
        char ch;
        result = recv(fd, &ch, 1, 0);

        if (result == 0) {
            break;
        } else if (result == -1) {
            error(1, errno, "read error");
            break;
        }

        if (outbuf_used < sizeof(outbuf)) {
            outbuf[outbuf_used++] = rot13_char(ch);
        }

        if (ch == '\n') {
            send(fd, outbuf, outbuf_used, 0);
            outbuf_used = 0;
            continue;
        }
    }
}

线程池模型

问题分析

每连接一线程模型存在以下缺陷：

  • 线程创建开销：频繁创建/销毁线程消耗 CPU 资源
  • 资源消耗：大量线程占用栈空间（默认每线程 8MB）
  • 调度开销：过多线程增加内核调度负担

解决方案：预创建线程池

服务器启动时预先创建固定数量的工作线程，形成线程池（thread pool）。主线程接受新连接后，将连接描述符放入任务队列，工作线程从队列中取出连接进行处理。

图表渲染中…

阻塞队列实现

线程池的核心是线程安全的任务队列，需使用互斥锁（mutex）和条件变量（condition variable）实现同步。

c
typedef struct {
    int number;              // 队列容量
    int *fd;                 // 文件描述符数组
    int front;               // 队头索引
    int rear;                // 队尾索引
    pthread_mutex_t mutex;   // 互斥锁
    pthread_cond_t cond;     // 条件变量
} block_queue;

void block_queue_init(block_queue *blockQueue, int number) {
    blockQueue->number = number;
    blockQueue->fd = calloc(number, sizeof(int));
    blockQueue->front = blockQueue->rear = 0;
    pthread_mutex_init(&blockQueue->mutex, NULL);
    pthread_cond_init(&blockQueue->cond, NULL);
}

void block_queue_push(block_queue *blockQueue, int fd) {
    pthread_mutex_lock(&blockQueue->mutex);
    
    blockQueue->fd[blockQueue->rear] = fd;
    if (++blockQueue->rear == blockQueue->number) {
        blockQueue->rear = 0;
    }
    
    pthread_cond_signal(&blockQueue->cond);
    pthread_mutex_unlock(&blockQueue->mutex);
}

int block_queue_pop(block_queue *blockQueue) {
    pthread_mutex_lock(&blockQueue->mutex);
    
    while (blockQueue->front == blockQueue->rear) {
        pthread_cond_wait(&blockQueue->cond, &blockQueue->mutex);
    }
    
    int fd = blockQueue->fd[blockQueue->front];
    if (++blockQueue->front == blockQueue->number) {
        blockQueue->front = 0;
    }
    
    pthread_mutex_unlock(&blockQueue->mutex);
    return fd;
}

同步机制说明：

  • pthread_mutex_lock/unlock：保护队列操作的原子性
  • pthread_cond_wait：队列为空时，工作线程阻塞等待
  • pthread_cond_signal：新任务入队时，唤醒等待的工作线程

线程池服务器实现

c
void *thread_run(void *arg) {
    pthread_t tid = pthread_self();
    pthread_detach(tid);

    block_queue *blockQueue = (block_queue *) arg;
    while (1) {
        int fd = block_queue_pop(blockQueue);
        printf("get fd in thread, fd==%d, tid==%lu\n", fd, (unsigned long)tid);
        loop_echo(fd);
    }
    return NULL;
}

int main(int c, char **v) {
    int listener_fd = tcp_server_listen(SERV_PORT);

    block_queue blockQueue;
    block_queue_init(&blockQueue, BLOCK_QUEUE_SIZE);

    pthread_t *thread_array = calloc(THREAD_NUMBER, sizeof(pthread_t));
    for (int i = 0; i < THREAD_NUMBER; i++) {
        pthread_create(&thread_array[i], NULL, &thread_run, (void *) &blockQueue);
    }

    while (1) {
        struct sockaddr_storage ss;
        socklen_t slen = sizeof(ss);
        int fd = accept(listener_fd, (struct sockaddr *) &ss, &slen);
        if (fd < 0) {
            error(1, errno, "accept failed");
        } else {
            block_queue_push(&blockQueue, fd);
        }
    }

    return 0;
}

模型评估

维度每连接一线程线程池
线程创建开销高（每次连接）低（启动时预创建）
资源可控性差（线程数不可控）好（固定线程数）
响应延迟低（立即处理）可能排队等待
适用场景连接数适中高并发、短连接

线程池模型显著降低了线程创建/销毁开销，但固定线程数配合阻塞 I/O 仍存在瓶颈：当所有线程都在处理长连接时，新连接可能得不到及时服务。要实现极致高并发，需结合 I/O 多路复用（I/O multiplexing）技术。

思考题

  1. 阻塞队列实现中未处理队列满的情况。当队列已满时，block_queue_push() 应如何处理？请给出改进方案。
  2. 基础示例程序中，两个线程并发修改共享变量，为何说"结果正确纯属巧合"？如何修复该数据竞争问题？

版本信息

项目说明
更新日期2026-06-09
目标内核Linux 7.0
关键 APIpthread_create() (POSIX.1, Linux 2.0+); pthread_join() (POSIX.1); pthread_detach() (POSIX.1); pthread_mutex_* (POSIX.1); pthread_cond_* (POSIX.1)
备注Linux 7.0 中 pthread 实现基于 NPTL（Native POSIX Thread Library），支持 1:1 线程模型；pidfd 系列系统调用（Linux 5.4+）提供了线程级别的文件描述符管理；futex2 系统调用（Linux 6.x 引入）优化了用户态锁性能