在Linux网络编程中,真正稳定的 TCP 通信并不只是“连接成功一次”这么简单.实际运行过程中,网络抖动、服务器重启、链路中断、端口异常等情况都可能导致连接断开.如果客户端没有合理的检测与重连机制,就很容易出现通信中断后无法恢复的问题.因此,理解 connect 的连接过程、断线检测以及自动重连机制,是写好 TCP 客户端程序的重要一步.相比UDP,TCP是一种面向连接、可靠传输、字节流的通信协议.在进行数据收发之前,客户端通常需要先通过
socket创建套接字,再使用connect主动向服务器发起连接.当连接失败或已经建立的连接异常断开时,客户端还需要根据实际情况完成资源释放、等待重试、重新建立连接以及恢复后续通信流程.不过,TCP 断线重连并不是简单地再次调用一次connect就能解决.在实际开发中,还需要考虑几个关键问题:如何判断连接已经失效？原有套接字是否还能继续使用？重连失败后应该立即重试还是延时重试？服务器重启后客户端如何自动恢复连接？又该如何通过 Linux 网络命令验证端口监听、连接状态和数据收发是否正常？ 本文将围绕 TCP Socket 编程实战 展开,从客户端与服务端的基本通信流程入手,重点讲解socket、bind、listen、accept、connect、send、recv等常用接口,并进一步分析connect失败后的处理方式、断线检测思路以及客户端自动重连机制.随后结合客户端访问Linux主机进行实际验证,通过网络命令检查 IP、端口和连接状态,并结合抓包与日志分析通信过程.通过本文的学习,你将能够完整掌握从 TCP 连接建立、数据通信、异常断开到自动重连和结果验证 的实践流程,并进一步理解客户端与 Linux 服务端之间的通信机制,为后续学习 心跳检测、超时控制、并发服务器、长连接管理以及高可靠网络程序设计 打下扎实基础.
目录
一、TCPSocket编程
TCP Socket 编程是网络通信开发的基础,主要用于实现两台设备或两个进程之间可靠、有序的双向数据传输.
常见应用包括：即时通信、服务器开发、设备通信、文件传输、工业控制和物联网.
1.TCP Socket 通信原理
TCP 通信采用典型的 Client(客户端)和 Server(服务端)模型.
TCP Socket 客户端与服务端基本流程
服务端一般按照以下顺序执行：
socket() → bind() → listen() → accept() → recv()/send() → close()
客户端一般按照以下顺序执行：
socket() → connect() → send()/recv() → close()
2.Python TCP Socket 完整示例
以本机 TCP 通信为例,服务器监听 127.0.0.1:8888，客户端连接后发送消息.
服务端（server.py）
import socketserver = socket.socket(socket.AF_INET, socket.SOCK_STREAM)server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)server.bind(("127.0.0.1", 8888))server.listen(5)print("服务器启动，等待客户端连接...")try: conn, addr = server.accept() print("客户端已连接:", addr) with conn: while True: data = conn.recv(1024) if not data: break print("收到:", data.decode("utf-8")) conn.sendall(b"Message received")finally: server.close()
客户端（client.py）
import socketwith socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client: client.connect(("127.0.0.1", 8888)) client.sendall("Hello TCP Server".encode("utf-8")) response = client.recv(1024) print("服务器响应:", response.decode("utf-8"))
运行方法
先启动服务器：
python server.py
再启动客户端：
python client.py
服务器输出：
服务器启动，等待客户端连接...
客户端已连接: ('127.0.0.1', 端口号)
收到: Hello TCP Server
客户端输出：
服务器响应: Message received
注意：这个示例用于演示基本通信,服务端一次只处理一个客户端连接.
3.必须掌握的几个核心概念
| 概念 | 说明 |
|---|---|
| IP 地址 | 标识网络中的主机 |
| Port 端口 | 标识主机上的通信服务 |
| Socket | 网络通信端点 |
| TCP 三次握手 | 建立 TCP 连接 |
| send / recv | 发送和接收数据 |
| 阻塞与非阻塞 | 决定 I/O 操作是否等待 |
| TCP 粘包 / 拆包 | TCP 是字节流，必须设计消息边界 |
| 多客户端并发 | 使用线程、select、epoll 或异步 I/O |
特别注意： TCP 保证字节流有序可靠,但不保证一次 send() 对应一次 recv().实际项目通常使用固定长度消息头或分隔符处理消息边界.
1.1TCP socket API详解
TCP Socket 编程的核心,是通过操作系统提供的API完成 创建套接字、建立连接、发送数据、接收数据、关闭连接.
下面以Linux+C语言(POSIX Socket API)为基础,系统讲解每个 API 的作用、函数参数、返回值、底层行为和实际使用方法.
1.TCP 服务端与客户端调用流程
服务端调用顺序：
socket()
↓
setsockopt() // 可选
↓
bind()
↓
listen()
↓
accept() // 等待客户端连接
↓
recv() / send()
↓
close()
客户端调用顺序：
socket()
↓
connect()
↓
send() / recv()
↓
close()
这里有一个关键区别：
- 监听 Socket（listen_fd）： 用于等待新的 TCP 连接。
- 通信 Socket（conn_fd）： 由
accept()返回，用于和某个客户端收发数据。
服务端可以保留同一个监听 Socket，连续接受多个客户端连接。
2.核心 API 速查表
| API | 作用 | 成功返回值 |
|---|---|---|
socket() | 创建 Socket | 文件描述符 fd |
bind() | 绑定本地 IP 和端口 | 0 |
listen() | 开始监听连接 | 0 |
accept() | 接受客户端连接 | 新的 fd |
connect() | 连接远程服务器 | 0 |
send() | 发送数据 | 实际发送字节数 |
recv() | 接收数据 | 实际接收字节数 |
setsockopt() | 设置 Socket 参数 | 0 |
getsockopt() | 获取 Socket 参数 | 0 |
shutdown() | 关闭连接的读写方向 | 0 |
close() | 关闭文件描述符 | 0 |
这些 API 出错时通常返回 -1,并通过 errno 表示错误原因.
socket() —— 创建套接字
函数原型：
#include <sys/socket.h>
int socket(int domain, int type, int protocol);
参数说明
| 参数 | 含义 | 常见取值 |
|---|---|---|
domain | 地址族 | AF_INET（IPv4）、AF_INET6（IPv6） |
type | Socket 类型 | SOCK_STREAM（TCP）、SOCK_DGRAM（UDP） |
protocol | 使用的协议 | 0（根据地址族和类型选择默认协议） |
TCP 创建方式：
int sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd == -1) {
perror("socket");
return -1;
}
返回值：
- 成功：返回非负整数文件描述符,例如
3. - 失败：返回
-1。
在 Linux 中,Socket也是文件描述符,能够使用 read()、write()、close() 等系统调用.
socket() 只创建通信端点,不会自动连接服务器.
bind() —— 绑定 IP 地址和端口
函数原型：
#include <sys/socket.h>
int bind(int sockfd,
const struct sockaddr *addr,
socklen_t addrlen);
参数：
sockfd：socket()返回的文件描述符.addr：要绑定的本地 IP 地址与端口.addrlen：地址结构体长度。
IPv4 地址结构：
struct sockaddr_in {
sa_family_t sin_family;
in_port_t sin_port;
struct in_addr sin_addr;
// 其他填充字段
};
使用示例：
#include <arpa/inet.h>
#include <netinet/in.h>
#include <string.h>
struct sockaddr_in addr;
memset(&addr, 0, sizeof(addr));
addr.sin_family = AF_INET;
addr.sin_port = htons(8888);
addr.sin_addr.s_addr = htonl(INADDR_ANY);
int ret = bind(sockfd,
(struct sockaddr *)&addr,
sizeof(addr));
if (ret == -1) {
perror("bind");
}
这里需要理解三个关键点.
① htons()：主机字节序转换为网络字节序
addr.sin_port = htons(8888);
htons：Host TO Network Short,转换16位整数.htonl：Host TO Network Long,转换32位整数.ntohs/ntohl：反向转换.
TCP/IP 使用网络字节序(大端序).
② INADDR_ANY：监听所有本地 IPv4 地址
addr.sin_addr.s_addr = htonl(INADDR_ANY);
表示监听本机所有 IPv4 网络接口.
如果只允许本机访问：
inet_pton(AF_INET, "127.0.0.1",
&addr.sin_addr);
对于有公网或内网网卡的服务器,两种绑定方式会直接影响服务能否被其他设备访问.
③ 为什么使用 (struct sockaddr *)？
因为 bind() 的接口使用通用地址结构 sockaddr,而 IPv4 实际上通过 sockaddr_in 传递地址信息.
bind() 成功返回 0,失败返回 -1.
listen() —— 开始监听连接
函数原型：
int listen(int sockfd, int backlog);
示例：
if (listen(sockfd, 128) == -1) {
perror("listen");
}
参数：
sockfd：已经绑定地址的 TCP Socket.backlog：等待被accept()取走的已建立连接队列长度的上限参数.
需要特别注意：
listen(sockfd, 128) 并不意味着服务端最多只能连接128个客户端.
在现代 Linux 中,backlog 主要限制已经完成 TCP 建连、但尚未被应用程序 accept() 的连接队列长度,实际生效值还受内核配置影响.
accept() —— 接受客户端连接
函数原型：
int accept(int sockfd,
struct sockaddr *addr,
socklen_t *addrlen);
示例：
struct sockaddr_in client_addr;
socklen_t len = sizeof(client_addr);
int conn_fd = accept(
sockfd,
(struct sockaddr *)&client_addr,
&len
);
if (conn_fd == -1) {
perror("accept");
}
调用成功后：
listen_fd = 3 // 原来的监听 Socket
conn_fd = 4 // 新建立的客户端连接
之后使用：
recv(conn_fd, ...);
send(conn_fd, ...);
而不是在 listen_fd 上读写业务数据.
阻塞行为：
默认阻塞模式下,如果没有已建立且等待接受的连接,accept() 会等待.
int conn_fd = accept(listen_fd, NULL, NULL);
上述写法用于不需要获取客户端地址的场景.
accept() 返回的新文件描述符对应一条已连接的 TCP Socket,监听Socket本身不受影响.
connect() —— 连接服务器
函数原型：
int connect(int sockfd,
const struct sockaddr *addr,
socklen_t addrlen);
客户端示例：
struct sockaddr_in server_addr = {0};
server_addr.sin_family = AF_INET;
server_addr.sin_port = htons(8888);
inet_pton(AF_INET, "127.0.0.1",
&server_addr.sin_addr);
int ret = connect(
sockfd,
(struct sockaddr *)&server_addr,
sizeof(server_addr)
);
if (ret == -1) {
perror("connect");
}
connect() 发起 TCP 连接建立过程.
阻塞 Socket 下,连接成功通常返回 0;连接失败返回 -1.
常见错误：
| errno | 含义 |
|---|---|
ECONNREFUSED | 连接被拒绝，常见于目标端口没有监听服务 |
ETIMEDOUT | 连接超时 |
EINPROGRESS | 非阻塞连接正在建立 |
ENETUNREACH | 网络不可达 |
EHOSTUNREACH | 目标主机不可达 |
特别地,非阻塞模式的 connect() 返回 EINPROGRESS 不表示连接失败.通常需要等待 Socket 可写,然后通过 getsockopt(SO_ERROR) 判断最终连接结果.
send() —— 发送 TCP 数据
函数原型：
ssize_t send(int sockfd,
const void *buf,
size_t len,
int flags);
参数：
sockfd：已连接 Socket.buf：待发送数据的内存地址.len：希望发送的字节数.flags：发送控制标志,常见为0.
示例：
const char *msg = "Hello TCP";
ssize_t n = send(
conn_fd,
msg,
strlen(msg),
0
);
假设 strlen(msg) == 9,send() 有可能返回 9,也可能只返回一个小于 9 的正数.
因为一次 send() 不保证发送完全部数据.
正确做法是处理部分发送（Partial Send）：
size_t total = 0;
size_t len = strlen(msg);
while (total < len) {
ssize_t n = send(
conn_fd,
msg + total,
len - total,
MSG_NOSIGNAL
);
if (n > 0) {
total += (size_t)n;
} else if (n < 0 && errno == EINTR) {
continue;
} else {
// 错误处理；本例假设阻塞 Socket
break;
}
}
上例使用 Linux 支持的 MSG_NOSIGNAL,避免向已经关闭的连接发送时触发 SIGPIPE 导致默认的进程终止行为.
最重要的概念：send() 成功不代表对方应用已经收到数据.
成功一般只说明内核接受了相应字节用于 TCP 发送,并不是业务级送达确认.
recv() —— 接收 TCP 数据
函数原型：
ssize_t recv(int sockfd,
void *buf,
size_t len,
int flags);
示例：
char buf[1024];
ssize_t n = recv(
conn_fd,
buf,
sizeof(buf),
0
);
返回值必须区分三种情况：
| 返回值 | 意义 |
|---|---|
n > 0 | 实际收到 n 字节 |
n == 0 | 对端已正常关闭发送方向（假定请求读取的长度大于 0） |
n == -1 | 出现错误，查看 errno |
标准处理：
char buf[1024];
while (1) {
ssize_t n = recv(conn_fd, buf, sizeof(buf), 0);
if (n > 0) {
// 收到 n 个字节
// 注意：buf 不一定以 '\0' 结尾
} else if (n == 0) {
// 对端正常结束发送
break;
} else if (errno == EINTR) {
continue;
} else {
perror("recv");
break;
}
}
recv() 默认收到当前可用的一部分数据就可能返回,不会等到缓冲区被填满.
例如：
recv(fd, buf, 1024, 0);
可能只返回 20 字节,不能认为是异常.
此外,TCP 接收的是字节流,不是天然分界的消息包.
shutdown() 与 close() —— 关闭连接
两个函数的作用不同.
int shutdown(int sockfd, int how);
int close(int fd);
shutdown() 可以按方向关闭 TCP 通信：
| 参数 | 作用 |
|---|---|
SHUT_RD | 禁止继续接收 |
SHUT_WR | 禁止继续发送 |
SHUT_RDWR | 禁止继续收发 |
示例：
shutdown(conn_fd, SHUT_WR);
表示本端不再发送数据,但仍然可以接收服务器的数据,这叫做 TCP 半关闭(Half-close).
close() 则关闭本进程持有的文件描述符：
close(conn_fd);
当该 Socket 不再被其他文件描述符引用时,会触发相应的连接关闭处理.
注意：close() 不等于“对端应用已完整处理所有数据”;如果业务要求可靠交付确认,还需要设计应用层应答机制.
setsockopt() —— 设置 Socket 属性
函数原型：
int setsockopt(int sockfd,
int level,
int optname,
const void *optval,
socklen_t optlen);
常用选项：
| 选项 | 用途 |
|---|---|
SO_REUSEADDR | 允许符合条件的本地地址重用，便于服务器重启 |
SO_REUSEPORT | 允许符合条件的多个 Socket 绑定相同地址端口 |
SO_KEEPALIVE | 开启 TCP 保活检测 |
SO_RCVTIMEO | 设置接收操作超时 |
SO_SNDTIMEO | 设置发送操作超时 |
SO_RCVBUF | 设置接收缓冲区 |
SO_SNDBUF | 设置发送缓冲区 |
TCP_NODELAY | 禁用 Nagle 算法，适合部分低延迟场景 |
示例：服务器设置地址重用.
int opt = 1;
setsockopt(
sockfd,
SOL_SOCKET,
SO_REUSEADDR,
&opt,
sizeof(opt)
);
通常在 bind() 之前调用.需要注意,SO_REUSEADDR 并不能让两个普通 TCP 服务端同时监听完全相同的 IP 和端口.
示例：开启 TCP Keepalive.
int enable = 1;
setsockopt(
sockfd,
SOL_SOCKET,
SO_KEEPALIVE,
&enable,
sizeof(enable)
);
Keepalive 用于检测长期空闲连接的可达性,并非应用层心跳的完全替代品.
fcntl() —— 设置非阻塞 Socket
默认创建的 Socket 一般处于阻塞模式.
例如：
recv(fd, buf, 1024, 0);
如果没有数据,默认会等待.
可以通过 fcntl() 设置非阻塞模式：
#include <fcntl.h>
int flags = fcntl(fd, F_GETFL, 0);
if (flags != -1) {
if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1) {
perror("fcntl");
}
}
非阻塞模式的特点：
| API | 暂时无法完成时的典型行为 |
|---|---|
accept() | 返回 -1，errno 为 EAGAIN / EWOULDBLOCK |
recv() | 返回 -1，errno 为 EAGAIN / EWOULDBLOCK |
send() | 不能立即发送更多数据时，可能返回相同错误 |
connect() | 连接正在进行时，返回 -1，errno 为 EINPROGRESS |
非阻塞编程通常需要结合事件通知机制,避免不断轮询造成 CPU 空转.
select() / poll() / epoll() —— I/O 多路复用
当服务器需要同时管理多个客户端时,常用 I/O 多路复用.
| API | 适用情况 | 主要特点 |
|---|---|---|
select() | 简单、小规模并发 | 兼容性广，但有文件描述符集合大小限制 |
poll() | 中小规模并发 | 无 select() 的固定 fd 集合限制 |
epoll() | Linux 大规模并发 | 使用事件通知管理大量 Socket |
Linux epoll 的核心 API：
int epoll_create1(int flags);
int epoll_ctl(int epfd, int op,
int fd, struct epoll_event *event);
int epoll_wait(int epfd,
struct epoll_event *events,
int maxevents,
int timeout);
基本工作模式：
epoll_create1()
↓
epoll_ctl(EPOLL_CTL_ADD)
↓
epoll_wait()
↓
检测就绪事件
↓
accept() / recv() / send()
↓
继续 epoll_wait()
epoll 支持 LT(水平触发)和 ET(边缘触发)模式.使用ET时,通常需要非阻塞Socket,并持续读取或写入直到出现 EAGAIN,避免遗漏未处理的数据.
1.2 V1—基于线程池的TCP Echo Server：模块解析与多客户端并发测试
//Cond.hpp
#ifndef __COND_HPP
#define __COND_HPP
#include <pthread.h>
#include "Mutex.hpp"
class Cond
{
public:
Cond()
{
pthread_cond_init(&_cond, nullptr);
}
void Wait(Mutex &mutex)
{
int n = pthread_cond_wait(&_cond, mutex.Ptr());
(void)n;
}
void Signal()
{
int n = pthread_cond_signal(&_cond);
(void)n;
}
void Broadcast()
{
int n = pthread_cond_broadcast(&_cond);
(void)n;
}
~Cond()
{
pthread_cond_destroy(&_cond);
}
private:
pthread_cond_t _cond;
};
#endif
//EchoTcpClient.cc
#include <iostream>
#include <string>
#include <cstdlib>
#include <unistd.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include "InetAddr.hpp"
static void Usage(const std::string &name)
{
std::cerr << "Usage:\n\t";
std::cerr << name << " server_ip server_port" << std::endl;
}
// ./client_tcp server_ip server_port
int main(int argc, char *argv[])
{
if (argc != 3)
{
Usage(argv[0]);
exit(1);
}
std::string server_ip = argv[1];
uint16_t server_port = std::stoi(argv[2]);
// 1. 创建tcpsocket套接字
int sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd < 0)
{
std::cerr << "socket error" << std::endl;
exit(2);
}
// 2. 要不要进行bind？需要
// 要不要显示的bind本地IP和端口？不需要自己手动显示的进行bind，OS会自动bind的！
// OS 帮助我们进行随机端口，防止端口号冲突
// 2. 应该干什么？bind端口号ip，发起建立连接的工作
// {client ip, client port} <-> {server ip, server port}
InetAddr serveraddress(server_port, server_ip);
int n = connect(sockfd, (struct sockaddr *)serveraddress.GetNetAddress(), serveraddress.Len());
if (n < 0)
{
std::cerr << "connect to " << serveraddress.ToString() << " failed!"<< std::endl;
exit(3);
}
std::cerr << "connect to " << serveraddress.ToString() << " success!" << std::endl;
// 3. 通信
while(true)
{
std::string line;
std::cout << "please Enter# ";
std::getline(std::cin, line);
write(sockfd, line.c_str(), line.size());
char inbuffer[1024];
ssize_t n = read(sockfd, inbuffer, sizeof(inbuffer));
if(n > 0)
{
inbuffer[n] = 0;
std::cout << inbuffer << std::endl;
}
else if(n == 0)
{
std::cout << "read enf of file!" << std::endl;
break;
}
else
{
std::cerr << "read error!" << std::endl;
break;
}
}
return 0;
}
//EchoTcpServer.hpp
#pragma once
#include <iostream>
#include <string>
#include <cstring>
#include <functional>
#include <unistd.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/wait.h>
// #include <signal.h>
#include <pthread.h>
#include "InetAddr.hpp"
#include "Logger.hpp"
#include "ThreadPool.hpp"
using namespace NS_LOG_MODULE;
using namespace NS_THREAD_POOL_MODULE;
using task_t = std::function<void()>;
enum
{
SUCCESS = 0,
USAGE_ERR,
SOCKET_ERR,
BIND_ERR,
LISTEN_ERR,
FORK_ERR
};
static const int gbacklog = 16;
static const uint16_t gport = 8888;
class TcpServer
{
public:
TcpServer(uint16_t port = gport) : _port(port)
{
}
void InitServer()
{
// 1. 创建socket
_listensockfd = socket(AF_INET, SOCK_STREAM, 0); // TCP
if (_listensockfd < 0)
{
LOG(LogLevel::FATAL) << "create socket error";
exit(SOCKET_ERR);
}
LOG(LogLevel::DEBUG) << "create socket success: " << _listensockfd;
// 2. 填充本地socket信息
struct sockaddr_in local;
memset(&local, 0, sizeof(local));
local.sin_family = AF_INET;
local.sin_port = htons(_port);
local.sin_addr.s_addr = INADDR_ANY; // 任意地址bind
// 3. bind
int n = bind(_listensockfd, (struct sockaddr *)&local, sizeof(local));
if (n < 0)
{
LOG(LogLevel::FATAL) << "bind error";
exit(BIND_ERR);
}
LOG(LogLevel::DEBUG) << "bind socket success";
// 4. tcp是面向连接的，所以，TCP服务器要处于一种叫做listen，监听状态。
n = listen(_listensockfd, gbacklog);
if (n < 0)
{
LOG(LogLevel::FATAL) << "listen error";
exit(LISTEN_ERR);
}
LOG(LogLevel::DEBUG) << "listen socket success";
}
void serviceIO(int sockfd, InetAddr address)
{
// 短连接，短服务
char inbuffer[1024] = {0};
// 读,字符串 -> "ls -a -l" -> "ls" "-a" "-l" -> argv -> fork() -> exec* ->得到结果->write
ssize_t n = read(sockfd, inbuffer, sizeof(inbuffer) - 1);
if (n > 0)
{
inbuffer[n] = 0;
LOG(LogLevel::INFO) << address.ToString() << " say# " << inbuffer;
// 写
std::string echo_string = "server echo# ";
echo_string += inbuffer;
write(sockfd, echo_string.c_str(), echo_string.size());
}
else if (n == 0)
{
LOG(LogLevel::INFO) << "client quit, address: " << address.ToString();
}
else
{
LOG(LogLevel::ERROR) << "client read error, address: " << address.ToString();
}
close(sockfd);
// tcp sockfd, 全双工
// 长服务，长连接, 小型应用。
// LOG(LogLevel::DEBUG) << "clinet info is : " << address.ToString();
// while (true)
// {
// char inbuffer[1024] = {0};
// // 读
// ssize_t n = read(sockfd, inbuffer, sizeof(inbuffer) - 1);
// if (n > 0)
// {
// inbuffer[n] = 0;
// LOG(LogLevel::INFO) << address.ToString() << " say# " << inbuffer;
// // 写
// std::string echo_string = "server echo# ";
// echo_string += inbuffer;
// write(sockfd, echo_string.c_str(), echo_string.size());
// }
// else if (n == 0)
// {
// LOG(LogLevel::INFO) << "client quit, address: " << address.ToString();
// break;
// }
// else
// {
// LOG(LogLevel::ERROR) << "client read error, address: " << address.ToString();
// break;
// }
// }
}
class ThreadData
{
public:
ThreadData(TcpServer *ts, int sockfd, const InetAddr &addr)
: _this(ts), _sockfd(sockfd), _addr(addr)
{
}
~ThreadData()
{
close(_sockfd);
}
public:
TcpServer *_this;
int _sockfd;
InetAddr _addr;
};
static void *thread_routine(void *args)
{
ThreadData *td = static_cast<ThreadData *>(args);
pthread_detach(pthread_self());
td->_this->serviceIO(td->_sockfd, td->_addr);
delete td;
return nullptr;
}
void Start()
{
// signal(SIGCHLD, SIG_IGN); // 最佳实践
while (true)
{
struct sockaddr_in clientaddr;
socklen_t len = sizeof(clientaddr);
// 5. 获取连接
int sockfd = accept(_listensockfd, (struct sockaddr *)&clientaddr, &len);
if (sockfd < 0)
{
LOG(LogLevel::WARNING) << "accept errr!";
continue;
}
LOG(LogLevel::DEBUG) << "accept success, sockfd: " << sockfd;
// version 3: 接入线程池
InetAddr clientaddress(clientaddr);
ThreadPool<task_t>::Instance()->Enqueue([this, sockfd, clientaddress]()->void{
this->serviceIO(sockfd, clientaddress);
});
// version 2: 多线程版本
// pthread_t tid;
// InetAddr clientaddress(clientaddr);
// ThreadData *td = new ThreadData(this, sockfd, clientaddress);
// pthread_create(&tid, nullptr, thread_routine, (void *)td);
// version 1: 多进程版本
// pid_t id = fork();
// if(id < 0)
// {
// LOG(LogLevel::FATAL) << "fork errr!";
// exit(FORK_ERR);
// }
// else if(id == 0)
// {
// // 关闭掉自己不需要的sockfd
// close(_listensockfd);
// // 子进程
// if(fork() > 0) exit(0); // 子进程直接退出
// // 孙子进程 - 孤儿进程
// InetAddr clientaddress(clientaddr);
// serviceIO(sockfd, clientaddress);
// close(sockfd);
// exit(0); // 重要
// }
// else{
// // 父进程
// // wait(id); waitpid: WNOHANG
// // 关闭掉自己不需要的sockfd
// close(sockfd);
// pid_t rid = waitpid(id, nullptr, 0);
// (void)rid;
// }
// 6. 处理新sockfd
// version 0
// InetAddr clientaddress(clientaddr);
// serviceIO(sockfd, clientaddress);
}
}
~TcpServer()
{
close(_listensockfd);
}
private:
uint16_t _port;
// 不需要显示包含ip
int _listensockfd;
};
//EchoTcpServerMain.cc
#include "EchoTcpServer.hpp"
#include <memory>
static void Usage(const std::string &process)
{
std::cerr << "Usage:\n\t";
std::cerr << process << " local_port" << std::endl;
}
int main(int argc, char *argv[])
{
if (argc != 2)
{
Usage(argv[0]);
exit(USAGE_ERR);
}
ENABLE_CONSOLE_LOG_STRATEGY();
uint16_t server_port = std::stoi(argv[1]);
std::unique_ptr<TcpServer> tsvr = std::make_unique<TcpServer>(server_port);
tsvr->InitServer();
tsvr->Start();
return 0;
}
//InetAddr.hpp
#pragma once
#include <iostream>
#include <string>
#include <strings.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
// 对客户端进行先描述
class InetAddr
{
public:
InetAddr(const struct sockaddr_in &address):_address(address), _len(sizeof(address))
{
_ip = inet_ntoa(_address.sin_addr);
_port = ntohs(_address.sin_port);
}
InetAddr(uint16_t port, const std::string &ip = "0.0.0.0"):_ip(ip), _port(port)
{
bzero(&_address, sizeof(_address));
_address.sin_family = AF_INET;
_address.sin_port = htons(_port); // h->n
_address.sin_addr.s_addr = inet_addr(_ip.c_str()); // 1. 字符串ip->4字节IP 2. hton
_len = sizeof(_address);
}
bool operator == (const InetAddr &addr)
{
return (this->_ip == addr._ip) && (this->_port == addr._port);
}
std::string ToString()
{
return "[" + _ip + ":" + std::to_string(_port) + "]";
}
InetAddr()
{}
struct sockaddr_in *GetNetAddress()
{
return &_address;
}
socklen_t Len()
{
return _len;
}
~InetAddr()
{}
private:
// net address
struct sockaddr_in _address;
socklen_t _len;
// host address
std::string _ip;
uint16_t _port;
};
//Logger.hpp
#ifndef __LOGGER_HPP
#define __LOGGER_HPP
#include <iostream>
#include <cstdio>
#include <string>
#include <memory>
#include <sstream>
#include <ctime>
#include <sys/time.h>
#include <unistd.h>
#include <filesystem> // C++17
#include <fstream>
#include "Mutex.hpp"
namespace NS_LOG_MODULE
{
enum class LogLevel
{
INFO,
WARNING,
ERROR,
FATAL,
DEBUG
};
std::string LogLevel2Message(LogLevel level)
{
switch (level)
{
case LogLevel::INFO:
return "INFO";
case LogLevel::WARNING:
return "WARNING";
case LogLevel::ERROR:
return "ERROR";
case LogLevel::FATAL:
return "FATAL";
case LogLevel::DEBUG:
return "DEBUG";
default:
return "UNKNOWN";
}
}
// 1. 时间戳 2. 日期+时间
std::string GetCurrentTime()
{
struct timeval current_time;
int n = gettimeofday(¤t_time, nullptr);
(void)n;
// current_time.tv_sec; current_time.tv_usec;
struct tm struct_time;
localtime_r(&(current_time.tv_sec), &struct_time); // r: 可重入函数
char timestr[128];
snprintf(timestr, sizeof(timestr), "%04d-%02d-%02d %02d:%02d:%02d.%ld",
struct_time.tm_year + 1900,
struct_time.tm_mon + 1,
struct_time.tm_mday,
struct_time.tm_hour,
struct_time.tm_min,
struct_time.tm_sec,
current_time.tv_usec);
return timestr;
}
// 输出角度 -- 刷新策略
// 1. 显示器打印
// 2. 文件写入
// 策略模式，策略接口
class LogStrategy
{
public:
virtual ~LogStrategy() = default;
virtual void SyncLog(const std::string &message) = 0;
};
// 控制台日志刷新策略, 日志将来要向显示器打印
class ConsoleStrategy : public LogStrategy
{
public:
void SyncLog(const std::string &message) override
{
LockGuard lockguard(_mutex);
std::cerr << message << std::endl; // ??
}
~ConsoleStrategy()
{
}
private:
Mutex _mutex;
};
const std::string defaultpath = "./log";
const std::string defaultfilename = "log.txt";
// 文件策略
class FileLogStrategy : public LogStrategy
{
public:
FileLogStrategy(const std::string &path = defaultpath, const std::string &name = defaultfilename)
: _logpath(path),
_logfilename(name)
{
LockGuard lockguard(_mutex);
if (std::filesystem::exists(_logpath))
return;
try
{
std::filesystem::create_directories(_logpath);
}
catch (const std::filesystem::filesystem_error &e)
{
std::cerr << e.what() << '\n';
}
}
void SyncLog(const std::string &message) override
{
{
LockGuard lockguard(_mutex);
if (!_logpath.empty() && _logpath.back() != '/')
{
_logpath += "/";
}
std::string targetlog = _logpath + _logfilename; // "./log/log.txt"
std::ofstream out(targetlog, std::ios::app); // 追加方式写入
if (!out.is_open())
{
std::cerr << "open " << targetlog << "failed" << std::endl;
return;
}
out << message << "\n";
out.close();
}
}
~FileLogStrategy()
{
}
private:
std::string _logpath;
std::string _logfilename;
Mutex _mutex;
};
// 交给大家
// const std::string defaultfilename = "log.info";
// const std::string defaultfilename = "log.warning";
// const std::string defaultfilename = "log.fatal";
// const std::string defaultfilename = "log.error";
// const std::string defaultfilename = "log.debug";
// 文件策略&&分日志等级来进行保存
// class FileLogLevelStrategy : public LogStrategy
// {
// public:
// private:
// };
// 日志类:
// 1. 日志的生成
// 2. 根据不同的策略，进行刷新
class Logger
{
// 日志的生成:
// 构建日志字符串
public:
Logger()
{
UseConsoleStrategy();
}
void UseConsoleStrategy()
{
_strategy = std::make_unique<ConsoleStrategy>();
}
void UseFileStrategy()
{
_strategy = std::make_unique<FileLogStrategy>();
}
// 内部类, 标识一条完整的日志信息
// 一条完整的日志信息 = 做半部分固定部分 + 右半部分不固定部分
// LogMessage RAII风格的方式，进行刷新
class LogMessage
{
public:
LogMessage(LogLevel level, std::string &filename, int line, Logger &logger)
: _level(level),
_curr_time(GetCurrentTime()),
_pid(getpid()),
_filename(filename),
_line(line),
_logger(logger)
{
// 先构建出来左半部分
std::stringstream ss;
ss << "[" << _curr_time << "] "
<< "[" << LogLevel2Message(_level) << "] "
<< "[" << _pid << "] "
<< "[" << _filename << "] "
<< "[" << _line << "] "
<< " - ";
_loginfo = ss.str();
}
template <typename T>
LogMessage &operator<<(const T &info)
{
std::stringstream ss;
ss << info;
_loginfo += ss.str();
return *this; // 返回当前LogMessage对象，方便下次继续进行<<
}
~LogMessage()
{
if (_logger._strategy)
{
_logger._strategy->SyncLog(_loginfo);
}
}
private:
LogLevel _level;
std::string _curr_time;
pid_t _pid;
std::string _filename;
int _line;
std::string _loginfo; // 一条完整的日志信息
// 一个引用，引用外部的Logger类对象
Logger &_logger; // 方便我们后续进行策略式刷新
};
// 这里已经不是内部类了
// 故意采用拷贝LogMessage
LogMessage operator()(LogLevel level, std::string filename, int line)
{
return LogMessage(level, filename, line, *this);
}
~Logger()
{
}
private:
std::unique_ptr<LogStrategy> _strategy; // 刷新策略
};
// 日志对象，全局使用
Logger logger;
#define ENABLE_CONSOLE_LOG_STRATEGY() logger.UseConsoleStrategy();
#define ENABLE_FILE_LOG_STRATEGY() logger.UseFileStrategy();
#define LOG(level) logger(level, __FILE__, __LINE__)
}
#endif
//Makefile
.PHONY:all
all:client_tcp server_tcp
server_tcp:EchoTcpServerMain.cc
g++ -o $@ $^ -std=c++17 -lpthread
client_tcp:EchoTcpClient.cc
g++ -o $@ $^ -std=c++17 -static
.PHONY:clean
clean:
rm -f client_tcp server_tcp
//Mutex.hpp
#pragma once
#include <iostream>
#include <pthread.h>
class Mutex
{
public:
Mutex()
{
pthread_mutex_init(&_lock, nullptr);
}
void Lock()
{
pthread_mutex_lock(&_lock);
}
pthread_mutex_t *Ptr()
{
return &_lock;
}
void Unlock()
{
pthread_mutex_unlock(&_lock);
}
~Mutex()
{
pthread_mutex_destroy(&_lock);
}
private:
pthread_mutex_t _lock;
};
class LockGuard // RAII风格代码
{
public:
LockGuard(Mutex &lock):_lockref(lock)
{
_lockref.Lock();
}
~LockGuard()
{
_lockref.Unlock();
}
private:
Mutex &_lockref;
};
//Thread.hpp
#pragma once
#include <iostream>
#include <string>
#include <functional>
#include <pthread.h>
namespace NS_THREAD_MODULE
{
static int gnumber = 1;
using callback_t = std::function<void()>;
enum class TSTATUS
{
THREAD_NEW,
THREAD_RUNNING,
THREAD_STOP
};
std::string Status2String(TSTATUS s)
{
switch (s)
{
case TSTATUS::THREAD_NEW:
return "THREAD_NEW";
case TSTATUS::THREAD_RUNNING:
return "THREAD_RUNNING";
case TSTATUS::THREAD_STOP:
return "THREAD_STOP";
default:
return "UNKNOWN";
}
}
std::string IsJoined(bool joinable)
{
return joinable ? "true" : "false";
}
class Thread
{
private:
void ToRunning()
{
_status = TSTATUS::THREAD_RUNNING;
}
void ToStop()
{
_status = TSTATUS::THREAD_STOP;
}
static void *ThreadRoutine(void *args)
{
Thread *self = static_cast<Thread *>(args);
pthread_setname_np(self->_tid, self->_name.c_str());
self->_cb();
self->ToStop();
return nullptr;
}
public:
Thread(callback_t cb)
: _tid(-1), _status(TSTATUS::THREAD_NEW), _joinable(true), _cb(cb), _result(nullptr)
{
_name = "Slaver-" + std::to_string(gnumber++);
}
bool Start()
{
int n = pthread_create(&_tid, nullptr, ThreadRoutine, this);
if (n != 0)
return false;
ToRunning();
return true;
}
void Join()
{
if (_joinable)
{
int n = pthread_join(_tid, &_result);
if (n != 0)
{
std::cerr << "join error: " << n << std::endl;
return;
}
(void)_result;
_status = TSTATUS::THREAD_STOP;
}
else
{
std::cerr << "error, thread join status: " << IsJoined(_joinable) << std::endl;
}
}
// 暂停
// void Stop() // restart()
// {
// // 让线程暂停
// }
void Die()
{
if (_status == TSTATUS::THREAD_RUNNING)
{
pthread_cancel(_tid);
_status = TSTATUS::THREAD_STOP;
}
}
void Detach()
{
if (_status == TSTATUS::THREAD_RUNNING && _joinable)
{
pthread_detach(_tid);
_joinable = false;
}
else
{
std::cerr << "detach " << _name << " failed" << std::endl;
}
}
void PrintInfo()
{
std::cout << "thread name : " << _name << std::endl;
std::cout << "thread _tid : " << _tid << std::endl;
std::cout << "thread _status : " << Status2String(_status) << std::endl;
std::cout << "thread _joinable : " << IsJoined(_joinable) << std::endl;
}
~Thread()
{
}
private:
std::string _name;
pthread_t _tid;
TSTATUS _status;
bool _joinable;
// 线程要有自己的任务处理，即回调函数
callback_t _cb;
// 线程退出信息
void *_result;
};
}
//ThreadPool.hpp
#pragma once
#include <iostream>
#include <vector>
#include <queue>
#include "Logger.hpp"
#include "Thread.hpp"
#include "Mutex.hpp"
#include "Cond.hpp"
namespace NS_THREAD_POOL_MODULE
{
using namespace NS_LOG_MODULE;
using namespace NS_THREAD_MODULE;
const int defaultnum = 15;
// void Test()
// {
// char name[128];
// pthread_getname_np(pthread_self(), name, sizeof(name));
// while (true)
// {
// LOG(LogLevel::DEBUG) << "我是一个线程，我要进行运行:" << name;
// sleep(1);
// }
// }
// 线程池要不要对多个线程进行管理呢？？
// 先描述，在组织！
template <typename T>
class ThreadPool
{
private:
void HandlerTask()
{
char name[128];
pthread_getname_np(pthread_self(), name, sizeof(name));
while (true)
{
T task;
{
// 保护临界区
LockGuard lockguard(_mutex);
// 检测任务。不休眠：1. 队列不为空 2. 线程池退出 -> 队列为空 && 线程池不退出
while (_tasks.empty() && _isrunning)
{
// 没有任务, 休眠
_slaver_sleep_count++;
_cond.Wait(_mutex);
_slaver_sleep_count--;
}
// 线程池退出了-> while 就要break -> 不能
// 1. 线程池退出 && _tasks empty
if (!_isrunning && _tasks.empty())
{
_mutex.Unlock();
break;
}
// 有任务, 取任务，本质：把任务由公共变成私有
// T -> task*
task = _tasks.front();
_tasks.pop();
}
// 处理任务, 约定
// 处理任务需要再临界区内部处理吗？不需要
LOG(LogLevel::INFO) << name << "处理任务:";
task();
}
// 线程退出
LOG(LogLevel::INFO) << name << " quit...";
}
ThreadPool(int slaver_num = defaultnum) : _isrunning(false), _slaver_sleep_count(0), _slaver_num(slaver_num)
{
// ThreadPool对象已经存在了
for (int idx = 0; idx < _slaver_num; idx++)
{
// auto f = std::bind(&ThreadPool::HandlerTask, this);
// // auto f = [this](){
// // this->HandlerTask();
// // };
// _slavers.emplace_back(f);
_slavers.emplace_back([this]()
{ this->HandlerTask(); });
}
}
// 赋值 拷贝构造禁止
ThreadPool<T> &operator=(const ThreadPool<T> &) = delete;
ThreadPool(const ThreadPool<T> &) = delete;
public:
// 如果多线程获取这个单例呢.加锁
// 多线程安全了，但是效率比较低，双if判断
static ThreadPool<T> *Instance()
{
if(nullptr == _instance) // 双if判断
{
// 多线程
LockGuard lockguard(_lock);
if (nullptr == _instance)
{
// 第一次调用
_instance = new ThreadPool<T>();
_instance->Start();
LOG(LogLevel::INFO) << "第一次使用线程池，创建线程池对象";
}
}
return _instance;
}
void Start()
{
if (_isrunning)
{
LOG(LogLevel::WARNING) << "Thread Pool Is Already Running";
return;
}
_isrunning = true;
for (auto &slave : _slavers)
{
slave.Start();
}
}
void Stop()
{
// version1 -- 后续调整
// if (!_isrunning)
// {
// LOG(LogLevel::WARNING) << "Thread Pool Is Not Running";
// return;
// }
// for (auto &slave : _slavers)
// {
// slave.Die(); // 太简单粗暴了
// }
// _isrunning = false;
// version 2
// 1. _isrunning = false
// 2. 处理完成tasks所有的任务
// 线程状态: 休眠，正在处理任务 -> 让所有线程全部唤醒
// HandlerTask自动break
_mutex.Lock();
_isrunning = false;
if (_slaver_sleep_count > 0)
_cond.Broadcast();
_mutex.Unlock();
}
void Wait()
{
for (auto &slave : _slavers)
{
slave.Join();
}
}
void Enqueue(T in)
{
_mutex.Lock();
_tasks.push(in);
if (_slaver_sleep_count > 0)
_cond.Signal();
_mutex.Unlock();
}
~ThreadPool()
{
}
private:
bool _isrunning;
int _slaver_num;
std::vector<Thread> _slavers;
std::queue<T> _tasks; // 任务队列，临界资源
Mutex _mutex;
Cond _cond;
int _slaver_sleep_count;
// 添加单例模式
static ThreadPool<T> *_instance;
static Mutex _lock; // 保证单例的安全
};
template <typename T>
ThreadPool<T> *ThreadPool<T>::_instance = nullptr;
template <typename T>
Mutex ThreadPool<T>::_lock;
}
1.整体架构：这些模块如何配合
| 层次 | 文件 | 主要职责 |
|---|---|---|
| 构建层 | Makefile | 编译客户端、服务端，清理可执行文件 |
| 基础组件层 | InetAddr.hpp | 封装 IPv4 地址、端口及其转换 |
| 基础组件层 | Mutex.hpp | 封装互斥锁与 RAII 锁管理 |
| 基础组件层 | Cond.hpp | 封装条件变量，实现工作线程等待与唤醒 |
| 基础组件层 | Logger.hpp | 提供日志格式化、控制台输出和文件输出 |
| 并发执行层 | Thread.hpp | 把 POSIX 线程封装为 C++ 对象 |
| 并发调度层 | ThreadPool.hpp | 管理工作线程、任务队列及任务分发 |
| 网络业务层 | EchoTcpServer.hpp | 监听连接、接收连接、提交回显任务 |
| 程序入口层 | EchoTcpServerMain.cc | 解析参数，创建并启动服务器 |
| 客户端 | EchoTcpClient.cc | 连接服务器，发送输入，接收回显 |
当前服务端的主要调用关系是：
EchoTcpServerMain.cc
main()
│
├── TcpServer::InitServer()
│ socket → bind → listen
│
└── TcpServer::Start()
│
└── accept() 得到一个客户端连接
│
├── InetAddr 封装客户端地址
│
└── ThreadPool::Enqueue() 提交任务
│
└── 某个工作线程取出任务
│
└── TcpServer::serviceIO()
read → write → close
可以把核心分工概括为：主线程接收连接，线程池分配任务，工作线程完成连接上的数据收发。
注意,放入任务队列的不是客户端发来的消息,而是一个**“处理这个客户端连接”的可调用对象**.
2.Makefile：管理程序构建
核心规则是：
.PHONY: all
all: client_tcp server_tcp
server_tcp: EchoTcpServerMain.cc
g++ -o $@ $^ -std=c++17 -lpthread
client_tcp: EchoTcpClient.cc
g++ -o $@ $^ -std=c++17 -static
.PHONY: clean
clean:
rm -f client_tcp server_tcp
1.all：同时构建客户端和服务端
all: client_tcp server_tcp
表示 all 依赖两个目标.执行 make 时,Make 会检查是否需要生成或更新这两个可执行文件.
.PHONY: all 表示 all 是一个逻辑目标,而不是要生成的同名文件.clean 同理.需要注意：把 all 声明为伪目标,并不会让它依赖的可执行文件每次都无条件重新编译. 是否重建仍取决于对应规则的依赖关系和时间戳.
2.$@ 和 $^：自动变量
以服务端规则为例：
server_tcp: EchoTcpServerMain.cc
g++ -o $@ $^ -std=c++17 -lpthread
| 写法 | 含义 | 这里的实际值 |
|---|---|---|
$@ | 当前规则的目标名称 | server_tcp |
$^ | 全部普通依赖文件，去除重复项 | EchoTcpServerMain.cc |
$< | 第一个依赖文件 | 此处也是 EchoTcpServerMain.cc |
因此,原来的命令相当于：
g++ -o server_tcp EchoTcpServerMain.cc -std=c++17 -lpthread
这些是GNU Make的自动变量,不是 Shell 变量.
3.编译参数的作用
-std=c++17 指定使用 C++17;-lpthread 链接 POSIX 线程库;-static 要求客户端采用静态链接.
这里建议将线程相关选项改成 -pthread.GCC 文档要求它在编译和链接阶段保持一致,因为它在部分平台上还会影响预处理配置,不只是链接一个库.客户端的 -static 并非 TCP 编程所必需,没有明确的静态部署需求时可以去掉.
4.当前 Makefile 的一个实际问题：没有跟踪头文件依赖
目前：
server_tcp: EchoTcpServerMain.cc
只声明了 .cc 文件。
因此,只修改 ThreadPool.hpp、EchoTcpServer.hpp 等头文件时,这条规则没有告诉 Make 应当重新构建服务端.
可以先采用较简单的写法：
server_tcp: EchoTcpServerMain.cc $(wildcard *.hpp)
g++ -std=c++17 -Wall -Wextra -pthread $< -o $@
这里改用 $<，只把 .cc 作为编译输入;头文件用于判断是否需要重建.项目继续扩大后,再引入 .o 文件和自动生成的 .d 依赖文件更合适.Make的重建判断依赖于规则中记录的依赖关系.
3.InetAddr.hpp：封装 IP 地址和端口
这个类负责把同一个地址在两种表示之间转换：
便于程序使用和打印的表示：
std::string IP + 主机字节序端口
便于 Socket API 使用的表示：
struct sockaddr_in
1.成员变量的作用
struct sockaddr_in _address;
socklen_t _len;
std::string _ip;
uint16_t _port;
其中：
| 成员 | 作用 |
|---|---|
_address | 保存供 Socket API 使用的 IPv4 地址结构 |
_len | 保存地址结构的字节大小 |
_ip | 保存可读的 IP 字符串 |
_port | 保存主机字节序的端口号 |
这里的 _len 是 sizeof(sockaddr_in) 对应的长度,不是 IP 字符串的长度.
2.从 sockaddr_in 构造：用于描述已经接入的客户端
InetAddr(const struct sockaddr_in &address)
: _address(address), _len(sizeof(address))
{
_ip = inet_ntoa(_address.sin_addr);
_port = ntohs(_address.sin_port);
}
accept() 获得客户端地址后,服务器可以用它创建：
InetAddr clientaddress(clientaddr);
这个构造函数把系统返回的地址结构,转换成方便打印和记录日志的形式.
inet_ntoa() 把 IPv4 地址转换为字符串;ntohs() 把端口从网络字节序转换为主机字节序.
3.从 IP 和端口构造：用于连接或绑定
InetAddr(uint16_t port,
const std::string &ip = "0.0.0.0")
: _ip(ip), _port(port)
{
bzero(&_address, sizeof(_address));
_address.sin_family = AF_INET;
_address.sin_port = htons(_port);
_address.sin_addr.s_addr = inet_addr(_ip.c_str());
_len = sizeof(_address);
}
这里做了四件事：清零地址结构、指定 IPv4、转换端口字节序、转换 IP 地址.
特别注意：inet_addr() 的结果已经是网络字节序，不应再额外调用一次 htonl()。
默认值 0.0.0.0 适合表示服务器绑定时的任意本地 IPv4 地址;客户端连接服务器时,应明确传入目标服务器地址.
4.其他接口
ToString() 把地址拼接成类似：
[127.0.0.1:8888]
GetNetAddress() 返回内部 sockaddr_in 的地址,供 connect() 等函数使用.
Len() 返回地址结构大小.
operator==() 比较保存的 IP 字符串和端口是否相同.
5.这个封装值得改进的地方
首先默认构造函数：
InetAddr() {}
没有给 _address、_len 和 _port 初始化。默认构造后直接调用相应接口,可能使用到未初始化的数据.
其次,建议使用 inet_pton() 解析地址并检查返回值,使用 inet_ntop() 输出地址.这样可以明确检查非法输入,并让输出写入调用者自己的缓冲区.
最后,ToString()、Len() 和比较运算符不修改对象,适合加上 const.这也便于通过常量引用使用地址对象.
4.Mutex.hpp：互斥锁和 RAII 锁管理
这个文件包含两个职责不同的类：Mutex 管理锁本身；LockGuard 管理一次加锁操作的生命周期。
Mutex：封装 pthread_mutex_t
pthread_mutex_t _lock;
它的接口与底层 API 基本一一对应：
| 接口 | 底层操作 | 作用 |
|---|---|---|
| 构造函数 | pthread_mutex_init() | 初始化互斥锁 |
Lock() | pthread_mutex_lock() | 获取锁 |
Unlock() | pthread_mutex_unlock() | 释放锁 |
Ptr() | 返回 &_lock | 给条件变量等待函数使用 |
| 析构函数 | pthread_mutex_destroy() | 销毁互斥锁 |
互斥锁用于让使用同一把锁的线程互斥访问共享数据.销毁锁前,必须保证没有线程仍在使用它;不能把正在使用的互斥锁直接当作普通内存复制或销毁.
LockGuard：进入作用域加锁,退出作用域解锁
核心逻辑是：
LockGuard(Mutex &lock) : _lockref(lock)
{
_lockref.Lock();
}
~LockGuard()
{
_lockref.Unlock();
}
例如：
{
LockGuard guard(_mutex);
// 在这里访问共享任务队列。
task = _tasks.front();
_tasks.pop();
}
// guard 析构，自动解锁。
它把解锁动作交给对象析构,避免每条正常退出路径都手动写一次 Unlock().
但也正因为如此，在 LockGuard 仍然存活时，不应再随意手动解锁。 线程池退出分支恰好存在这个问题，后面会具体指出.
这个模块在项目里的位置
线程池用它保护任务队列及相关状态;日志模块用它避免多个工作线程通过同一个日志输出器同时写入.
5.Cond.hpp：让空闲工作线程等待任务
互斥锁解决的是：多个线程不能同时修改任务队列.
条件变量解决的是：队列没有任务时，工作线程如何等待，而不是一直循环检查。
1.三个核心接口
void Wait(Mutex &mutex);
void Signal();
void Broadcast();
| 接口 | 底层 API | 含义 |
|---|---|---|
Wait() | pthread_cond_wait() | 等待共享状态发生变化 |
Signal() | pthread_cond_signal() | 有等待者时，唤醒至少一个等待线程 |
Broadcast() | pthread_cond_broadcast() | 唤醒所有正在等待的线程 |
严格来说,Signal() 不应理解为“绝对只唤醒一个线程”;程序仍然必须在醒来后检查条件.
2.Wait() 为什么必须传入互斥锁
pthread_cond_wait(&_cond, mutex.Ptr());
调用它之前,当前线程必须已经持有这把锁.
它会原子地释放互斥锁并进入等待;被唤醒后,在成功返回前重新获取该锁.所以,线程等待期间,生产者仍然可以获取同一把锁并向队列插入任务.
3.为什么线程池使用 while 检查队列
条件是：
while (_tasks.empty() && _isrunning)
{
_cond.Wait(_mutex);
}
不能简单改成：
if (_tasks.empty())
{
_cond.Wait(_mutex);
}
因为一次唤醒并不保证队列现在一定非空：可能发生虚假唤醒,也可能任务已经被其他线程取走.醒来后必须重新检查共享状态.
另外,条件变量不是“保存通知次数”的消息队列.没有等待者时发出的通知,不会替以后到来的线程保留;真正需要保存的是 _tasks 和 _isrunning 这样的状态.
图中的：
int n = pthread_cond_wait(...);
(void)n;
只是忽略返回值,并不代表做了错误处理.POSIX 线程接口通常直接返回错误码,不能机械套用 Socket API 的“返回 -1 后检查 errno”方式.
6.Logger.hpp：日志格式化和输出策略
这个文件虽然比较长,但可以拆成三部分：日志信息的组织、日志输出方式的选择、流式调用接口的封装。
1.LogLevel：描述日志级别
图中包括：
INFO
WARNING
ERROR
FATAL
DEBUG
LogLevel2Message() 把枚举转换为相应字符串，用于组成日志前缀.
这里要注意：从实现看，级别主要用于标识日志类型。写一条 FATAL 日志本身不会自动退出程序，服务器里的退出行为来自后面的 exit(...)。
2.GetCurrentTime()：生成时间字符串
函数通过 gettimeofday() 取得秒和微秒,再使用 localtime_r() 得到日历时间,最后格式化为日志中的时间部分.
其目的不是参与TCP通信,而是方便分析：
何时接入连接
何时开始处理任务
何时收到数据
何时发生错误
3.LogStrategy：定义统一输出接口
class LogStrategy
{
public:
virtual ~LogStrategy() = default;
virtual void SyncLog(const std::string &message) = 0;
};
这是一种策略模式设计.
业务代码只需要表达“输出这条日志”,不需要知道日志最终写到控制台还是文件.
4.ConsoleStrategy：输出到控制台
它的 SyncLog() 先获取自己的互斥锁,再通过 std::cerr 输出完整日志.
因此,经由同一个输出策略实例写日志的多个线程,可以在输出这一段互斥,减少日志互相穿插的问题.
5.FileLogStrategy：追加到日志文件
构造时检查或创建日志目录;输出时拼接目录和文件名,以追加方式打开文件,再写入日志.
6.Logger：持有当前日志策略
std::unique_ptr<LogStrategy> _strategy;
UseConsoleStrategy() 和 UseFileStrategy() 切换具体输出方式.
服务端入口中的：
ENABLE_CONSOLE_LOG_STRATEGY();
就是选择控制台输出.
当前是在工作线程启动之前切换策略.以后若增加运行过程中切换日志目标的功能,还需要考虑 _strategy 的并发访问和生命周期.
7.LogMessage：实现 LOG(...) << ... 的写法
例如：
LOG(LogLevel::INFO) << "accept success, sockfd: " << sockfd;
其工作过程是：
LOG 宏补充源码文件名和行号
→ Logger::operator() 创建临时 LogMessage
→ 多次 operator<< 追加内容
→ 表达式结束，临时对象析构
→ 析构函数调用策略输出完整日志
LogMessage 的构造函数组织时间、级别、进程号、文件名和行号;模板化的 operator<< 把不同类型转换成文本并追加.
这里的进程号是 PID,不是工作线程 ID.要区分哪个工作线程执行了任务,可以结合线程池已经打印的线程名称.
8.头文件中的全局定义需要留意
直接在头文件中定义了：
Logger logger;
同时还有一些普通的非 inline 函数定义.
当前每个可执行文件只编译一个 .cc,问题不一定暴露;以后多个 .cc 包含同一头文件并链接到一个程序时,会有重复定义问题.可以采用“头文件 extern 声明、一个 .cc 中定义”,或使用符合条件的 C++17 inline 定义.头文件保护不能替代整个程序的单一定义要求.
7.Thread.hpp：把 POSIX 线程封装成 C++ 对象
这个模块把线程编号、状态和执行函数组织在一起.
1.成员变量分别描述什么
std::string _name;
pthread_t _tid;
TSTATUS _status;
bool _joinable;
callback_t _cb;
void *_result;
_name 是便于日志识别的线程名称;_tid 是 POSIX 线程标识;_status 是程序自己维护的状态;_joinable 表示是否可以回收;_cb 保存线程执行函数;_result 保存线程退出结果.
其中：
using callback_t = std::function<void()>;
表示线程执行的任务接口是一个无参数、无返回值的可调用对象.
2.构造函数并没有真正启动线程
构造函数只是保存回调、设置初始状态,并生成类似 Slaver-1 的名称.
真正创建系统线程的是：
pthread_create(&_tid, nullptr, ThreadRoutine, this);
pthread_create() 成功后,新线程从指定入口函数开始执行.
3.为什么线程入口 ThreadRoutine() 是静态成员函数
POSIX 线程入口需要接受一个 void* 参数并返回 void*.普通非静态成员函数还需要隐含的 this 对象,不适合直接作为这里的入口.
代码通过静态入口桥接：
static void *ThreadRoutine(void *args)
{
Thread *self = static_cast<Thread *>(args);
self->_cb();
self->ToStop();
return nullptr;
}
pthread_create() 的第四个参数传入 this,线程入口再把它转换回来.
在线程池中,_cb() 执行的是 HandlerTask() 的工作循环,而不是只处理一个客户端. 某个连接处理完毕后,工作线程会继续取下一个任务.
4.主要成员函数
| 方法 | 作用 |
|---|---|
Start() | 调用 pthread_create() 启动线程 |
Join() | 等待线程结束并回收其线程资源 |
Detach() | 让线程结束时自行释放相应线程资源，不再由调用方 Join() |
Die() | 通过 pthread_cancel() 发出取消请求 |
PrintInfo() | 打印线程名称、标识和程序维护的状态 |
Join() 不是启动线程,也不是停止线程;它是等待和回收.Detach() 也不会终止线程.
尤其要注意,pthread_cancel() 是提出取消请求,不等于调用返回时目标线程已经退出.因此，Die() 里马上把状态改为 THREAD_STOP,不能作为线程已经结束的证明.
5.当前封装的几个问题
Join() 成功后没有把 _joinable 改为 false. 后续再次 Join() 时,封装层无法阻止错误调用.线程已经结束和线程已经被回收,是两个不同的状态.
_status 存在并发访问风险。 主线程在 Start() 中写入 RUNNING,工作线程结束时写入 STOP.如果工作线程很快结束,还可能先写入 STOP,随后被主线程写回 RUNNING.需要同时处理访问同步和状态迁移顺序,而不是只给变量换一个名字.未同步的冲突访问可能构成 C++ 数据竞争.
析构函数为空。 因此,销毁 Thread 对象并不会自动等待其线程结束.由于入口函数仍通过 this 访问对象,必须确保对象在线程使用期间一直有效.
8.ThreadPool.hpp：连接任务的调度中心
这是实现多个客户端并发处理的核心模块.
1.线程池的基本组成
std::vector<Thread> _slavers;
std::queue<T> _tasks;
Mutex _mutex;
Cond _cond;
bool _isrunning;
int _slaver_sleep_count;
其中,_slavers 保存工作线程对象,_tasks 保存等待执行的任务,_mutex 保护共享状态,_cond 用于等待和唤醒.
默认工作线程数是：
const int defaultnum = 15;
这表示默认准备 15 个工作线程，不表示整个服务器最多只能建立 15 条 TCP 连接。
2.为什么写成模板
template <typename T>
class ThreadPool;
线程池不把任务类型固定为 TCP 连接,而是保存一个可以执行的 T.
服务器使用：
using task_t = std::function<void()>;
所以,这里的线程池实际上保存了一批“无参数、无返回值的操作”.
3.构造函数：创建工作线程对象
构造函数循环执行：
_slavers.emplace_back(
[this]()
{
this->HandlerTask();
}
);
每个 Thread 对象持有一个回调：运行当前线程池的 HandlerTask().
此时只是构造 C++ 对象.后续 Start() 遍历 _slavers,调用每个线程对象的 Start(),才真正启动系统线程.
4.Instance()：首次使用时创建并启动线程池
图中的首次初始化分支不仅执行：
_instance = new ThreadPool<T>();
还会执行：
_instance->Start();
所以,服务器第一次接入客户端并调用 Instance() 时,线程池会被创建并启动.后续连接复用同一个实例.
这也解释了为什么服务器入口中没有单独写一行“启动线程池”.
5.Enqueue()：生产任务
它的逻辑是：
获取队列锁
→ 将任务压入队列
→ 存在等待线程时发送通知
→ 释放队列锁
服务器主线程通过它提交连接处理任务,因而充当生产者.
当前接口没有检查停止状态,也没有限制队列长度.以后完善时,需要明确：
线程池停止后是否拒绝新任务，队列已满时如何处理新连接。
另外,当前使用手动加锁和解锁;若 _tasks.push() 抛出异常,解锁语句可能执行不到.这里也适合使用 RAII.
6.HandlerTask()：消费任务
工作线程反复执行：
加锁
→ 队列为空且线程池运行中：等待
→ 停止且队列为空：退出
→ 否则取出一个任务
解锁
→ 执行 task()
→ 继续下一轮
其退出判断对应下面四种情况：
| 运行状态 | 队列状态 | 工作线程行为 |
|---|---|---|
| 运行中 | 空 | 等待任务 |
| 运行中 | 非空 | 取出并执行任务 |
| 已停止 | 非空 | 继续处理剩余任务 |
| 已停止 | 空 | 退出工作循环 |
由此可见,当前 Stop() 的设计意图是停止等待新任务，处理完队列中剩余任务后退出，不是立即丢弃全部任务。
还有一个很重要的正确设计：task() 在任务队列锁之外执行。
否则,一个工作线程在 read() 中等待客户端数据时还占着队列锁,其他线程就无法取任务,主线程也无法正常提交任务.
7.必须修复：退出分支重复解锁
图的工作循环中已经创建了：
LockGuard lockguard(_mutex);
但退出分支又手动执行：
_mutex.Unlock();
break;
break 离开作用域时,LockGuard 还会析构并再次解锁.
正确做法是删除手动解锁：
if (!_isrunning && _tasks.empty())
{
break; // LockGuard 析构时负责解锁。
}
对不再由当前线程持有的普通互斥锁再次解锁,可能产生未定义行为,不能视为无害的重复操作.
8.Stop() 和 Wait() 的边界
Stop() 修改 _isrunning，并唤醒条件变量上的等待线程.
Wait() 对各工作线程执行 Join(),等待它们结束.
但要区分两种等待：
空闲工作线程：
等待条件变量
正在处理连接的工作线程：
可能阻塞在 socket 的 read()
条件变量的 Broadcast() 只能通知等待该条件变量的线程,不能直接让阻塞在 Socket read() 上的线程返回. 因此,客户端一直不发数据时,单纯调用 Stop() 和 Wait() 仍可能一直等下去.需要另外设计连接超时或中断连接 I/O 的停止流程.
9.单例的“双重检查”并不完整安全
当前使用普通指针 _instance,先在锁外判断,再在锁内判断并初始化.
在多个线程可能同时调用 Instance() 时,锁外读取与另一个线程写入指针之间缺少同步,不能仅凭“双重检查”就宣称线程安全.当前服务器主要由主线程调用它,可能暂时不触发,但这个通用组件的实现仍有问题.
建议改用局部静态对象的线程安全初始化,或用一次性初始化机制完整保护“创建并启动”的过程.C++ 对局部静态对象的并发初始化提供了等待初始化完成的保证,但线程池的启动失败处理、停止和析构仍需另外设计.
9.EchoTcpServer.hpp：监听连接并执行回显
这个文件将前面的基础模块和线程池组合成服务器.
1.常量、错误码和成员变量
图中的错误码包括参数错误、创建 Socket 失败、绑定失败、监听失败等,用于不同失败路径的退出状态.
主要成员为：
uint16_t _port;
int _listensockfd;
_port 是服务器端口；_listensockfd 是监听套接字.
监听套接字不是某个客户端的通信连接.每次 accept() 成功后返回的 sockfd，才用于和对应客户端收发数据.
2.InitServer()：完成网络初始化
它依次执行：
socket(AF_INET, SOCK_STREAM, 0)
→ 填充 sockaddr_in
→ bind()
→ listen()
这里使用 IPv4 TCP;端口经过 htons() 转换;地址设置为 INADDR_ANY,表示绑定所有本地 IPv4 接口地址.
图中的：
static const int gbacklog = 16;
用于 listen() 的等待队列参数。
它和线程池的 15 个工作线程完全不是一回事。 在 Linux 中,backlog 主要约束已建立但尚未被应用 accept() 取走的连接队列,实际还受系统配置限制,不是服务器总连接数上限.
工程上还建议将 _listensockfd 初始值设为 -1，析构时检查其有效性；需要便于重启时，可在 bind() 前按需设置 SO_REUSEADDR。该选项允许的是符合条件的地址重用，并非任意抢占其他服务占用的监听地址.
3.Start()：循环接收客户端
核心调用为：
int sockfd = accept(
_listensockfd,
reinterpret_cast<struct sockaddr *>(&clientaddr),
&len
);
成功后,sockfd 对应一条新连接;clientaddr 保存对端地址;原监听 Socket 继续用于接收后续连接.阻塞模式下,没有可接受连接时,accept() 会等待.
当前代码的优点是：它没有直接在主线程里调用阻塞的 serviceIO(),而是把工作提交到线程池.
4.Lambda：把“一个连接的处理过程”变成任务
图中的关键代码是：
InetAddr clientaddress(clientaddr);
ThreadPool<task_t>::Instance()->Enqueue(
[this, sockfd, clientaddress]() -> void
{
this->serviceIO(sockfd, clientaddress);
}
);
三个捕获值的作用分别是：
| 捕获项 | 作用 |
|---|---|
this | 让任务能够调用当前服务器的成员函数 |
sockfd | 保存这次接入连接的描述符数值 |
clientaddress | 保存这次连接的客户端地址副本 |
值捕获地址对象可以避免任务稍后执行时引用到已经离开作用域的局部变量.
不过,复制 sockfd 的整数值并不等于复制一个独立 Socket,也不等于调用 dup();复制 this 指针也不会延长服务器对象寿命.必须让连接资源和服务器对象在任务执行期间保持有效.
在线程池模型下，主线程提交任务后不能立即 close(sockfd),因为同一进程的线程共享文件描述符表,工作线程还需要使用该描述符.
5.serviceIO()：实际执行收发
当前启用的代码可以概括为：
read() 一次
│
├── n > 0：记录内容，添加 "Server echo" 前缀，write() 一次
├── n == 0：记录对端结束发送
└── n < 0：记录读取错误
│
close(sockfd)
服务端读取时使用：
read(sockfd, inbuffer, sizeof(inbuffer) - 1);
因此,在 n > 0 时补：
inbuffer[n] = 0;
为字符串结束符预留了位置.
但是,一次 read() 得到的是当前读取到的字节片段,不保证就是一条完整业务消息. TCP 不保留应用层消息边界,读取也可能少于请求长度.
当前还存在两个限制：write() 的返回值没有检查;通过 C 字符串拼接回显内容,遇到数据中的 '\0' 会提前截断.因此,这是一份文本回显演示,不是完整的二进制流回显实现.一次写调用也可能只完成部分数据,需要根据返回值继续发送.
6.注释中的长连接版本
图保留了一个带 while (true) 的收发循环,其意图是让同一连接能够进行多轮通信.
若恢复它,应当替换当前单次读写实现，并把连接关闭放到收发循环退出之后,不能保留前面的 close(sockfd),再把注释循环接在后面.
而且,增加循环只解决连接复用问题,并没有自动解决消息边界问题.
7.四种连接处理版本
| 截图标注 | 处理方式 | 特点 | 当前状态 |
|---|---|---|---|
version 0 | 主线程直接处理连接 | 业务处理与接入串行执行 | 注释 |
version 1 | 多进程处理 | 接入后交给子进程或后代进程 | 注释 |
version 2 | 每连接创建一个线程 | 一个连接对应一个新线程 | 注释 |
version 3 | 线程池处理 | 复用固定数量的工作线程 | 启用 |
因此,不应把它们理解成“当前程序同时使用四种并发模型”.
8.ThreadData：旧多线程版本的参数包装
它把：
TcpServer *_this;
int _sockfd;
InetAddr _addr;
打包后通过 void* 传给旧版线程入口.入口恢复类型、分离线程、执行业务，再删除参数对象.
这里有一个隐藏问题：当前 serviceIO() 已经关闭连接,ThreadData 的析构函数又调用了 close(_sockfd). 恢复旧线程版本前,必须统一关闭责任,否则会重复关闭同一个描述符.由于描述符数值可能被系统复用,重复关闭甚至可能误伤后来分配的其他连接.
10.EchoTcpServerMain.cc：程序启动入口
这个文件负责组织启动步骤,不负责具体收发.
auto tsvr = std::make_unique<TcpServer>(server_port);
tsvr->InitServer();
tsvr->Start();
1.Usage() 和参数检查
服务端的启动形式为：
./server_tcp 8888
因此要求：
argc == 2
参数数量不对时,输出用法并退出.
2.创建和管理服务器对象
std::make_unique<TcpServer>() 创建服务器对象,unique_ptr 在正常析构时负责删除它.
但它只直接管理这个 TcpServer 对象，不会自动替你停止独立单例线程池，也不会自动回收所有排队和运行中的连接任务。
当前 TcpServer::Start() 是无限接入循环,也没有接入完整的正常停止流程.
3.两个 Start() 不要混淆
TcpServer::Start()
开始循环 accept() 并提交连接任务
ThreadPool::Start()
创建并启动工作线程
它们同名,但职责不同.
4.端口解析还需要校验
图直接使用：
uint16_t server_port = std::stoi(argv[1]);
std::stoi() 可能抛出转换异常,也可能只解析字符串的合法前缀.因此,应检查转换是否成功、是否消费完整输入,再检查端口范围,最后转换为 uint16_t.对本例这种手动指定固定服务端口的方式,建议要求 1～65535.
11.EchoTcpClient.cc：交互式测试客户端
客户端的整体流程是：
检查命令行参数
→ 创建 TCP Socket
→ 构造服务器地址
→ connect()
→ 读取终端输入
→ write()
→ read()
→ 打印响应
→ 尝试下一轮
1.为什么客户端没有显式 bind()
客户端通过：
InetAddr serveraddress(server_port, server_ip);
connect(sockfd, ...);
连接服务器.
对于没有显式绑定的 Socket，系统可以在连接时自动分配本地端口,并完成相应的本地地址选择.因此,“客户端通常不显式 bind()”不代表客户端没有本地地址和端口.
2.终端输入到发送数据
std::getline(std::cin, line);
write(sockfd, line.c_str(), line.size());
getline() 读取一行,但默认不会把终止输入的换行符保存到 line 中.因此,用户在终端按回车,并不意味着客户端向网络发送了一个 '\n' 分隔符.
这也说明当前程序没有真正定义“按行传输”的网络协议.
3.必须修复：接收缓冲区越界
图中：
char inbuffer[1024];
ssize_t n = read(sockfd, inbuffer, sizeof(inbuffer));
if (n > 0)
{
inbuffer[n] = 0;
}
如果 n == 1024,就会写入：
inbuffer[1024]
而数组最大合法下标是 1023.
保留字符串输出方式时,应少读一个字节：
ssize_t n = read(sockfd, inbuffer, sizeof(inbuffer) - 1);
if (n > 0)
{
inbuffer[n] = '\0';
std::cout << inbuffer << std::endl;
}
也可以按实际长度输出,避免追加结束符：
if (n > 0)
{
std::cout.write(inbuffer, n);
std::cout << '\n';
}
这些修改只解决本地缓冲区处理问题,不保证一次读取就得到了完整响应.
4.空输入和输入结束可能导致等待
用户只按回车时,line 为空,当前代码没有可发送的有效载荷,却继续等待服务器响应;服务端也可能正在等待客户端发数据.
输入流结束时,代码同样没有检查 getline() 是否成功.
可以先明确采用“空行不发送”的策略：
if (!std::getline(std::cin, line))
{
break;
}
if (line.empty())
{
continue;
}
如果业务确实允许空消息,就需要通过消息长度字段或分隔符来表示,而不是依靠一次零长度写入.
5.服务端和客户端的连接使用方式不一致
客户端循环意味着它希望：
同一连接：
请求 1 → 响应 1 → 请求 2 → 响应 2 → …
但当前服务端是：
同一连接：
读取一次 → 尝试回写一次 → 关闭
所以,第二轮通信不能按这个客户端的预期稳定完成.具体可能表现为读取到 EOF、连接错误,或者写入失效连接时触发错误.
此外,双方都需要检查写入返回值。Linux 上可通过 send(..., MSG_NOSIGNAL) 避免该次发送因 SIGPIPE 导致默认的进程终止，但仍必须处理发送失败.
运行测试结果：
1.3 V2—基于fork的多进程TCP Echo Server：连接分发与子进程回收
这一版的核心变化：不再把连接交给线程池，而是由父进程接收连接，再为每个连接创建一个子进程完成收发。
采用 单次 fork() + SIGCHLD + waitpid(..., WNOHANG)：父进程持续接入客户端,子进程独立处理连接,已经退出的子进程由父进程回收.
1.父进程接连接,子进程处理连接
父进程：TcpServer
│
├── socket → bind → listen
│
└── 循环 accept()
│
├── 客户端 A 接入 → fork() → 子进程 A
│ └── serviceIO(A)
│
├── 客户端 B 接入 → fork() → 子进程 B
│ └── serviceIO(B)
│
└── 客户端 C 接入 → fork() → 子进程 C
└── serviceIO(C)
父进程继续接受连接
子进程完成业务后退出
父进程接收 SIGCHLD，回收已经退出的子进程
这里是 “一个连接创建一个子进程”,不是预先创建固定数量工作进程的“进程池”.
例如,客户端 A 连接后一直不发送数据,等待的是子进程 A;父进程仍可以继续接收客户端B,再创建子进程 B 来处理它.
2.fork() 为什么能够分出两条执行路径
pid_t pid = fork();
成功时,父进程得到子进程的 PID,子进程得到 0;失败时,父进程得到 -1,并且没有创建子进程.子进程从 fork() 返回处继续执行,不是重新从 main() 开始运行.
因此可以通过返回值分工：
if (pid < 0) {
// 创建失败。
} else if (pid == 0) {
// 子进程：处理客户端连接。
} else {
// 父进程：继续接受新连接。
}
主要修改 EchoTcpServer.hpp 中的 Start(),不需要重写整套 Socket 初始化.
| 模块 | V2 中的处理方式 |
|---|---|
EchoTcpServerMain.cc | 保留解析参数、创建对象、初始化和启动流程 |
InetAddr.hpp | 继续用于封装客户端地址 |
InitServer() | 保留 socket → bind → listen |
Start() | 把线程池入队改成 fork() 分发 |
serviceIO() | 在子进程中执行 |
ThreadPool.hpp、Thread.hpp、Cond.hpp | 不再参与这一版的连接调度 |
Logger.hpp、Mutex.hpp | 可以保留，但不能把原来的线程互斥当成跨进程互斥 |
切换到这一版后,不要先启动线程池,再调用这里的 fork() 处理业务. 多线程程序调用 fork() 后,子进程只保留调用线程,其他线程持有的锁状态却可能被复制;在执行新程序之前,子进程可安全调用的函数受到严格限制.这里的实现以父进程始终保持单线程为前提.
另外,你的 Mutex 使用默认的进程私有互斥锁,不能直接协调父子进程之间的日志写入.保留日志模块时,多进程输出仍可能交错.
核心代码：替换 Start(),增加子进程回收
下面这部分用于接入现有的 TcpServer 类,沿用原来的：
_listensockfd
serviceIO(sockfd, clientaddress)
InetAddr
1.补充头文件
在 EchoTcpServer.hpp 中补齐：
#include <cerrno>
#include <cstdio>
#include <cstdlib>
#include <signal.h>
#include <sys/wait.h>
#include <unistd.h>
2.在类中增加回收函数
private:
static void ReapChildren(int)
{
// 信号处理不能破坏被中断代码原来的 errno。
const int saved_errno = errno;
while (true)
{
pid_t pid = waitpid(-1, nullptr, WNOHANG);
if (pid > 0)
{
// 已回收一个子进程，继续检查其他子进程。
continue;
}
if (pid < 0 && errno == EINTR)
{
continue;
}
// pid == 0：暂时没有可回收的子进程。
// pid == -1：例如已经没有子进程，errno == ECHILD。
break;
}
errno = saved_errno;
}
这段代码中的：
waitpid(-1, nullptr, WNOHANG);
表示检查任意直接子进程,不保存退出状态,而且没有可回收的子进程时立即返回,不阻塞等待.
这里必须循环回收,而不是收到一次信号只调用一次 waitpid().SIGCHLD 属于标准信号,多次发生的通知可能合并,不能假设“一次信号严格对应一个子进程退出”.
不要在这个函数里调用你的 LOG(...)、printf()、C++ 流或加锁. 信号处理函数需要遵守异步信号安全规则;waitpid() 可以使用,errno 则需要保存并恢复.
3.替换 Start()
public:
void Start()
{
// 必须在第一次 fork 之前安装回收逻辑。
struct sigaction action {};
action.sa_handler = ReapChildren;
sigemptyset(&action.sa_mask);
action.sa_flags = SA_RESTART | SA_NOCLDSTOP;
if (sigaction(SIGCHLD, &action, nullptr) == -1)
{
std::perror("sigaction");
std::exit(EXIT_FAILURE);
}
while (true)
{
struct sockaddr_in clientaddr {};
socklen_t len = sizeof(clientaddr);
int sockfd = accept(
_listensockfd,
reinterpret_cast<struct sockaddr*>(&clientaddr),
&len
);
if (sockfd == -1)
{
if (errno == EINTR || errno == ECONNABORTED)
{
continue;
}
std::perror("accept");
// 教学版简单退避，避免错误时持续空转。
sleep(1);
continue;
}
pid_t pid = fork();
if (pid == -1)
{
const int saved_errno = errno;
// 创建失败，没有子进程接管这条连接。
close(sockfd);
errno = saved_errno;
std::perror("fork");
sleep(1);
continue;
}
if (pid == 0)
{
// 子进程不接受新连接，关闭自己的监听描述符。
close(_listensockfd);
_listensockfd = -1;
InetAddr clientaddress(clientaddr);
// 按你的现有实现，serviceIO 内部负责 close(sockfd)。
serviceIO(sockfd, clientaddress);
// 不要再重复 close(sockfd)，也不要回到 accept 循环。
_exit(EXIT_SUCCESS);
}
// 父进程不处理业务，只关闭自己持有的通信描述符。
close(sockfd);
// 不在这里阻塞等待业务子进程。
// 继续下一轮 accept，退出的子进程由 SIGCHLD 处理器回收。
}
}
其中,SA_RESTART 让部分被信号中断的阻塞调用能够自动重新开始;SA_NOCLDSTOP 则避免因子进程停止或恢复运行而收到相应的 SIGCHLD 通知.代码仍保留了 EINTR 检查.
原来的线程池 Enqueue() 分支、每连接创建线程的分支和双重 fork() 分支都不要同时保留为可执行代码.一次连接只能由选定的一种模型接管.
1.为什么父子进程都要 close(),却不会互相断开连接？
fork() 后,子进程继承父进程的文件描述符副本;对应描述符引用的是同一个内核打开对象,并不是重新建立了一条 TCP 连接.
假设接入连接时：
监听描述符：listen_fd = 3
通信描述符：conn_fd = 4
执行 fork() 后：
父进程 子进程
fd 3 ────────┐ ┌────── fd 3
└─ 监听 Socket ─┘
fd 4 ────────┐ ┌────── fd 4
└─ 通信 Socket ─┘
分工后应该是：
| 进程 | 保留 | 立即关闭 |
|---|---|---|
| 父进程 | 监听描述符 | 本次接入的通信描述符 |
| 子进程 | 本次接入的通信描述符 | 监听描述符 |
| 子进程处理完毕 | 无 | 自己的通信描述符 |
父进程关闭自己的通信描述符,只是放弃本进程持有的引用,子进程仍然可以使用它.如果父进程一直不关闭,则会泄漏描述符,也可能让客户端迟迟观察不到预期的连接结束.
这里不能把父进程中的 close(sockfd) 换成 shutdown(sockfd, SHUT_RDWR). shutdown() 修改的是连接的收发状态,不只是丢弃父进程的一份描述符引用,会影响子进程对同一连接的使用.
2.为什么不能在父进程里直接阻塞等待？
下面的写法会破坏这一版希望实现的接入并发：
pid_t pid = fork();
if (pid > 0)
{
close(sockfd);
waitpid(pid, nullptr, 0); // 等待业务子进程结束。
}
因为父进程必须等当前业务子进程结束,才能再次执行 accept().客户端 A 一直不发数据,其业务子进程不结束,父进程就无法继续从接入队列中取出后续连接.waitpid() 不带 WNOHANG 时可能阻塞调用者.
但完全不回收也不正确：在通常的信号配置下,已退出而未被等待回收的子进程会保留为僵尸进程.
所以这里采用：
正常路径：accept → fork → 继续 accept
回收路径：SIGCHLD → 循环 waitpid(WNOHANG)
3.子进程为什么用 _exit()？
serviceIO(sockfd, clientaddress);
_exit(EXIT_SUCCESS);
一方面要确保子进程处理完连接后直接结束,不会落回服务器的接入循环.
另一方面,_exit() 不会像 exit() 那样执行已注册的退出处理和刷新标准 I/O 缓冲区,避免再次处理从父进程继承的缓冲内容.它也不负责执行 C++ 局部对象的正常析构,所以连接应在调用它之前完成必要的清理.
原来的注释代码不是单次 fork(),而是类似：
// 第一次 fork 的子进程中：
if (fork() > 0)
{
exit(0);
}
// 第二次 fork 创建的孙进程执行：
serviceIO(sockfd, clientaddress);
它的结构是：
原父进程
│
└── 中间子进程
│
├── 再次 fork
├── 中间子进程退出
│
└── 孙进程处理客户端连接
原父进程中的：
waitpid(id, nullptr, 0);
等待的是很快退出的中间子进程,不是负责完整业务的孙进程.
因此,不能仅仅因为这里出现了阻塞 waitpid(),就断言这份双重 fork() 实现一定按连接串行处理.
这种设计把后续工作交给孙进程;其父进程退出后,它会被适当的收养进程接管,通常涉及 init 或 subreaper,不能一概认为一定由 PID 1 直接接管.
不过,对这一阶段的学习,我更推荐前面的单次 fork() + 父进程显式回收：父子关系更容易观察,也更容易对应每条连接的生命周期.
原双重 fork() 分支还有两处需要修正：第二次 fork() 没有完整区分失败情况;serviceIO() 已经关闭连接,后面又写了一次 close(sockfd).不要原样取消注释就直接使用.
1.4 V3—Echo Server多线程版：远程命令执行与结果回传
这一版把“收到什么就回显什么”，改成“接收命令 → 在服务端执行 → 返回执行结果”，并由每个客户端连接对应的工作线程独立处理。
1.从回显服务变成命令服务
原来的业务逻辑是：
客户端发送：hello
服务端返回：Server echo hello
现在变成：
客户端发送：uname
服务端执行：预设的 uname -sr
服务端返回：状态码 + 命令输出
命令在服务端机器执行,不是在客户端执行. 本工程没有切换用户身份,执行账户和工作目录来自服务端进程.
2.从每连接一个进程,改成每连接一个线程
服务端进程
│
主线程：accept()
│
┌────────────────┼────────────────┐
↓ ↓ ↓
工作线程 A 工作线程 B 工作线程 C
客户端连接 A 客户端连接 B 客户端连接 C
│ │ │
接收命令 接收命令 接收命令
↓ ↓ ↓
执行程序 执行程序 执行程序
↓ ↓ ↓
返回结果 返回结果 返回结果
↓ ↓ ↓
等待下一条 等待下一条 等待下一条
这里要区分两个层次：
连接由线程处理;外部命令仍通过创建短生命周期的子进程执行. 所以,在执行命令时看到临时子进程,并不代表服务器又变回了 V2 的“每连接一个进程”.posix_spawn() 的职责就是创建执行指定程序的新进程.
附件是独立实现,不依赖截图中的 Thread.hpp 和日志模块,便于直接编译验证.与原项目的对应关系如下：
| 模块 | V3 的职责或修改 |
|---|---|
EchoTcpServerMain.cc | 解析端口，配置停止流程，启动服务器 |
EchoTcpServer.hpp | accept() 后创建工作线程；在线程中循环处理命令 |
CommandExecutor.hpp | 新增：命令白名单、启动程序、收集输出、取得退出状态 |
Protocol.hpp | 新增：请求与响应的长度协议、完整收发、大小校验 |
UniqueFd.hpp | 新增：明确 Socket 和管道描述符的所有权 |
EchoTcpClient.cc | 不再发送裸文本，而是发送命令帧、读取完整响应 |
ThreadPool.hpp | 本版不使用线程池 |
| V2 的子进程回收器 | 不直接沿用，改由命令执行器回收对应的确切 PID |
旧版客户端和 test_echo.py、test_v2.py 不能直接测试这个新协议. 需要使用附件中的 V3 客户端和测试脚本.
1.主线程接入,工作线程接管
附件中创建线程的核心代码是：
record->thread = std::thread(
[this, record, id, conn = std::move(connection)]() mutable
{
ServiceIO(std::move(conn), id);
record->done.store(
true,
std::memory_order_release
);
}
);
这里有三个关键点.
conn = std::move(connection)：转移连接的关闭责任.
connection 是一个不可复制的 UniqueFd 对象.移动到线程后,由工作线程负责使用和关闭该连接,主线程不再持有它的所有权.
ServiceIO()：处理一条连接上的多轮请求.
不是执行完一条命令就结束线程,而是继续等待当前客户端发送下一条命令,直到 quit、连接关闭、超时或服务停止.
done：通知主线程这个线程已经完成业务.
主线程周期性检查已完成记录,再调用 join() 回收线程;不会在刚创建线程后立即阻塞等待.join() 会等待对应线程执行结束,因此创建后马上 join() 会阻塞主线程的后续接入.
2.不能照搬 V2 中父进程的 close()
V2 中常见的正确写法是：
// 父进程创建子进程后，关闭父进程持有的连接描述符。
close(sockfd);
但 V3 不能在创建工作线程后这样做：
std::thread worker(...);
close(sockfd); // 错误：工作线程还需要使用这个描述符。
同一进程内的线程共享文件描述符表。 主线程关闭这个描述符,工作线程使用的也是同一个进程里的描述符,不存在V2中那种父子进程各自持有描述符副本的关系.
同理,工作线程也不能像 V2 子进程那样关闭公共监听 Socket,或通过 exit()、_exit() 结束自己;进程退出会终止其他线程.线程任务完成后正常返回即可.
3.不要把循环变量地址传给线程
下面这种方式存在生命周期和并发访问问题：
while (true)
{
int sockfd = accept(...);
pthread_create(
&tid,
nullptr,
ThreadRoutine,
&sockfd
);
}
新线程还没有读取 sockfd,主线程可能就已经进入下一轮,甚至该局部变量已经离开作用域.
使用值捕获、独立参数对象,或者像附件这样移动连接所有者,才能让“这个线程处理哪个连接”保持明确.
1.system()、popen() 和 posix_spawn() 的区别
| 方式 | 能做什么 | 在本项目中的取舍 |
|---|---|---|
system() | 通过 Shell 执行命令并等待结束 | 本身不提供读取命令输出的流 |
popen() | 启动 Shell 命令，通过管道读取输出 | 教学代码较短，但输入字符串会被 Shell 解释 |
posix_spawn() | 按指定程序路径和参数启动子进程 | 便于固定参数、重定向输出并管理子进程 |
system() 和 popen() 都涉及 Shell 命令字符串的解释;popen(command, "r") 提供读取命令标准输出的流.附件选择 posix_spawn() + 固定参数 + 管道,不把网络输入原文交给 Shell.
2.常见的 popen() 怎么理解
接口为：
FILE* popen(const char* command, const char* type);
int pclose(FILE* stream);
例如,使用固定命令：
FILE* stream = popen("uname -sr", "r");
它的逻辑是：
启动 Shell 命令
↓
命令的标准输出进入管道
↓
当前程序从 FILE* 读取结果
↓
pclose() 关闭流并等待对应进程
必须使用 pclose(),不能用 fclose() 替代;而且 pclose() 的返回值是等待状态,需要按进程退出状态解析,并不是可以无条件当作普通退出码打印的整数.
问题在于：直接把客户端输入传给 popen(),客户端控制的就不仅是“命令名称”,还可能包含 Shell 语法.因此,本版使用完整匹配的预设命令映射.
3.本版支持的命令
help
pwd
ls
date
uname
whoami
uptime
quit
例如：
输入 uname
→ 匹配预设名称
→ 执行固定程序和参数：uname -sr
输入 ls
→ 执行固定程序和参数：ls -1
输入 ls /tmp
→ 不匹配预设名称
→ 返回状态 400
这不是简单地删除几个特殊字符,而是：客户端只能选择已定义的操作,不能自行提供程序路径、参数或 Shell 表达式.
pwd 和 ls 对应服务端启动目录;whoami 用来观察服务端运行账户.cd、管道、重定向和任意脚本不属于本版功能.
4.实际结果回传流程
客户端命令
↓
完整读取请求帧
↓
检查预设命令映射
↓
建立管道
↓
posix_spawn() 启动指定程序
│
├── stdin → /dev/null
├── stdout → 管道写端
└── stderr → 管道写端
↓
工作线程读取管道
↓
取得输出和退出状态
↓
发送完整响应帧
标准输出和标准错误可以通过 spawn 文件动作重定向到管道.服务端必须关闭自己不再需要的管道写端,否则读端可能无法观察到预期 EOF.
不能先阻塞等待命令结束，再读取它的大量输出。 管道容量有限,子进程可能因管道写满而等待读取,父线程却在等待子进程结束,形成互相等待.附件在执行期间持续读取并检查时间、输出大小,之后再取得对应进程的退出状态.
V2 使用过：
waitpid(-1, nullptr, WNOHANG);
在那个版本中,它负责回收连接处理子进程.
但 V3 的命令执行器需要：
waitpid(command_pid, &status, ...);
取得自己启动的那条命令的退出状态.
若仍有全局 SIGCHLD 处理器使用 waitpid(-1, ...),它可能先把某个命令子进程回收,导致对应工作线程无法再取得状态.显式将 SIGCHLD 设置为 SIG_IGN,同样可能影响后续等待操作.
所以附件采用：
一个命令子进程，由创建它的工作线程负责等待和回收。
另外，不要直接在工作线程中调用 execve() 来“运行命令”：成功的 execve() 会替换当前服务进程的程序，并销毁其他线程，不只是给当前线程换一项任务.
原客户端只执行一次：
read(sockfd, buffer, sizeof(buffer));
这不足以判断“一条命令的完整输出已经接收完毕”.
命令输出可能有多行、超过缓冲区,也可能为空;TCP 只提供字节流,不保留应用层消息边界.
本版采用：
请求：
┌──────────────────┬──────────────────┐
│ 命令长度 N：4字节 │ 命令内容：N字节 │
└──────────────────┴──────────────────┘
响应：
┌────────────────────┬───────────────┬─────────────────┐
│ 正文长度 4+M：4字节 │ 状态码：4字节 │ 输出内容：M字节 │
└────────────────────┴───────────────┴─────────────────┘
长度和状态码均使用网络字节序.响应长度包含状态码,不包含长度字段本身.
这样,即使命令没有任何输出,也可以返回：
正文长度 = 4
状态码 = 0
输出长度 = 0
客户端不需要等待连接关闭,就能知道这轮请求已经完成,并继续发送下一条命令.
附件的默认限制为：最多 32 个同时处理的连接、请求正文最多 64 字节、命令输出最多 64 KiB、命令执行截止时间 3 秒、接收整个请求帧的截止时间 30 秒。 超过连接上限时关闭新连接,不进入线程池队列.
完整发送由循环处理;发送使用 MSG_NOSIGNAL,避免该次发送因 SIGPIPE 导致默认的进程终止,同时仍检查实际发送错误.
1.编译
unzip echo_server_v3.zip
cd echo_server_v3
make
生成：
server_thread_cmd
client_thread_cmd
Makefile 已经配置 C++17、线程选项和头文件依赖.线程选项使用 -pthread,同时用于编译和链接.
2.终端 A：启动服务端
./server_thread_cmd 8888
日志会显示实际的 PID、线程 ID 和连接编号,例如以下形式：
[pid=实际PID tid=实际TID conn=0] listening on 127.0.0.1:8888 max_clients=32
保持这个终端运行.
3.终端 B：启动客户端
./client_thread_cmd 127.0.0.1 8888
可以在同一连接上依次输入：
help
pwd
uname
date
whoami
quit
其中 help 的响应为：
cmd> help
[status=0]
Commands: help, pwd, ls, date, uname, whoami, uptime, quit
Only exact names are accepted; no arguments or shell syntax.
输入 quit：
cmd> quit
[status=0]
bye
也支持只执行一次命令：
./client_thread_cmd 127.0.0.1 8888 uname
与前面版本不同,这次客户端出现下一次输入提示时,服务端确实会继续处理该连接上的下一条命令. 但空闲超过本版请求截止时间后,需要重新连接.
建议保持服务只监听回环地址,通过 SSH 转发访问.SSH 支持将本地端口的连接通过加密通道转发到远程主机上的指定地址和端口.
1.在远程服务器上运行服务
./server_thread_cmd 8888
使用普通账户启动,不要为了这个实验使用 sudo.
2.在本地建立隧道
将 user@SERVER_IP 替换成你的 SSH 账户和服务器地址：
ssh -N -o ExitOnForwardFailure=yes \
-L 127.0.0.1:18888:127.0.0.1:8888 \
user@SERVER_IP
这里的路径是：
本地客户端
↓
本地 127.0.0.1:18888
↓
SSH 加密通道
↓
远程服务器 127.0.0.1:8888
↓
V3 命令服务
-N表示只建立转发而不执行远程命令;-L 指定本地转发.SSH服务端需要允许相应转发.
3.本地另开终端连接转发端口
./client_thread_cmd 127.0.0.1 18888
此时输入 pwd、uname 等,取得的是远程服务端的结果.
需要明确：附件没有独立的应用层认证或 TLS;本机其他能够访问该端口的进程也可以请求预设命令.SSH 隧道用于保护跨主机访问,不会把这个程序自动变成完整的多用户权限系统.
1.手动测试：空闲连接不阻塞其他连接
启动三个客户端.
前两个客户端连接后暂时不输入;第三个客户端执行：
uname
在没有达到连接上限的情况下,第三个客户端应能正常取得响应,而不是等待前两个客户端发出命令.
从服务端日志取得 PID,再查看线程：
# 将 12345 替换为实际服务端 PID。
ps -T -p 12345 -o pid,spid,stat,comm
ps -T 用于显示线程.多个工作线程会显示在同一个服务进程下,通过不同的 SPID 区分.
2.自动测试
让测试自行启动临时服务、完成验证并清理：
make test
测试已经运行的本地服务：
python3 test_v3.py --port 8888
测试通过 SSH 转发的服务：
python3 test_v3.py --port 18888
连接已有服务的模式不会替你停止服务.
3.附件已完成的本地验证
| 验证项目 | 本次结果 |
|---|---|
GCC 严格警告检查，包含 -Werror | 通过 |
| 同一连接执行多条命令并正常退出 | 通过 |
| 请求逐字节分段、多个请求一次合并发送 | 通过 |
| 24 个并发客户端，每个执行 4 轮请求 | 通过 |
| 32 个连接对应 32 个工作线程，额外连接被关闭 | 通过 |
| 标准输出、标准错误、退出码及空输出 | 通过 |
| 大输出、输出超限、命令超时和子进程回收 | 通过 |
| 客户端异常断开、线程和描述符回收 | 通过 |
| AddressSanitizer / UndefinedBehaviorSanitizer 构建测试 | 通过 |
1.5 V4—Echo Server线程池版：引入英汉翻译与业务解耦
这一版将“每个连接创建一个线程”改成“固定线程池处理连接”,并把业务从远程命令执行替换为英汉词典查询.
这里将“翻译”理解为：客户端发送 apple,服务端返回 苹果.采用本地词典实现,不调用外部翻译API,也不提供任意整句的机器翻译.
1.线程的创建方式改变了
V3 的处理方式是：
客户端 A 连接 → 创建线程 A → 处理连接 → 线程退出
客户端 B 连接 → 创建线程 B → 处理连接 → 线程退出
客户端 C 连接 → 创建线程 C → 处理连接 → 线程退出
V4 则提前创建一组工作线程：
服务器启动
│
└── 创建固定数量的工作线程
│
└── 没有任务时等待，有任务时取出执行
客户端连接
│
└── accept() → 封装连接任务 → 放入队列
│
└── 已有工作线程处理
一个连接处理结束后,工作线程不会退出,而是继续取下一个任务.
| 对比项 | V3：每连接一个线程 | V4：固定线程池 |
|---|---|---|
| 接入新连接 | 创建新工作线程 | 提交连接任务 |
| 连接处理结束 | 对应线程结束 | 工作线程继续等待任务 |
| 工作线程都忙时 | 在连接上限内继续创建线程 | 新任务进入有界等待队列 |
| 本版业务 | 执行预设命令 | 查询本地英汉词典 |
这里的线程池是附件实现的调度机制,不是 listen() 或 TCP 自动提供的功能.
2.网络逻辑与业务逻辑分开了
本版分为三个核心职责：
TcpServer 负责连接、收发和协议处理
ThreadPool 负责连接任务的排队和执行
Dict 负责英文查询与中文释义
这样,之后把“翻译”换回“回显”,主要修改业务回调,不需要重写线程池.
客户端发送 apple
│
▼
主线程 accept() 接收连接
│
▼
把“处理这条连接”封装成任务
│
▼
┌─────────────────────────────────────┐
│ 任务队列 │
│ 连接任务 A → 连接任务 B → … │
└─────────────────────────────────────┘
│ │
▼ ▼
工作线程 1 工作线程 2
│ │
└──────┬───────────┘
▼
ServiceIO()
│
├── 接收完整请求
├── 调用业务回调
├── Dict::Translate("apple")
└── 返回“苹果”
│
▼
继续等待该连接的下一条请求
特别注意：本版排队的是“连接任务”,不是“单词查询任务”.
一条连接进入 ServiceIO() 后,会由同一个工作线程持续处理,直到客户端退出、断开、超时或服务停止.即使客户端暂时没有发送数据,它仍然占用这个工作线程.
这会直接影响后面的容量测试.
附件中的 ThreadPool.hpp 使用标准库组件,对应你原来封装的模块：
| 原项目封装 | 本版使用 |
|---|---|
Thread | std::thread |
Mutex | std::mutex |
Cond | std::condition_variable |
LockGuard | std::lock_guard / std::unique_lock |
线程池内部的关键成员是：
using Task = std::function<void()>;
std::vector<std::thread> workers_; // 固定工作线程
std::queue<Task> tasks_; // 等待任务
std::mutex mutex_;
std::condition_variable ready_;
bool stopping_ = false;
1.提交任务：加锁入队，再通知工作线程
核心代码：
bool TryEnqueue(Task task)
{
if (!task)
throw std::invalid_argument("empty task");
{
std::lock_guard<std::mutex> lock(mutex_);
if (stopping_ || tasks_.size() >= capacity_)
return false;
tasks_.push(std::move(task));
}
ready_.notify_one();
return true;
}
这里有三个重要设计.
首先，队列有容量上限。 队列已满或线程池停止后,返回 false,不会无限积累等待连接.
其次，提交者不等待队列出现空位。 服务端可以及时关闭无法接管的新连接,而不是把接入主线程也堵住.
最后，通知不是任务本身。 任务保存在队列中;条件变量只是帮助等待线程重新检查队列状态。标准库的 notify_one() 会唤醒一个正在等待该条件变量的线程,但业务是否可以继续,仍需要检查共享状态.
2.工作线程：锁内取任务,锁外执行
工作循环的核心是：
for (;;)
{
Task task;
{
std::unique_lock<std::mutex> lock(mutex_);
ready_.wait(lock, [this]
{
return stopping_ || !tasks_.empty();
});
if (stopping_ && tasks_.empty())
return;
task = std::move(tasks_.front());
tasks_.pop();
} // 离开作用域，释放队列锁。
try
{
task();
}
catch (...)
{
task_failures_.fetch_add(1, std::memory_order_relaxed);
}
}
其中最重要的一点是：执行 task() 时，不能持有任务队列的锁。
否则,一个线程等待客户端数据时,其他工作线程可能无法取任务,主线程也无法正常入队.
wait(lock, predicate) 相当于循环检查条件：不满足时等待,醒来后重新检查.等待期间会释放互斥锁,返回前重新持有锁,因此不是“线程抱着锁睡觉”.
截图的线程池在 LockGuard 生效的作用域内,又手动执行过一次 Unlock().本版通过作用域自动解锁,不再混用那种写法.
3.停止线程池,不只是调用 notify_all()
附件区分两种线程池停止方式：
pool.Shutdown(false); // 不再接新任务，处理完队列后退出。
pool.Shutdown(true); // 不再接新任务，取消尚未开始的任务。
但两者都不能强行终止一个正在执行的业务函数.
服务端采用的停止流程是：
停止接入新连接
→ 设置业务停止标志
→ 取消尚未开始的连接任务
→ 正在收发的连接检查停止标志并退出
→ join 所有工作线程
协议层使用可检查截止时间和停止状态的I/O等待,避免只唤醒条件变量、却把线程留在网络等待里.join() 则等待线程真正结束,并不是“发出停止请求”.
服务端不再这样创建线程：
std::thread worker(...);
而是提交一个可调用对象：
auto client =
std::make_shared<UniqueFd>(std::move(connection));
const bool accepted = pool_.TryEnqueue(
[this, client, id]
{
if (!stopping_.load(std::memory_order_relaxed))
ServiceIO(client->get(), id);
}
);
1.Lambda 保存了什么？
this 用于调用服务器的 ServiceIO();client 保存连接资源的所有者;id 是本次连接的日志编号.
连接任务离开主线程后,这些信息仍然有效,不依赖循环里的临时变量地址.
2.为什么使用 shared_ptr<UniqueFd>？
UniqueFd 负责在释放时关闭描述符,而且禁止复制.
但 C++17 的 std::function 要求其保存的可调用目标能够复制.因此,直接把只可移动的连接所有者捕获进 Lambda,通常不能放入这里的 std::function<void()>.附件使用 shared_ptr 包装连接所有者,让闭包可以复制,同时确保最终只关闭一次连接.
这不意味着多个工作线程同时读取同一条连接.当前每个连接只提交一个处理任务.
3.入队后,主线程不能马上 close(sockfd)
V2 多进程版中,父进程可以关闭自己持有的通信描述符副本.
线程池版不同：同一进程的线程共享文件描述符表。 主线程直接关闭工作线程还要使用的描述符,会破坏该连接的处理.
附件统一由连接所有者负责释放,因此：
入队失败 → 连接所有者释放 → 关闭连接
任务结束 → 连接所有者释放 → 关闭连接
ServiceIO() 中不再额外执行一次 close(),避免重复关闭.
1.词典文件保存什么？
工程中的 dict.txt 使用 UTF-8，每行一个词条：
apple:苹果
banana:香蕉
hello:你好
world:世界
thread:线程
thread pool:线程池
socket:套接字
mutex:互斥锁
冒号使用英文半角 :。
本版启动时加载词典,之后只读.修改文件后需要重启服务器,不提供在线热更新.
2.查询逻辑
Translate() 的核心步骤是：
const auto key = Normalize(raw);
if (!ValidKey(key))
return {400, "请输入英文单词或词组。"};
const auto it = entries_.find(key);
if (it == entries_.end())
return {404, "未收录: " + key};
return {0, it->second};
完整实现还处理了 help、输入长度和词典文件格式等情况.
例如：
| 请求 | 本版结果 |
|---|---|
apple | 苹果 |
APPLE | 苹果 |
apple | 苹果 |
thread pool | 线程池 |
| 未收录的单词 | 状态 404 |
| 空请求或不符合约定的输入 | 状态 400 |
这里的 400、404 是本项目自定义响应状态,不是说这个服务使用了 HTTP.
3.多个线程查询同一份词典，要不要加锁？
本版满足两个前提：
词典在工作线程启动前加载完成；运行期间所有线程只读取，不修改。
因此可以使用只读的 find() 查询,而不必给每次查询套一把全局互斥锁.标准库将 find() 等操作按只读操作考虑;但是,这个结论不能推广为“一个线程修改容器时,其他线程也能随便读取”.
不要写成：
return entries_[word];
对于关联容器,operator[] 在键不存在时可能插入元素,不适合作为这里的并发只读查询接口.
4.将业务回调注入服务器
入口中的关键代码是：
auto dict = std::make_shared<const v4::Dict>("dict.txt");
v4::TcpServer server(
8888,
[dict](const std::string& word)
{
return dict->Translate(word);
},
4, // 工作线程数
64 // 等待队列容量
);
这里使用 shared_ptr<const Dict>,让共享只读和生命周期都比较明确.
服务器只调用回调：
response = handler_(query);
它不需要知道词典内部使用了什么容器.
要重新演示Echo,可以把回调替换成：
[](const std::string& text)
{
return v4::Response{0, text};
}
线程池和 TCP 收发框架都不需要改变.
不能把“一次 recv() 返回”直接当成“一次翻译响应结束”.TCP是字节流,不保留应用消息边界;中文结果也应按实际字节长度处理.
本版沿用 V3 附件的帧结构：
请求：
[4 字节正文长度 N] [N 字节查询内容]
响应：
[4 字节正文长度 4+M] [4 字节状态码] [M 字节查询结果]
长度和状态码采用网络字节序;长度字段不包含自身.
协议层负责循环收发、检查长度、识别帧中途断开,并设置整个请求帧的截止时间.发送使用 MSG_NOSIGNAL,但仍然检查返回值和发送错误,不把一次成功发送当成完整业务交付确认.
本版默认限制为：
| 项目 | 默认值 |
|---|---|
| 请求正文 | 最多 128 字节 |
| 单次查询结果 | 最多 4096 字节 |
| 工作线程开始读取后，接收整个请求帧的时间 | 30 秒 |
| 发送整个响应的时间 | 5 秒 |
V1/V2 的裸字符串客户端不能直接使用这个协议。 推荐使用附件中的 V4 客户端.
二、验证TCP—Windows作为客户端访问Linux
//Makefile
.PHONY:all
all:tcp_server udp_server
tcp_server:TcpServer.cc
g++ -o $@ $^ -std=c++14
udp_server:UdpServer.cc
g++ -o $@ $^ -std=c++14
.PHONY:clean
clean:
rm -rf tcp_server udp_server
//TcpClient.cc
#include <winsock2.h>
#include <iostream>
#include <string>
#pragma warning(disable : 4996)
#pragma comment(lib, "ws2_32.lib")
std::string serverip = ""; // 填写你的云服务器ip
uint16_t serverport = 8888; // 填写你的云服务开放的端口号
int main()
{
WSADATA wsaData;
int result = WSAStartup(MAKEWORD(2, 2), &wsaData);
if (result != 0)
{
std::cerr << "WSAStartup failed: " << result << std::endl;
return 1;
}
SOCKET clientSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
if (clientSocket == INVALID_SOCKET)
{
std::cerr << "socket failed" << std::endl;
WSACleanup();
return 1;
}
sockaddr_in serverAddr;
serverAddr.sin_family = AF_INET;
serverAddr.sin_port = htons(serverport); // 替换为服务器端口
serverAddr.sin_addr.s_addr = inet_addr(serverip.c_str()); // 替换为服务器IP地址
result = connect(clientSocket, (SOCKADDR *)&serverAddr, sizeof(serverAddr));
if (result == SOCKET_ERROR)
{
std::cerr << "connect failed" << std::endl;
closesocket(clientSocket);
WSACleanup();
return 1;
}
while (true)
{
std::string message;
std::cout << "Please Enter@ ";
std::getline(std::cin, message);
if(message.empty()) continue;
send(clientSocket, message.c_str(), message.size(), 0);
char buffer[1024] = {0};
int bytesReceived = recv(clientSocket, buffer, sizeof(buffer) - 1, 0);
if (bytesReceived > 0)
{
buffer[bytesReceived] = '\0'; // 确保字符串以 null 结尾
std::cout << "Received from server: " << buffer << std::endl;
}
else
{
std::cerr << "recv failed" << std::endl;
}
}
closesocket(clientSocket);
WSACleanup();
return 0;
}
//TcpServer.cc
#include <iostream>
#include <string>
#include <cerrno>
#include <cstring>
#include <cstdlib>
#include <memory>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/wait.h>
#include <unistd.h>
const static int default_backlog = 6;
enum
{
Usage_Err = 1,
Socket_Err,
Bind_Err,
Listen_Err
};
#define CONV(addr_ptr) ((struct sockaddr *)addr_ptr)
class TcpServer
{
public:
TcpServer(uint16_t port) : _port(port), _isrunning(false)
{
}
// 都是固定套路
void Init()
{
// 1. 创建socket, file fd, 本质是文件
_listensock = socket(AF_INET, SOCK_STREAM, 0);
if (_listensock < 0)
{
exit(0);
}
int opt = 1;
setsockopt(_listensock, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt));
// 2. 填充本地网络信息并bind
struct sockaddr_in local;
memset(&local, 0, sizeof(local));
local.sin_family = AF_INET;
local.sin_port = htons(_port);
local.sin_addr.s_addr = htonl(INADDR_ANY);
// 2.1 bind
if (bind(_listensock, CONV(&local), sizeof(local)) != 0)
{
exit(Bind_Err);
}
// 3. 设置socket为监听状态，tcp特有的
if (listen(_listensock, default_backlog) != 0)
{
exit(Listen_Err);
}
}
void ProcessConnection(int sockfd, struct sockaddr_in &peer)
{
uint16_t clientport = ntohs(peer.sin_port);
std::string clientip = inet_ntoa(peer.sin_addr);
std::string prefix = clientip + ":" + std::to_string(clientport);
std::cout << "get a new connection, info is : " << prefix << std::endl;
while (true)
{
char inbuffer[1024];
ssize_t s = ::read(sockfd, inbuffer, sizeof(inbuffer)-1);
if(s > 0)
{
inbuffer[s] = 0;
std::cout << prefix << "# " << inbuffer << std::endl;
std::string echo = inbuffer;
echo += "[tcp server echo message]";
write(sockfd, echo.c_str(), echo.size());
}
else
{
std::cout << prefix << " client quit" << std::endl;
break;
}
}
}
void Start()
{
_isrunning = true;
while (_isrunning)
{
// 4. 获取连接
struct sockaddr_in peer;
socklen_t len = sizeof(peer);
int sockfd = accept(_listensock, CONV(&peer), &len);
if (sockfd < 0)
{
continue;
}
ProcessConnection(sockfd, peer);
}
}
~TcpServer()
{
}
private:
uint16_t _port;
int _listensock; // TODO
bool _isrunning;
};
using namespace std;
void Usage(std::string proc)
{
std::cout << "Usage : \n\t" << proc << " local_port\n"
<< std::endl;
}
// ./tcp_server 8888
int main(int argc, char *argv[])
{
if (argc != 2)
{
Usage(argv[0]);
return Usage_Err;
}
uint16_t port = stoi(argv[1]);
std::unique_ptr<TcpServer> tsvr = make_unique<TcpServer>(port);
tsvr->Init();
tsvr->Start();
return 0;
}
三、Connect的断线重连
客户端会⾯临服务器崩溃的情况,我们可以试着写⼀个客户端重连的代码,模拟并理解⼀些客户端⾏为,⽐如游戏客户端等.采⽤状态机,实现⼀个简单的tcp client可以实现重连效果.
//Makefile
tcpclient:TcpClient.cc
g++ -o $@ $^ -std=c++11
.PHONY:clean
clean:
rm -f tcpclient
//TcpClient.cc
#include <iostream>
#include <string>
#include <cstring>
#include <cstdlib>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
using namespace std;
void Usage(const std::string &process)
{
std::cout << "Usage: " << process << " server_ip server_port" << std::endl;
}
enum class Status // C++11 强类型枚举
{
NEW, // 新建状态，就是单纯的连接
CONNECTING, // 正在连接，仅仅方便查询conn状态
CONNECTED, // 连接或者重连成功
DISCONNECTED, // 重连失败
CLOSED // 连接失败，经历重连，无法连接
};
class ClientConnection
{
public:
ClientConnection(uint16_t serverport, const std::string &serverip)
: _sockfd(-1),
_serverport(serverport),
_serverip(serverip),
_retry_interval(1),
_max_retries(5),
_status(Status::NEW)
{
}
void Connect()
{
// 1. 创建socket
_sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (_sockfd < 0)
{
cerr << "socket error" << endl;
exit(1);
}
// 2. 要不要bind？必须要有Ip和Port, 需要bind，但是不需要用户显示的bind，client系统随机端口
// 发起连接的时候，client会被OS自动进行本地绑定
// 2. connect
struct sockaddr_in server;
memset(&server, 0, sizeof(server));
server.sin_family = AF_INET;
server.sin_port = htons(_serverport);
// p:process(进程), n(网络) -- 不太准确，但是好记忆
inet_pton(AF_INET, _serverip.c_str(), &server.sin_addr); // 1. 字符串ip->4字节IP 2. 网络序列
int n = connect(_sockfd, (struct sockaddr *)&server, sizeof(server)); // 自动进行bind哦！
if (n < 0)
{
Disconnect(); // 恢复_sockfd的默认值，是连接没有成功，不代表sockfd创建没有成功
_status = Status::DISCONNECTED; // 没有连接成功
return;
}
_status = Status::CONNECTED; // 连接成功
}
int SocketFd()
{
return _sockfd;
}
void Reconnect()
{
_status = Status::CONNECTING; // 正在重连
int count = 0;
while (count < _max_retries)
{
Connect(); // 重连
if (_status == Status::CONNECTED)
{
return;
}
sleep(_retry_interval);
count++;
std::cout << "重连次数: " << count << ", 最大上限: " << _max_retries << std::endl;
}
_status = Status::CLOSED; // 重连失败，可以关闭了
}
void Disconnect()
{
if (_sockfd != -1)
{
close(_sockfd);
_status = Status::CLOSED;
_sockfd = -1;
}
}
Status GetStatus()
{
return _status;
}
void Process()
{
// 简单的IO即可
while (true)
{
string inbuffer;
cout << "Please Enter# ";
getline(cin, inbuffer);
if(inbuffer.empty()) continue;
ssize_t n = write(_sockfd, inbuffer.c_str(), inbuffer.size());
if (n > 0)
{
char buffer[1024];
ssize_t m = read(_sockfd, buffer, sizeof(buffer) - 1);
if (m > 0)
{
buffer[m] = 0;
cout << "echo messsge -> " << buffer << endl;
}
else if (m == 0) // 这里证明server端掉线了
{
_status = Status::DISCONNECTED;
break;
}
else
{
std::cout << "read m : " << m << "errno: " << errno << "errno string: " << strerror(errno) << std::endl;
_status = Status::CLOSED;
break;
}
}
else
{
std::cout << "write n : " << n << "errno: " << errno << "errno string: " << strerror(errno) << std::endl;
_status = Status::CLOSED;
break;
}
}
}
~ClientConnection()
{
Disconnect();
}
private:
int _sockfd;
uint16_t _serverport; // server port 端口号
std::string _serverip; // server ip地址
int _retry_interval; // 重试时间间隔
int _max_retries; // 重试次数
Status _status; // 连接状态
};
class TcpClient
{
public:
TcpClient(uint16_t serverport, const std::string &serverip) : _conn(serverport, serverip)
{
}
void Execute()
{
while (true)
{
switch (_conn.GetStatus())
{
case Status::NEW:
_conn.Connect();
break;
case Status::CONNECTED:
std::cout << "连接成功, 开始进行通信." << std::endl;
_conn.Process();
break;
case Status::DISCONNECTED:
std::cout << "连接失败或者对方掉线，开始重连." << std::endl;
_conn.Reconnect();
break;
case Status::CLOSED:
_conn.Disconnect();
std::cout << "重连失败, 退出." << std::endl;
return; // 退出
default:
break;
}
}
}
~TcpClient()
{
}
private:
ClientConnection _conn; // 简单组合起来即可
};
// class Tcp
// ./tcp_client serverip serverport
int main(int argc, char *argv[])
{
if (argc != 3)
{
Usage(argv[0]);
return 1;
}
std::string serverip = argv[1];
uint16_t serverport = stoi(argv[2]);
TcpClient client(serverport, serverip);
client.Execute();
return 0;
}
Connect 断线重连的核心思想是：客户端在 TCP 连接断开后,不立即退出程序,而是通过连接状态管理和重试机制,重新建立与服务器的连接,尽可能恢复正常通信.
一、图1：Connect 断线重连的代码模块与职责
1.Makefile：项目编译
tcpclient:TcpClient.cc
g++ -o $@ $^ -std=c++11
.PHONY:clean
clean:
rm -f tcpclient
Makefile 负责把 TcpClient.cc 编译为 tcpclient 可执行文件,使用C++11标准.
其中 $@ 表示编译目标,$^ 表示依赖文件.
2.ClientConnection：连接管理类
从代码结构来看,客户端将连接建立、数据收发、异常处理和重新连接封装在连接管理类中.
| 核心函数 | 主要作用 |
|---|---|
Connect() | 创建 Socket，并尝试连接服务器 |
Send() | 向服务器发送数据 |
Recv() | 接收服务器返回的数据 |
Reconnect() | 连接断开后重新尝试建立连接 |
Disconnect() | 关闭连接、释放 Socket 资源 |
GetStatus() | 获取当前连接状态 |
其中,Connect() 负责单次连接尝试,Reconnect() 负责在失败后重复调用连接操作,两者的职责不同.
3.连接状态管理
程序通过状态枚举管理 TCP 客户端的生命周期.
这样设计的好处是：程序不需要只依赖某个 Socket API 的返回值决定全部行为,而是可以根据连接状态选择下一步操作.
例如,连接正常时执行收发;检测到断开后进入重连;达到重试上限仍失败时结束客户端流程.
二、图2：Connect 的断线重连流程
1.第一次建立连接
程序启动后,首先调用 Connect().
其内部主要执行：
socket()
↓
填充 sockaddr_in
↓
connect()
↓
判断连接结果
如果 connect() 成功,说明TCP连接建立,客户端进入 CONNECTED 状态.
如果失败,则进入异常处理流程,等待后续重连.
2.正常收发数据
连接成功后,客户端可以通过 Send() 和 Recv() 与服务器通信：
客户端 服务器
│ │
├───── Send("hello") ────→│
│ │
│←──── 返回响应数据 ──────┤
│ │
└──── Recv() 读取响应 │
只要连接保持正常,客户端就可以继续通信.
3.如何发现连接断开？
断线可能由服务器关闭、网络中断或对端异常引起.
特别要理解 recv() 的三种返回情况：
| 返回值 | 意义 | 处理 |
|---|---|---|
n > 0 | 接收到数据 | 继续通信 |
n == 0 | 对端已正常结束发送 | 根据协议决定关闭或处理半关闭 |
n < 0 | 接收出现错误 | 检查 errno，判断是否需要重连 |
例如：
ssize_t n = recv(sockfd, buffer, sizeof(buffer), 0);
if (n > 0)
{
// 处理服务器数据
}
else if (n == 0)
{
// 对端结束发送
// 根据连接策略进行清理和重连
}
else
{
// 检查 errno
// EINTR 或 EAGAIN 等情况不能一概视为断线
}
需要注意,服务器突然宕机或网络突然断开,不代表客户端立即就能知道连接已经失效.如果客户端一直没有进行网络 I/O,也没有设置心跳或超时机制,可能无法及时发现异常.
4.Reconnect()：重新建立 TCP 连接
根据图中的重试逻辑及图2的演示配置,客户端采用有限次数的重试机制.
以最多重试3次、间隔1秒为例：
检测到连接断开
↓
关闭原有 Socket
↓
第 1 次 Connect()
│
├── 成功 → 恢复通信
│
└── 失败
↓
等待 1 秒
↓
第 2 次 Connect()
│
├── 成功 → 恢复通信
│
└── 失败
↓
等待 1 秒
↓
第 3 次 Connect()
│
├── 成功 → 恢复通信
│
└── 失败 → CLOSE
这里有一个非常重要的 Socket 编程知识：
TCP 连接断开后,不能把已经失效的 Socket 当作原连接继续使用,而应该关闭旧描述符,创建新的Socket,再执行 connect().
因此,断线重连不是恢复原有TCP连接,而是重新建立一条新的连接.
三、图3：断线重连的实际运行场景
假设客户端正在连接一台 Linux TCP 服务器.
场景一：服务器正常运行
客户端成功建立连接：
客户端：connect() success
状态：CONNECTED
客户端发送：
hello
服务器正常返回响应,通信继续进行.
场景二：服务器突然关闭
假设此时关闭 Linux 服务器进程.
客户端在后续读写过程中检测到连接被关闭或重置,进入 DISCONNECTED 状态.
客户端不会立即放弃,而是启动 Reconnect(),重新连接服务器.
场景三：服务器恢复
这里可能出现两种情况：
情况 A：服务器在重试期间恢复
某次 Connect() 成功,连接状态重新变为 CONNECTED,客户端可以继续收发数据.但原连接上的未完成请求不会自动恢复,需要应用层另行处理.
情况 B：所有重试都失败
连接始终无法建立,客户端进入 CLOSE 状态并退出正常通信流程.有限次数重试意味着客户端不会无限等待服务器恢复.
这个实验说明：断线重连能够提高客户端的容错能力,但连接恢复时间、重试次数和服务器恢复时机都会影响最终结果.
四、如何验证这个断线重连程序？
按照当前的项目,可以设计以下实验.
| 步骤 | 操作 | 验证内容 |
|---|---|---|
| 1 | 启动 Linux TCP 服务器 | 确保服务监听指定 IP 和端口 |
| 2 | 编译并运行 tcpclient | 检查第一次连接能否成功 |
| 3 | 客户端发送 hello | 验证正常数据通信 |
| 4 | 手动停止服务器 | 制造连接断开的场景 |
| 5 | 观察客户端输出 | 检查是否进入重连状态 |
| 6 | 在重试窗口内重新启动服务器 | 验证能否重新建立连接 |
| 7 | 再发送一条消息 | 验证重连后能否正常通信 |
需要特别注意：如果代码最多只重试3次,而服务器恢复得太慢,客户端可能已经进入 CLOSE 状态,此时再启动服务器也不会触发新的重试.
另外,重连测试不能只看 connect() 成功,还应当验证重新连接后是否可以发送和接收业务数据.
五、当前断线重连方案的优点与不足
| 设计 | 优点 | 局限 |
|---|---|---|
| 状态枚举 | 连接生命周期清晰 | 状态变化需要正确同步 |
封装 Connect() | 连接代码便于复用 | 每次尝试需要正确管理 Socket |
| 自动重试 | 提高临时故障恢复能力 | 有限重试可能错过服务器恢复 |
sleep() 控制重试频率 | 避免失败后立即反复连接 | 阻塞当前线程 |
Recv() 检查连接异常 | 能检测部分断线场景 | 无 I/O 时不能保证及时发现断线 |
对于后续版本,我建议优先完善三个方面：增加心跳或超时检测以发现长期失效的连接;使用有上限的指数退避调整重试间隔;明确重连后未完成请求的处理方式,避免重复执行或丢失业务数据.
六、总结
这个项目的核心不只是重复调用 connect(),而是构建了一套连接状态管理机制.
程序通过 Connect() 建立连接,利用 Send() 和 Recv() 完成数据通信;当网络异常或服务器断开时,通过状态转换进入 Reconnect(),重新创建 Socket 并尝试连接服务器;如果连接恢复,就重新进入正常通信状态,否则在达到重试上限后结束客户端运行.
因此,Connect 断线重连的本质是：通过状态管理、异常检测和有限重试,实现 TCP 客户端连接的自动恢复,提高程序在网络不稳定环境下的可靠性.
敬请期待下一篇文章内容
每日心灵鸡汤: 人生最大的清醒,是夺回自己的选择权!
成长的一生,就是不断对抗的一生:对抗原生家庭留在你身上的恐惧、匮乏、讨好与限制;对抗社会环境塞给你的标准答案、身份模板和成功定义;对抗宏大叙事替你决定什么重要、什么正确、什么值得追求;对抗群体共识带来的从众、安全感和自我怀疑;对抗欲望、情绪、虚荣、懒惰与即时满足;最后还要对抗大脑自身的惯性,它天然偏爱熟悉、重复和低消耗.而真正的成长往往意味着进入陌生、承受不适、重新选择.人成熟的过程,本质上就是不断辨认:哪些东西真正属于自己,哪些只是家庭、社会、时代和大脑写进你身体里的程序,然后一点点夺回对自己人生的解释权和选择权.
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/2401_87629362/article/details/167221450
