从零开始：C++ 线程池 TCP 服务器实战（续篇）

引言

上一篇教程实现的基础多线程 TCP 服务器，通过 “主线程 accept + 子线程处理客户端” 的模型解决了单客户端阻塞问题，但仍存在明显局限：

• 客户端数量激增时（如数千连接），“一个连接一个线程” 会频繁创建 / 销毁线程，系统调用开销占比骤升；
• 每个线程默认占用 8MB 栈空间（Linux 环境），过多线程会直接耗尽进程内存，导致 pthread_create 失败；
• 线程创建峰值可能触发系统资源限制（如 /proc/sys/kernel/threads-max），导致服务不可用。

为解决这些问题，工业级高并发服务通常采用线程池模型：预先创建固定数量的工作线程，通过 “任务队列” 调度客户端请求，避免频繁线程切换与内存浪费。

本文将基于前序教程的 TcpServer 类，续写线程池版 TCP 服务器，核心完成：

• 基于 std::mutex 封装 RAII 风格锁（Lock.hpp），保障线程安全；
• 实现通用线程池（ThreadPool.hpp），支持任务入队与线程复用；
• 改造 TcpServer 类，将客户端连接封装为任务提交至线程池，实现 “主线程收连接 + 线程池处理业务” 的高并发模型。

全文延续模块化设计与清晰的码风，确保代码可直接编译运行，并兼容前序教程的客户端。

1. 核心设计：线程池模型的原理与优势

1.1 线程池解决的核心问题

基础多线程模型的痛点本质是 “线程生命周期与连接生命周期强绑定”，线程池通过线程复用与任务解耦解决该问题：

问题场景	基础多线程模型	线程池模型
线程创建 / 销毁开销	每个连接触发一次	仅初始化时创建固定线程
线程数量控制	无限制（易超资源）	固定线程数（可控）
线程上下文切换	频繁（连接增减时）	低（线程数量稳定）
内存占用	随连接数线性增长	固定（线程栈内存不增加）

1.2 线程池核心逻辑链

线程池版 TCP 服务器的核心流程分为三层，完全解耦 “连接接收” 与 “业务处理”：

1. 主线程（TcpServer）：仅负责 TCP 核心流程（创建套接字→绑定→监听→accept），不处理任何业务逻辑；每次 accept 成功后，将 “客户端通信” 封装为任务，提交至线程池的任务队列。
2. 线程池（ThreadPool）：初始化时创建 N 个工作线程，所有线程阻塞等待任务队列的通知；当任务入队时，通过条件变量唤醒一个工作线程，执行任务逻辑。
3. 工作线程：从任务队列获取 “客户端信息（client_fd/IP/ 端口）”，调用业务处理函数（如 HandleClient）完成收发数据，任务结束后不退出，继续等待下一个任务。

关键保障：

• 线程安全：任务队列的读写通过 std::mutex 加锁，避免多线程竞争；
• 无死锁：通过 RAII 锁自动释放锁资源，条件变量避免线程空等；
• 优雅启停：线程池支持 Stop 接口，确保所有任务执行完毕后再销毁线程。

2. 基础组件封装：线程安全锁与线程池

2.1 Lock.hpp：RAII 风格锁封装（基于 std::mutex）

为避免手动加锁 / 解锁导致的死锁问题，封装 RAII 风格的锁工具类 ——MutexLock 封装 std::mutex，LockGuard 封装 “构造加锁、析构解锁” 的自动逻辑。

#ifndef __LOCK_HPP__
#define __LOCK_HPP__#include <mutex>// 封装 std::mutex，提供基础加锁/解锁接口
class MutexLock {
public:MutexLock() = default;~MutexLock() = default;// 禁用拷贝构造与赋值（避免资源被复制）MutexLock(const MutexLock&) = delete;MutexLock& operator=(const MutexLock&) = delete;// 加锁（阻塞式）void Lock() {_mutex.lock();}// 解锁void Unlock() {_mutex.unlock();}// 获取原生 mutex（供 condition_variable 使用）std::mutex& GetNativeMutex() {return _mutex;}
private:std::mutex _mutex;
};// RAII 风格锁：构造时加锁，析构时解锁
class LockGuard {
public:// 构造时自动加锁explicit LockGuard(MutexLock& mutex): _mutex(mutex) {_mutex.Lock();}// 析构时自动解锁~LockGuard() {_mutex.Unlock();}// 禁用拷贝（避免锁提前释放）LockGuard(const LockGuard&) = delete;LockGuard& operator=(const LockGuard&) = delete;// 获取内部持有的 MutexLock 引用MutexLock& GetMutex() const {return _mutex;}
private:MutexLock& _mutex;  // 引用外部的 MutexLock，避免拷贝
};#endif /* __LOCK_HPP__ */

关键设计说明：

• MutexLock 禁用拷贝构造，避免锁资源被意外复制导致的多线程安全问题；
• LockGuard 仅通过引用持有 MutexLock，确保析构时解锁的是同一个锁；
• 提供 GetNativeMutex 接口，方便后续与 std::condition_variable 配合使用（条件变量需操作原生 std::mutex）。

2.2 ThreadPool.hpp：通用线程池实现

线程池需支持 “初始化线程数、提交任务、优雅停止” 三大核心能力，采用任务队列 + 工作线程 + 条件变量的经典设计，兼容任意无返回值的任务（基于 std::function）。

#ifndef __THREAD_POOL_HPP__
#define __THREAD_POOL_HPP__#include <vector>
#include <queue>
#include <thread>
#include <functional>
#include <condition_variable>
#include <iostream>
#include "Lock.hpp"// 任务类型定义：无参数、无返回值的函数
using Task = std::function<void()>;class ThreadPool {
public:// 构造函数：初始化线程池（指定工作线程数量）explicit ThreadPool(size_t num): _thread_num(num), _is_running(false) {}// 析构函数：确保线程池优雅停止~ThreadPool() {if (_is_running) {Stop();}}// 禁用拷贝（线程池不可复制）ThreadPool(const ThreadPool&) = delete;ThreadPool& operator=(const ThreadPool&) = delete;// 启动线程池：创建工作线程bool Start() {if (_is_running) {std::cout << "线程池已启动，无需重复调用" << std::endl;return false;}_is_running = true;// 创建 _thread_num 个工作线程，绑定 Worker 函数for (int i = 0; i < _thread_num; ++i) {_workers.emplace_back(&ThreadPool::Worker, this);std::cout << "线程池创建工作线程：" << _workers.back().get_id() << std::endl;}return true;}// 停止线程池：通知所有线程退出，等待线程结束void Stop() {{// 加锁修改运行状态，避免多线程竞争LockGuard lock(_mutex);_is_running = false;_cond.notify_all(); // 唤醒所有阻塞的工作线程}// 等待所有工作线程执行完毕for (auto& thread: _workers) {if (thread.joinable()) {thread.join();std::cout << "线程池工作线程退出：" << thread.get_id() << std::endl;}}_workers.clear();std::cout << "线程池已停止" << std::endl;}// 提交任务到任务队列bool AddTask(const Task& task) {if (!_is_running) {std::cerr << "线程池已停止，无法提交任务" << std::endl;return false;}// 加锁将任务入队，入队后唤醒一个工作线程LockGuard lock(_mutex);_task_queue.push(task);_cond.notify_one();return true;}    size_t GetThreadNum() const {return _thread_num;}
private:// 工作线程入口函数：循环获取任务并执行void Worker() {std::cout << "工作线程启动：" << std::this_thread::get_id() << std::endl;while (_is_running) {Task task;{// 用 std::unique_lock 包装原生 mutexstd::unique_lock<std::mutex> ulock(_mutex.GetNativeMutex());// 等待条件：队列非空 或 线程池停止_cond.wait(ulock, [this]() {return !_task_queue.empty() || !_is_running;});// 若线程池已停止，退出循环if (!_is_running) {break;}// 从任务队列获取任务task = _task_queue.front();_task_queue.pop();}// 执行任务（此时锁已释放，不阻塞其他线程）if (task) {try {task();} catch (const std::exception& e) {std::cerr << "任务异常: " << e.what() << std::endl;} catch (...) {std::cerr << "未知任务异常" << std::endl;}}}std::cout << "工作线程退出：" << std::this_thread::get_id() << std::endl;}
private:size_t _thread_num;                 // 工作线程数量bool _is_running;                   // 线程池运行状态std::vector<std::thread> _workers;  // 工作线程列表std::queue<Task> _task_queue;       // 任务队列MutexLock _mutex;                   // 保护任务队列的锁std::condition_variable _cond;      // 唤醒工作线程的条件变量
};#endif /* __THREAD_POOL_HPP__ */

核心逻辑说明：

1. 任务提交（AddTask）：
   • 加锁将任务入队，避免多线程同时修改队列；
   • 用 notify_one 唤醒一个工作线程（而非 notify_all），减少惊群效应（多个线程被唤醒但仅一个能获取任务）。
2. 工作线程（Worker）：
   • 循环判断线程池运行状态与任务队列是否为空，为空则阻塞在 _cond 上（释放锁，避免占用）；
   • 用 lambda 判断任务队列是否为空，避免虚假唤醒；
   • 任务执行前释放锁，确保任务处理不阻塞其他线程的任务入队。
3. 优雅停止（Stop）：
   • 加锁设置 _is_running = false，并唤醒所有线程；
   • 调用 thread.join() 等待所有线程执行完毕，避免线程资源泄漏。

3. Server 端改造：线程池版本实现

基于前序教程的 TcpServer 类，新增线程池成员，将 “创建子线程处理客户端” 改为 “提交任务至线程池”，核心逻辑仅需修改 TcpServer 的成员定义与 Start 函数。

3.1 TcpServer 类扩展（TcpServer.hpp）

新增线程池成员、调整构造函数参数（增加线程数），复用原有 HandleClient 函数作为任务逻辑。

#ifndef _TCP_SERVER_HPP_
#define _TCP_SERVER_HPP_#include <iostream>
#include <string>
#include <cstring>
#include <functional>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <csignal>
#include "ThreadPool.hpp"  // 引入线程池// 数据处理回调函数类型（与前序教程一致）
typedef std::function<std::string(const std::string&)> func_t;class TcpServer {
public:// 构造函数：新增线程池数量参数（thread_num）TcpServer(uint16_t port, func_t handler, size_t thread_num): _listen_fd(-1), _listen_port(port), _is_running(false), _data_handler(handler), _thread_pool(thread_num) {}  // 初始化线程池~TcpServer() {Stop();  // 析构时停止服务器与线程池}bool Init();    // 初始化：创建套接字、绑定、监听、启动线程池void Start();   // 启动：主线程接收连接，提交任务至线程池void Stop();    // 停止服务器private:// 客户端通信处理（与前序教程逻辑一致，改为私有成员函数）void HandleClient(int client_fd, const std::string& client_ip, uint16_t client_port);private:int _listen_fd;         // 监听套接字（与前序一致）uint16_t _listen_port;  // 监听端口（与前序一致）bool _is_running;       // 服务器运行状态（与前序一致）func_t _data_handler;   // 数据处理回调（与前序一致）ThreadPool _thread_pool;// 线程池实例（新增核心成员）
};

3.2 Init 函数：初始化线程池

在原有初始化逻辑（创建套接字、绑定、监听）的基础上，新增线程池启动逻辑，确保线程池在服务器接收连接前就绪。

bool TcpServer::Init()
{// 1. 忽略 SIGPIPE 信号signal(SIGPIPE, SIG_IGN);// 2. 创建套接字_listen_fd = socket(AF_INET, SOCK_STREAM, 0);if (_listen_fd == -1) {perror("socket 创建失败！");return false;}std::cout << "套接字创建成功，listen_fd: " << _listen_fd << std::endl;// 3. 填充服务器地址结构struct sockaddr_in server_addr;memset(&server_addr, 0, sizeof(server_addr)); // 清空内存避免随机值server_addr.sin_family = AF_INET; // IPV4 协议server_addr.sin_port = htons(_listen_port); // 本地字节序 -> 网络字节序server_addr.sin_addr.s_addr = htonl(INADDR_ANY); // 监听所有网卡// 4. 绑定套接字与地址int bind_ret = bind(_listen_fd, (struct sockaddr*)&server_addr, sizeof(server_addr));if (bind_ret == -1) {perror("绑定失败");close(_listen_fd);_listen_fd = -1;return false;}std::cout << "绑定成功，成功监听端口：" << _listen_port << std::endl;// 5. 开始监听连接int listen_ret = listen(_listen_fd, 5); // backlog=5if (listen_ret == -1) {perror("listen 失败");close(_listen_fd);_listen_fd = -1;return false;}std::cout << "监听中，等待客户端连接..." << std::endl;// 6. 新增：启动线程池if (!_thread_pool.Start()) {std::cerr << "线程池启动失败，服务器初始化终止" << std::endl;close(_listen_fd);_listen_fd = -1;return false;}std::cout << "线程池启动成功（工作线程数：" << _thread_pool.GetThreadNum() << "）" << std::endl;_is_running = true;return true;
}

3.3 Start 函数：主线程入队任务

主线程仅负责 accept 接收连接，将 “客户端通信” 封装为 Task 提交至线程池，不再创建新线程，核心逻辑大幅简化。

void TcpServer::Start() {if (!_is_running || _listen_fd == -1) {perror("服务器未初始化，无法启动");return;}// 主线程循环：仅接收连接，不处理业务while (_is_running) {struct sockaddr_in client_addr; // 存储客户端地址socklen_t client_addr_len = sizeof(client_addr);// 1. 接收客户端连接（阻塞）int client_fd = accept(_listen_fd, (struct sockaddr*)&client_addr, &client_addr_len);if (client_fd == -1) {perror("accept 失败！");continue; // 接收连接失败肯定不可能让整个服务器都退出啊，跳过该次循环就可以了}// 2. 解析客户端地址（网络字节序 -> 本地字节序）std::string client_ip = inet_ntoa(client_addr.sin_addr); // IP：网络字节序 -> 点分十进制uint16_t client_port = ntohs(client_addr.sin_port); // 端口：网络字节序 -> 本地字节序std::cout << "\n客户端连接成功：[" << client_ip << ":" << client_port << "]，client_fd: " << client_fd << std::endl;// 3. 封装任务并提交至线程池（核心修改）// 注意 lambda 生命周期问题Task task = [this, client_fd, client_ip, client_port]() {this->HandleClient(client_fd, client_ip, client_port);};if (!_thread_pool.AddTask(task)) {std::cerr << "任务提交失败，关闭客户端连接：[" << client_ip << ":" << client_port << "]" << std::endl;close(client_fd); // 失败时关闭客户端 FD}}
}

关键细节：
用 lambda 表达式封装任务：捕获 this 指针以调用 HandleClient，值捕获 client_fd/client_ip/client_port（避免主线程循环覆盖栈上变量导致的野指针问题）；
任务提交失败时主动关闭 client_fd：防止文件描述符泄漏（线程池停止时无法处理任务，需释放客户端连接资源）。

3.4 任务逻辑封装：HandleClient 复用

HandleClient 函数逻辑与前序多线程版本完全一致，负责与客户端收发数据，通信结束后关闭 client_fd，无需修改。

3.5 服务器入口（TcpServer.cc）

命令行参数新增 “线程数”，其余逻辑与前序一致，确保用户可灵活指定线程池大小。

#include <memory>
#include "TcpServer.hpp"// 打印用法提示
void Usage(std::string proc) {std::cerr << "Usage: " << proc << " <listen_port> <thread_num>" << std::endl;std::cerr << "示例：" << proc << " 8888 4" << std::endl;std::cerr << "说明：thread_num 建议设置为 CPU 核心数的 1~2 倍" << std::endl;
}// 自定义数据处理回调（与前序一致）
std::string DefaultDataHandler(const std::string& client_data) {return "TCP ThreadPool Server Response: " + client_data;
}int main(int argc, char* argv[]) {// 检查命令行参数（需传入端口与线程数）if (argc != 3) {Usage(argv[0]);return 1;}// 解析端口（1024~65535）uint16_t listen_port = std::stoi(argv[1]);if (listen_port < 1024 || listen_port > 65535) {std::cerr << "端口号无效！需在 1024~65535 之间" << std::endl;return 2;}// 解析线程数（1~1024，避免过多线程）size_t thread_num = std::stoi(argv[2]);if (thread_num < 1 || thread_num > 1024) {std::cerr << "线程数无效！需在 1~1024 之间" << std::endl;return 3;}// 创建服务器实例（智能指针自动释放资源）std::unique_ptr<TcpServer> tcp_server = std::make_unique<TcpServer>(listen_port, DefaultDataHandler, thread_num);// 初始化并启动服务器if (!tcp_server->Init()) {std::cerr << "服务器初始化失败，退出程序" << std::endl;return 4;}tcp_server->Start();return 0;
}

4. 客户端兼容性：无需修改

线程池仅改变服务器端的 “任务调度逻辑”，不影响 TCP 通信协议（三次握手、收发数据格式），因此前序教程实现的 TcpClient.cc 可直接复用，无需任何代码修改。

客户端核心逻辑回顾（与前序一致）：

1. 创建 TCP 套接字；
2. 调用 connect 连接服务器；
3. 循环输入数据并发送，接收服务器响应；
4. 输入 “exit” 时关闭连接并退出。

5. 编译与多客户端压测

5.1 Makefile 配置

因为使用了线程，所以需要链接 phtread 库，加上 -lpthread 就行。

5.2 压测步骤与日志验证

步骤 1：启动线程池服务器

指定监听端口 8888，线程池大小 4（建议与 CPU 核心数匹配）：

./tcpserver 8888 4

服务器启动日志（关键验证：线程池创建 4 个工作线程）：

套接字创建成功，listen_fd: 3
绑定成功，成功监听端口：8888
监听中，等待客户端连接...
线程池创建工作线程：140703344566016
线程池创建工作线程：140703336173312
线程池创建工作线程：140703327780608
线程池创建工作线程：140703319387904
工作线程启动：140703344566016
工作线程启动：140703336173312
工作线程启动：140703327780608
工作线程启动：140703319387904
线程池启动成功（工作线程数：4）

步骤 2：启动多个客户端（模拟高并发）

打开 8 个终端，分别启动客户端连接服务器：

# 每个终端执行（共 8 次）
./tcpclient 127.0.0.1 8888

客户端连接成功日志：

客户端创建套接字成功，client_fd: 3
已成功连接到服务器[127.0.0.1:8888]请输入发送给服务器的数据（输入"exit"退出）：

步骤 3：客户端发送数据，验证线程复用

每个客户端输入不同数据（如 “Client 1 Data”“Client 2 Data”…），观察服务器日志：
服务器日志（关键验证：8 个客户端任务被 4 个工作线程复用处理）：

# 客户端 1~8 连接成功
客户端连接成功：[127.0.0.1:51234]，client_fd: 4
客户端连接成功：[127.0.0.1:51235]，client_fd: 5
客户端连接成功：[127.0.0.1:51236]，client_fd: 6
客户端连接成功：[127.0.0.1:51237]，client_fd: 7
客户端连接成功：[127.0.0.1:51238]，client_fd: 8
客户端连接成功：[127.0.0.1:51239]，client_fd: 9
客户端连接成功：[127.0.0.1:51240]，client_fd: 10
客户端连接成功：[127.0.0.1:51241]，client_fd: 11# 工作线程处理任务（仅 4 个线程 ID 循环出现）
工作线程[140703344566016]开始处理客户端[127.0.0.1:51234]
工作线程[140703336173312]开始处理客户端[127.0.0.1:51235]
工作线程[140703327780608]开始处理客户端[127.0.0.1:51236]
工作线程[140703319387904]开始处理客户端[127.0.0.1:51237]
工作线程[140703344566016]收到[127.0.0.1:51234]的数据：Client 1 Data
工作线程[140703344566016]向[127.0.0.1:51234]发送响应：TCP ThreadPool Server Response: Client 1 Data
工作线程[140703336173312]收到[127.0.0.1:51235]的数据：Client 2 Data
工作线程[140703336173312]向[127.0.0.1:51235]发送响应：TCP ThreadPool Server Response: Client 2 Data
# ... 后续客户端 5~8 的数据由同一批工作线程处理 ...

步骤 4：停止服务器，验证优雅退出

按下 Ctrl+C 停止服务器，观察日志（所有工作线程正常退出）：

^C
工作线程准备退出：140703344566016
工作线程准备退出：140703336173312
工作线程准备退出：140703327780608
工作线程准备退出：140703319387904
线程池工作线程退出：140703344566016
线程池工作线程退出：140703336173312
线程池工作线程退出：140703327780608
线程池工作线程退出：140703319387904
线程池已停止
监听套接字已关闭

6. 线程池方案的优缺点

6.1 优势（对比基础多线程）

优势点	具体说明
资源开销低	仅初始化时创建线程，避免频繁创建 / 销毁的系统调用开销（节省 90%+ 线程调度成本）
内存占用可控	线程数量固定，不会因客户端激增导致栈内存耗尽（如 4 个线程仅占用～32MB 栈内存）
稳定性高	避免线程数量超系统限制（如 threads-max），减少服务崩溃风险
扩展性强	线程池可独立复用至其他模块（如后续的 UDP 服务器、HTTP 服务器）

6.2 局限（需注意的细节）

1. 任务队列瓶颈：若客户端连接数远超任务处理速度，任务队列会持续堆积，导致新连接无法及时处理；需设置队列最大长度，避免内存溢出。
2. 任务优先级问题：基础版线程池采用 “先进先出”（FIFO）调度，无法优先处理紧急任务（如心跳包、断开连接请求）。
3. CPU 密集型任务适配：若 HandleClient 是 CPU 密集型（如大数据计算），线程数设置超过 CPU 核心数会导致上下文切换增加，反而降低性能（建议线程数 = CPU 核心数）。

7. 后续扩展方向

基于当前线程池服务器，可进一步优化以满足工业级需求：

7.1 任务队列边界控制

给任务队列添加最大长度限制，队列满时采用 “阻塞等待” 或 “丢弃低优先级任务” 策略：

// ThreadPool.hpp 中修改 AddTask 函数
bool AddTask(const Task& task) {if (!_is_running) {std::cerr << "线程池已停止，无法提交任务" << std::endl;return false;}LockGuard lock(_mutex);// 新增：队列满时阻塞等待（或返回 false 丢弃任务）while (_task_queue.size() >= 1000) {  // 队列最大长度 1000std::this_thread::sleep_for(std::chrono::milliseconds(10));  // 等待 10ms 再尝试}_task_queue.push(task);_cond.notify_one();return true;
}

7.2 动态线程数调整

根据任务队列长度动态增减工作线程（如队列长度 > 50 时增加线程，< 10 时减少线程），平衡资源占用与处理速度。

7.3 任务优先级调度

将任务队列改为 “优先级队列”（std::priority_queue），给任务添加优先级标识（如 0~5），高优先级任务优先执行：

// 定义带优先级的任务
struct PriorityTask {int priority;  // 0：最低，5：最高Task task;// 优先级队列排序规则（优先级高的先出队）bool operator<(const PriorityTask& other) const {return priority < other.priority;}
};// 线程池任务队列改为 priority_queue
std::priority_queue<PriorityTask> _task_queue;

7.4 连接超时管理

在 HandleClient 中给 recv 设置超时（通过 setsockopt 设置 SO_RCVTIMEO），避免客户端断网导致工作线程长期阻塞：

// HandleClient 中新增超时设置
struct timeval timeout = {5, 0};  // 5 秒超时
setsockopt(client_fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout));

8. 总结

本文基于前序教程的 TCP 服务器，完成了线程池版改造，核心成果包括：

1. 封装了 RAII 风格的锁工具（Lock.hpp），简化线程安全代码编写；
2. 实现了通用线程池（ThreadPool.hpp），支持任务提交、线程复用、优雅停止；
3. 改造 TcpServer 类，实现 “主线程收连接 + 线程池处理业务” 的高并发模型，解决了基础多线程的资源耗尽问题。

通过本文的实践，你可以掌握线程池的核心设计思想，以及 TCP 服务器从 “单客户端” 到 “高并发” 的演进逻辑。后续可结合 “任务优先级”“动态线程数” 等扩展特性，逐步构建出支持上万并发连接的工业级 TCP 服务。