线程池已经限制了线程数，为什么服务器仍然会被请求压垮？

时间：2026/05/04

关键词：pthread、有界队列、背压、线程池、进程池、fork、waitpid、连接所有权

为了避免“每来一个请求就创建一个线程”，我们通常会提前创建固定数量的 worker：

请求 → 任务队列 → 8 个 worker 线程

线程数已经固定为 8，看起来并发资源有了上限。可压测一开始，内存仍然持续上涨，接口延迟从几十毫秒变成几十秒，最终进程被 OOM Killer 终止。

原因常常藏在没有画出来的地方：任务队列没有上限。

每秒到达 1000 个任务
每秒只能处理 800 个任务
               ↓
队列每秒净增加 200 个任务
               ↓
排队内存与等待时间持续增长

固定 worker 数只限制了“同时执行多少任务”，并没有限制“系统愿意积压多少任务”。一个可用的池化模型必须同时定义：

• worker 数量；
• 队列容量；
• 队列满时的拒绝或反压策略；
• 任务截止时间；
• 停止接收、排空与 join 的退出流程。

本文会先实现一个可运行的 POSIX 有界线程池，再讨论它怎样嵌入网络服务器，以及什么时候应该换成多进程或进程池。

---

一、为什么无限队列只是把过载问题推迟了？

假设平均到达速率为 λ，线程池最大处理速率为 μ。当 λ > μ 持续发生时，不存在一种队列能永久吸收差值：

积压增长速度 ≈ λ - μ

无限队列带来的“暂时不拒绝”会转化成：

• 每个任务及其请求数据占用内存；
• 新任务排在大量已经过期的任务后面；
• 客户端超时重试，进一步放大流量；
• 关停时需要处理或丢弃巨量积压；
• 最终以 OOM 或全局超时的方式拒绝所有请求。

有界队列不会凭空增加吞吐量。它的价值是把不可避免的过载变成及时、局部、可观测的拒绝，而不是让整个服务在更晚的时候一起失败。

因此，“线程池大小”与“任务队列容量”必须一起评审。

---

二、多线程、多进程和事件循环分别解决什么问题？

模型	隔离性	状态共享	主要成本	常见用途
单线程 Reactor	进程内无并行隔离	最简单	回调不能阻塞	大量轻量网络 I/O
多线程	一个线程崩溃常影响整个进程	直接共享内存	锁、竞态、栈与调度	CPU 任务、阻塞库调用
多进程	worker 之间隔离较强	需要 IPC	进程资源、序列化与通信	不可信任务、故障隔离
I/O 线程 + worker 池	连接与计算职责分离	通过有界队列传递	所有权与回投设计	常见高并发服务器

选择并发模型时，先问任务会阻塞什么、需要共享什么、失败时允许影响谁。

例如，纯事件驱动 I/O 不一定需要线程池；把一个 100 微秒的轻量解析任务投递到 worker，可能让排队、唤醒和 cache 切换比计算本身更贵。反过来，压缩、图像处理或同步数据库调用若直接放在 I/O 线程，则会让该线程管理的所有连接一起停顿。

---

三、一个最小线程池必须定义哪些状态？

本文的线程池只有这些组件：

             try_submit()
生产者 ─────────────────────▶ 固定容量环形队列
                                  │
                              has_work 条件变量
                                  │
                       ┌──────────┴──────────┐
                    worker 0             worker 1

共享状态全部由同一把 mutex 保护：

• head / tail：环形队列读写位置；
• count：当前任务数；
• accepting：是否仍接受新任务；
• stopping：worker 是否应在队列排空后退出。

提交采用非阻塞的 try_submit()：队列满就立即返回拒绝。这样调用者能根据请求类型决定返回错误、暂停读取或丢弃过期状态，而不是把 I/O 线程堵在“等待队列空位”上。

---

四、最小可运行版本：固定 worker、固定队列、排空退出

下面的程序兼容 Linux 与 macOS 的 POSIX threads，使用 C11。它创建 2 个 worker 和容量为 3 的队列，快速提交 12 个模拟任务；队列满时立即拒绝，最后停止接收并排空已接受任务。

由于线程调度具有不确定性，每次运行具体接受数量可能不同，但必须满足：

accepted + rejected == 12
completed == accepted

#define _POSIX_C_SOURCE 200809L

#include <pthread.h>

#include <errno.h>
#include <stdatomic.h>
#include <stddef.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

typedef void (*task_function)(void *argument);

struct task {
    task_function function;
    void *argument;
};

struct thread_pool {
    pthread_mutex_t mutex;
    pthread_cond_t has_work;
    pthread_t *threads;
    struct task *queue;
    size_t thread_count;
    size_t queue_capacity;
    size_t head;
    size_t tail;
    size_t count;
    int accepting;
    int stopping;
};

static void *worker_main(void *argument) {
    struct thread_pool *pool = argument;

    for (;;) {
        pthread_mutex_lock(&pool->mutex);

        while (pool->count == 0 && !pool->stopping) {
            pthread_cond_wait(&pool->has_work, &pool->mutex);
        }

        if (pool->count == 0 && pool->stopping) {
            pthread_mutex_unlock(&pool->mutex);
            return NULL;
        }

        struct task task = pool->queue[pool->head];
        pool->head = (pool->head + 1) % pool->queue_capacity;
        pool->count -= 1;

        pthread_mutex_unlock(&pool->mutex);
        task.function(task.argument);
    }
}

static void pool_release_storage(struct thread_pool *pool) {
    free(pool->queue);
    free(pool->threads);
    pthread_cond_destroy(&pool->has_work);
    pthread_mutex_destroy(&pool->mutex);
}

static int pool_init(struct thread_pool *pool,
                     size_t thread_count,
                     size_t queue_capacity) {
    if (thread_count == 0 || queue_capacity == 0) {
        return EINVAL;
    }

    memset(pool, 0, sizeof(*pool));
    int error = pthread_mutex_init(&pool->mutex, NULL);
    if (error != 0) {
        return error;
    }

    error = pthread_cond_init(&pool->has_work, NULL);
    if (error != 0) {
        pthread_mutex_destroy(&pool->mutex);
        return error;
    }

    pool->threads = calloc(thread_count, sizeof(*pool->threads));
    pool->queue = calloc(queue_capacity, sizeof(*pool->queue));
    if (pool->threads == NULL || pool->queue == NULL) {
        pool_release_storage(pool);
        return ENOMEM;
    }

    pool->thread_count = thread_count;
    pool->queue_capacity = queue_capacity;
    pool->accepting = 1;

    size_t created = 0;
    for (; created < thread_count; ++created) {
        error = pthread_create(&pool->threads[created], NULL,
                               worker_main, pool);
        if (error != 0) {
            pthread_mutex_lock(&pool->mutex);
            pool->accepting = 0;
            pool->stopping = 1;
            pthread_cond_broadcast(&pool->has_work);
            pthread_mutex_unlock(&pool->mutex);

            for (size_t i = 0; i < created; ++i) {
                pthread_join(pool->threads[i], NULL);
            }
            pool_release_storage(pool);
            return error;
        }
    }
    return 0;
}

static int pool_try_submit(struct thread_pool *pool,
                           task_function function,
                           void *argument) {
    int accepted = 0;
    pthread_mutex_lock(&pool->mutex);

    if (pool->accepting && pool->count < pool->queue_capacity) {
        pool->queue[pool->tail] = (struct task){function, argument};
        pool->tail = (pool->tail + 1) % pool->queue_capacity;
        pool->count += 1;
        accepted = 1;
        pthread_cond_signal(&pool->has_work);
    }

    pthread_mutex_unlock(&pool->mutex);
    return accepted;
}

static void pool_shutdown_and_join(struct thread_pool *pool) {
    pthread_mutex_lock(&pool->mutex);
    pool->accepting = 0;
    pool->stopping = 1;
    pthread_cond_broadcast(&pool->has_work);
    pthread_mutex_unlock(&pool->mutex);

    for (size_t i = 0; i < pool->thread_count; ++i) {
        pthread_join(pool->threads[i], NULL);
    }
    pool_release_storage(pool);
}

struct demo_task {
    int id;
    atomic_int *completed;
};

static void do_demo_work(void *argument) {
    struct demo_task *task = argument;
    struct timespec delay = {.tv_sec = 0, .tv_nsec = 100000000L};

    while (nanosleep(&delay, &delay) < 0 && errno == EINTR) {
    }
    atomic_fetch_add_explicit(task->completed, 1, memory_order_relaxed);
}

int main(void) {
    enum { TOTAL_TASKS = 12 };
    struct thread_pool pool;
    int error = pool_init(&pool, 2, 3);
    if (error != 0) {
        fprintf(stderr, "pool_init: %s\n", strerror(error));
        return EXIT_FAILURE;
    }

    atomic_int completed = 0;
    struct demo_task tasks[TOTAL_TASKS];
    int accepted = 0;
    int rejected = 0;

    for (int i = 0; i < TOTAL_TASKS; ++i) {
        tasks[i] = (struct demo_task){.id = i, .completed = &completed};
        if (pool_try_submit(&pool, do_demo_work, &tasks[i])) {
            accepted += 1;
        } else {
            rejected += 1;
        }
    }

    pool_shutdown_and_join(&pool);
    int finished = atomic_load_explicit(&completed, memory_order_relaxed);
    printf("accepted=%d rejected=%d completed=%d\n",
           accepted, rejected, finished);

    return accepted + rejected == TOTAL_TASKS && finished == accepted
               ? EXIT_SUCCESS
               : EXIT_FAILURE;
}

编译运行：

cc -std=c11 -O2 -Wall -Wextra -Wpedantic -pthread \
  bounded_pool.c -o bounded_pool
./bounded_pool

一种可能的输出是：

accepted=5 rejected=7 completed=5

具体数字由提交线程和 worker 的调度时机决定。示例验证的不是固定拒绝数，而是队列不会突破容量，且退出前会完成所有已接受任务。

---

五、条件变量为什么必须配合 while？

worker 等待任务的代码是：

while (pool->count == 0 && !pool->stopping) {
    pthread_cond_wait(&pool->has_work, &pool->mutex);
}

pthread_cond_wait() 会原子地释放 mutex 并进入等待，返回前重新取得 mutex。但“被唤醒”不等于“条件现在一定成立”：

• 条件变量允许虚假唤醒；
• 多个 worker 被唤醒后，另一个线程可能先取走任务；
• shutdown 的广播也会唤醒没有任务的 worker。

因此条件变量只表示“某件相关的事可能变化了”，真正的谓词仍要在锁内重新检查。若把 while 写成 if，worker 可能从空队列中取任务。

任务函数必须在解锁后执行。若持锁运行任务，所有 worker 实际会串行执行，提交线程也无法入队，线程池就退化成一把大锁。

---

六、队列满了以后，应该拒绝、等待还是丢弃？

没有适合所有任务的统一策略：

任务类型	队列满时可考虑	关键前提
HTTP 查询	快速返回 429/503	客户端有合理退避
玩家位置快照	丢弃旧状态或覆盖	新状态能替代旧状态
支付命令	拒绝并由客户端安全重试	请求有幂等键与结果查询
后台批处理	阻塞或持久化到外部队列	不阻塞关键 I/O 线程
内部控制任务	预留独立容量	防止被普通流量饿死

让提交者阻塞等待空位看起来不会丢任务，但如果提交者就是唯一 I/O 线程，它一旦阻塞，就不能继续读取响应、处理超时或释放连接，可能形成全局停顿。

更常见的网络服务策略是：入口尽早判断容量，拒绝低优先级请求，或暂时禁止对应连接的读事件，把背压传回客户端。还应设置每用户或每租户配额，避免一个来源占满全局队列。

所有拒绝都要可观测：至少记录队列深度、等待时长、拒绝数、任务执行时长和各类超时。只有平均值不够，尾延迟更能暴露排队问题。

---

七、任务截止时间为什么要包含排队时间？

假设客户端允许请求在 500 毫秒内完成。任务已在队列等待 490 毫秒，即使执行只需 50 毫秒，它也注定无法按时返回。

如果 worker 只从“开始执行”计时，就会继续浪费资源处理客户端已经放弃的工作。更合理的任务元数据包括：

request_id
enqueue_time
deadline
cancellation state
connection owner

worker 取出任务后先检查 deadline。对于可以安全取消的任务，过期就直接结束；对于已经产生外部副作用的操作，则必须依据业务协议完成、补偿或查询状态，不能粗暴终止线程。

不要使用 pthread_cancel() 作为通用业务取消机制。取消点、锁状态和清理处理器很容易让共享状态复杂化。协作式取消——任务定期检查停止标志或 deadline——通常更容易推理。

---

八、为什么 worker 不应该随意操作连接 socket？

常见服务器结构是：

I/O 线程
  ├─ accept / read / 协议拆包
  ├─ 把纯业务输入投递到有界池
  └─ 管理连接状态与输出缓冲

worker
  ├─ 执行 CPU 或阻塞业务
  └─ 把结果回投给原 I/O 线程

如果 worker 直接关闭或写入 socket，而 I/O 线程同时处理该 fd，可能发生：

• 连接状态竞态；
• 两个响应交错写入；
• fd 关闭后被内核快速复用，旧任务误操作新连接；
• 输出缓冲和事件注册在不同线程中失去一致性。

更清晰的所有权模型是：一个连接始终归固定 I/O 线程，worker 只持有连接 ID、请求 ID 和不可变业务数据。结果回投时，I/O 线程验证连接代次仍匹配，再决定是否发送。

回投队列同样必须有界。只限制“去 worker”的任务队列，却让“回 I/O”的结果队列无限增长，仍会在慢客户端场景下耗尽内存。

---

九、线程数是不是等于 CPU 核数就最好？

没有脱离工作负载的万能公式。

CPU 密集任务通常从接近可用 CPU 核数开始测试。线程远多于核心会增加调度与 cache 抖动，却不能增加算力。包含大量阻塞 I/O 的任务可以使用更多线程，但线程栈、数据库连接、文件描述符和下游容量都会形成新的上限。

不要把 CPU 任务和长时间阻塞任务混进同一个小池：

2 个 worker
  ├─ worker 0 等慢数据库
  └─ worker 1 等外部 HTTP

短 CPU 任务只能在队列中等待

如果两类任务的资源和延迟特征明显不同，可以使用独立的有界池和容量预算。但不要为每个函数建立一个线程池；池越多，越难控制总线程数和跨池排队。

正确做法是以目标机器和真实请求分布做压测，观察吞吐、CPU 利用率、上下文切换、队列等待与尾延迟，再调整线程和队列容量。

---

十、多进程模型解决了线程池解决不了的什么问题？

进程的主要价值是隔离，而不是“肯定更快”。fork() 后，子进程获得父进程地址空间的逻辑副本，现代系统通常通过写时复制（copy-on-write）延迟实际页面复制；已打开的 fd 也会被继承。

典型进程池：

master
  ├─ 监听、健康检查、worker 管理
  ├─ worker 0：独立地址空间
  ├─ worker 1：独立地址空间
  └─ worker 2：独立地址空间

某个 worker 因解析不可信输入而崩溃时，master 可以记录退出状态、限制重启频率并拉起替代进程。它适合不可信插件、编解码任务、多租户隔离和崩溃风险较高的组件。

代价是跨进程状态必须通过 IPC 明确传递。常见方式包括：

• pipe：简单的单向字节流；
• Unix domain socket：双向消息，并可通过 SCM_RIGHTS 传递 fd；
• 共享内存：少复制，但仍需进程间同步和崩溃恢复；
• eventfd、信号量：通知或资源计数；
• 消息队列：由内核或外部系统保存消息边界。

通过 Unix domain socket 传 fd，传递的是对同一个打开文件描述（open file description）的引用，不是复制 socket 中的数据。master 可以统一 accept，再把连接分给 worker；另一种方案是多个 worker 共享监听 socket，或使用 SO_REUSEPORT，具体负载均衡语义需结合内核和部署环境验证。

---

十一、fork、exec 和僵尸进程有哪些隐藏边界？

exec 系列用新程序映像替换当前进程。成功后不会返回；未设置 close-on-exec 的 fd 会被新程序继承。因此监听 socket、密钥文件和内部 pipe 应在创建时优先使用 O_CLOEXEC、SOCK_CLOEXEC 等标志。

多线程进程调用 fork() 更危险：子进程只保留调用 fork() 的线程，其他线程消失，但它们当时持有的 mutex 状态可能被复制下来。POSIX 要求子进程在 fork() 后、execve() 前只能调用异步信号安全函数。需要启动外部程序时，优先评估 posix_spawn()，或把 fork/exec 严格集中到可控阶段。

子进程退出后，父进程必须读取其退出状态，否则会留下僵尸进程。常见非阻塞回收循环是：

#include <errno.h>
#include <sys/wait.h>

for (;;) {
    pid_t pid = waitpid(-1, NULL, WNOHANG);
    if (pid > 0) {
        continue;
    }
    if (pid < 0 && errno == EINTR) {
        continue;
    }
    break;
}

通常由 SIGCHLD 通知主循环再执行回收。master 还要区分正常退出、配置错误、崩溃和被 OOM 杀死，避免错误配置导致 worker 高速重启。

---

十二、优雅关闭一个池，为什么顺序很重要？

本文示例采用“停止接收新任务，但完成已接受任务”的 drain 策略：

accepting = false
  → stopping = true
  → broadcast 唤醒所有空闲 worker
  → worker 继续取完队列
  → 队列为空后退出
  → 主线程 pthread_join
  → 销毁条件变量、mutex 和队列

如果先销毁队列或锁，再让 worker 退出，就会产生 use-after-free。如果只设置停止标志却不 broadcast，空闲 worker 可能永远睡在条件变量上，join() 随之卡死。

生产服务还必须设置关闭 deadline。到期后如何处理剩余任务取决于业务：无副作用计算可以丢弃；事务型任务则可能需要持久化、转移或等待外部结果。线程不能像独立 worker 进程那样被安全地强制杀死而不影响整个地址空间，这也是强隔离任务适合进程池的原因之一。

示例为保持聚焦，没有实现重复 shutdown、提交者并发销毁、任务析构回调和运行时扩缩容；若项目需要这些能力，优先使用已有且经过测试的执行器，而不是不断扩展教学线程池。

---

十三、常见误区

误区 1：固定线程数等于资源有界

无限任务队列、结果队列和每任务内存仍可无限增长。资源上限必须覆盖完整数据流。

误区 2：队列越长，吞吐量越高

吞吐由处理能力决定。过长队列主要增加等待和过期任务，不会增加 CPU 或下游容量。

误区 3：条件变量被唤醒就一定有任务

唤醒只是提示状态可能变化，必须持锁用 while 重新检查谓词。

误区 4：线程池 worker 越多越快

CPU、锁、内存带宽、下游连接池和调度成本都会限制收益。需要基于指标调优。

误区 5：进程之间完全没有共享

fd、映射和某些内核对象可以共享或传递，但共享必须通过明确 IPC 与同步建立，不能像线程那样直接使用普通指针。

误区 6：worker 崩溃后立即无限重启最可靠

持续崩溃可能来自坏配置或确定性输入。master 应限速、熔断并保留诊断信息。

---

十四、什么时候选择线程池，什么时候选择进程池？

优先考虑线程池的场景：

• 任务需要低成本共享只读数据或缓存；
• 任务代码可信，崩溃可接受影响整个进程；
• 需要把 CPU 计算或有限阻塞操作移出 I/O 线程；
• 能明确控制队列、deadline 和协作式取消。

优先考虑进程池的场景：

• 处理不可信输入、插件或第三方库；
• 单任务可能崩溃、泄漏或长期卡死；
• 需要不同 UID、capability 或文件系统视图；
• 愿意承担 IPC、序列化和 worker 管理成本。

如果任务本身很轻、完全非阻塞，单线程或多 I/O 线程 Reactor 可能比任何池都简单。并发模型的目标不是“使用更多线程或进程”，而是让隔离、所有权、排队和过载行为都可预测。

---

十五、总结

回到开头：固定线程数后服务器仍被压垮，是因为执行并发受限了，等待并发却没有受限。

最重要的结论是：

1. 线程池必须同时限制 worker 数和任务队列容量；
2. 队列满时要有与业务语义匹配的拒绝、覆盖或背压策略；
3. 条件变量只负责通知，实际条件必须在 mutex 下用 while 检查；
4. 网络连接最好由固定 I/O 线程持有，worker 通过有界队列接收任务并回投结果；
5. 线程池共享方便，进程池隔离更强，选择依据是失败边界而不只是性能；
6. 停止接收、排空任务、唤醒 worker、join 和释放资源必须按顺序完成。

可以直接用于实践的一条建议是：上线前把任务队列容量临时调小并主动制造过载，确认服务会快速、可观测地拒绝部分请求，而不是内存上涨或事件循环卡死。

---

参考资料

• Linux pthreads(7) 手册
• POSIX 条件变量：pthread_cond_wait(3)
• Linux fork(2) 手册
• Linux waitpid(2) 手册
• Linux Unix domain socket 手册