Socket：应用程序看世界的那扇窗

目录一、什么是Socket二、 Socket API 调用流程三、Socket 实战TCP / UDP 完整代码3.1 TCP 回显服务器3.2 、TCP客户端3.3、UDP收发四、如何判断一个请求包收完了粘包与拆包4.1 现象4.2 成因4.3 三种解决方案五、IO 模型演进从阻塞到 epoll Reactor5.1 五种 IO 模型5.2 select / poll / epoll 对比5.3 LT 与 ET5.4 epoll 非阻塞 Reactor 骨架5.5 Reactor 模式与主从多线程模型Socket 不是协议而是操作系统提供给应用程序的一套网络编程 API 抽象。在 Unix 一切皆文件 的哲学下socket 就是一个特殊的文件描述符fd指向内核里的一个数据结构这个结构里最重要的是两个缓冲区理解这张图就能解释很多现象• send() 返回成功 ≠ 对方收到了。它只表示数据已经从用户缓冲区拷贝到内核发送缓冲区。真正的发送由内核协议栈择机进行。• send() 可能返回小于请求的字节数短写。当发送缓冲区快满时阻塞模式会阻塞非阻塞模式会返回已写入的字节数或 EAGAIN。所以必须循环写。• recv() 返回 0 表示对端优雅关闭收到 FIN返回 -1 才是出错。• 粘包的本质应用层的消息边界信息在写入字节流缓冲区的那一刻就丢失了。一个 TCP 连接由四元组唯一标识(源IP, 源端口, 目的IP, 目的端口)。这解释了为什么一个服务器的 80 端口能同时服务几万个客户端——因为客户端的 IP/端口不同四元组就不同。二、 Socket API 调用流程几个容易踩的坑• listen(fd, backlog) 的 backlog 在 Linux 2.2 之后指的是全连接队列的长度实际值取 min(backlog, net.core.somaxconn)后者默认 128高并发场景要调大。• accept() 返回的是一个新的 fd用于和这个客户端通信监听 fd 继续监听。• 不设 SO_REUSEADDR服务器重启时会因为上次连接的 TIME_WAIT 而 bind 失败Address already in use。三、Socket 实战TCP / UDP 完整代码以下代码在 Linux 下 g -stdc11 可直接编译。Windows 下需要 WSAStartup() 并把 close 换成 closesocket。3.1 TCP 回显服务器// tcp_server.cpp g tcp_server.cpp -o server -lpthread #include sys/socket.h #include netinet/in.h #include netinet/tcp.h #include arpa/inet.h #include unistd.h #include cstring #include cstdio #include thread constexpr int kPort 8888; // 循环写处理短写返回 true 表示全部写出 static bool sendAll(int fd, const char* buf, size_t len) { size_t sent 0; while (sent len) { // MSG_NOSIGNAL对端已关闭时返回 EPIPE 而不是触发 SIGPIPE 杀掉进程 ssize_t n ::send(fd, buf sent, len - sent, MSG_NOSIGNAL); if (n 0) { sent static_castsize_t(n); continue; } if (n 0 (errno EINTR)) continue; return false; } return true; } static void handleClient(int connFd, sockaddr_in peer) { char ip[INET_ADDRSTRLEN] {0}; inet_ntop(AF_INET, peer.sin_addr, ip, sizeof(ip)); printf([] client %s:%d connected, fd%d\n, ip, ntohs(peer.sin_port), connFd); char buf[4096]; while (true) { ssize_t n ::recv(connFd, buf, sizeof(buf), 0); if (n 0) { if (!sendAll(connFd, buf, static_castsize_t(n))) break; } else if (n 0) { printf([-] peer closed (FIN), fd%d\n, connFd); // 收到 FIN break; } else { if (errno EINTR) continue; // 被信号打断重试 perror(recv); break; } } ::close(connFd); // 必须关闭否则堆积 CLOSE_WAIT } int main() { int listenFd ::socket(AF_INET, SOCK_STREAM, 0); if (listenFd 0) { perror(socket); return -1; } int on 1; setsockopt(listenFd, SOL_SOCKET, SO_REUSEADDR, on, sizeof(on)); sockaddr_in addr{}; addr.sin_family AF_INET; addr.sin_addr.s_addr htonl(INADDR_ANY); // 主机序 → 网络序大端 addr.sin_port htons(kPort); if (::bind(listenFd, (sockaddr*)addr, sizeof(addr)) 0) { perror(bind); return -1; } if (::listen(listenFd, 1024) 0) { perror(listen); return -1; } printf(server listening on %d ...\n, kPort); while (true) { sockaddr_in peer{}; socklen_t len sizeof(peer); int connFd ::accept(listenFd, (sockaddr*)peer, len); if (connFd 0) { if (errno EINTR) continue; perror(accept); break; } int nodelay 1; setsockopt(connFd, IPPROTO_TCP, TCP_NODELAY, nodelay, sizeof(nodelay)); std::thread(handleClient, connFd, peer).detach(); } ::close(listenFd); return 0; }3.2 、TCP客户端// tcp_client.cpp #include sys/socket.h #include netinet/in.h #include arpa/inet.h #include unistd.h #include cstdio #include cstring int main(int argc, char* argv[]) { const char* ip (argc 1) ? argv[1] : 127.0.0.1; int port (argc 2) ? atoi(argv[2]) : 8888; int fd ::socket(AF_INET, SOCK_STREAM, 0); sockaddr_in srv{}; srv.sin_family AF_INET; srv.sin_port htons(port); inet_pton(AF_INET, ip, srv.sin_addr); if (::connect(fd, (sockaddr*)srv, sizeof(srv)) 0) { perror(connect); return -1; } printf(connected to %s:%d\n, ip, port); char line[1024]; while (fgets(line, sizeof(line), stdin)) { size_t len strlen(line); if (::send(fd, line, len, 0) ! (ssize_t)len) { perror(send); break; } char buf[1024] {0}; ssize_t n ::recv(fd, buf, sizeof(buf) - 1, 0); if (n 0) { printf(server closed\n); break; } printf(echo: %s, buf); } ::close(fd); return 0; }3.3、UDP收发// udp_server.cpp —— 注意没有 listen/accept也不存在连接 #include sys/socket.h #include netinet/in.h #include arpa/inet.h #include unistd.h #include cstdio int main() { int fd ::socket(AF_INET, SOCK_DGRAM, 0); // SOCK_DGRAM 是关键区别 sockaddr_in addr{}; addr.sin_family AF_INET; addr.sin_addr.s_addr htonl(INADDR_ANY); addr.sin_port htons(9999); ::bind(fd, (sockaddr*)addr, sizeof(addr)); printf(udp server on 9999\n); char buf[1500]; while (true) { sockaddr_in peer{}; socklen_t len sizeof(peer); // recvfrom 一次恰好读一个数据报天然保留消息边界 ssize_t n ::recvfrom(fd, buf, sizeof(buf), 0, (sockaddr*)peer, len); if (n 0) continue; char ip[INET_ADDRSTRLEN]; inet_ntop(AF_INET, peer.sin_addr, ip, sizeof(ip)); printf(recv %zd bytes from %s:%d\n, n, ip, ntohs(peer.sin_port)); ::sendto(fd, buf, n, 0, (sockaddr*)peer, len); } }字节序提醒网络传输统一使用大端序Big-Endian而 x86/ARM 通常是小端。所有多字节整数端口、长度字段都必须用 htons/htonl/ntohs/ntohl 转换。自定义二进制协议时忘记转字节序是跨平台通信最常见的低级 BUG。四、如何判断一个请求包收完了粘包与拆包这是 TCP 编程绕不开的问题也是面试必问。4.1 现象4.2 成因• 发送方Nagle 算法把小包合并发送缓冲区未满时数据积压。• 接收方应用层读取速度慢于到达速度缓冲区积压多个包。• 协议本身超过 MSS 的数据必然被 TCP 分段。根本原因只有一个TCP 是面向字节流的协议它不保存也不理解应用层的消息边界。 所以解决方案必须在应用层。4.3 三种解决方案方案一定长包。简单粗暴浪费带宽只适用于固定格式的控制指令。方案二分隔符。HTTP 头部用 \r\n\r\n、Redis RESP 用 \r\n。缺点是消息体中出现分隔符需要转义且解析需要逐字节扫描。方案三长度字段TLV最通用。头部固定长度其中包含 body 长度。这是绝大多数二进制私有协议、gRPC、Protobuf 帧、RTMP 的做法。Magic Number 用于快速校验和重同步——一旦解析错位可以扫描寻找下一个 Magic 恢复。4.4 完整的解包状态机实现关键思路每个连接维护一个独立的接收缓冲区recv() 到的数据先追加进去然后循环尝试从缓冲区中切出完整的包切不出来就等下次数据到达。#include cstdint #include vector #include string #include arpa/inet.h #pragma pack(push, 1) // 关闭内存对齐保证 sizeof(PacketHeader)10 struct PacketHeader { uint32_t magic; // 0xABCD1234 uint8_t version; uint8_t type; uint32_t length; // payload 长度网络字节序 }; #pragma pack(pop) constexpr uint32_t kMagic 0xABCD1234; constexpr uint32_t kMaxBody 16 * 1024 * 1024; // 防御拒绝异常大包避免内存被打爆 class PacketDecoder { public: // 把新到的数据喂进来 void feed(const char* data, size_t len) { buf_.append(data, len); } // 尝试取出一个完整包返回 false 表示数据还不够 bool nextPacket(PacketHeader hdr, std::string body) { while (true) { if (buf_.size() sizeof(PacketHeader)) return false; // ① 包头都不够 const PacketHeader* p reinterpret_castconst PacketHeader*(buf_.data()); uint32_t magic ntohl(p-magic); uint32_t bodyLen ntohl(p-length); if (magic ! kMagic) { // ② 错位丢一个字节重同步 buf_.erase(0, 1); continue; } if (bodyLen kMaxBody) { buf_.clear(); return false; } // ③ 非法长度断开连接更稳妥 size_t total sizeof(PacketHeader) bodyLen; if (buf_.size() total) return false; // ④ 半包等下次 hdr *p; hdr.magic magic; hdr.length bodyLen; body.assign(buf_.data() sizeof(PacketHeader), bodyLen); buf_.erase(0, total); // ⑤ 消费掉这个包 return true; } } private: std::string buf_; // 生产环境建议换成环形缓冲区避免 erase 的 O(n) 拷贝 }; // 使用方式 void onReadable(int fd, PacketDecoder dec) { char tmp[65536]; ssize_t n ::recv(fd, tmp, sizeof(tmp), 0); if (n 0) return; // 0对端关闭 0错误(注意 EAGAIN) dec.feed(tmp, n); PacketHeader hdr; std::string body; while (dec.nextPacket(hdr, body)) { // 关键while 而不是 if一次读可能含多个包 handleBusiness(hdr, body); } }几个工程要点1. 必须用 while 循环取包因为一次 recv 可能含 N 个完整包。2. 必须校验 length 的合法性否则攻击者发一个 length 0xFFFFFFFF 就能让你 OOM。3. 生产环境用环形缓冲区RingBuffer或 muduo 那种 readIndex/writeIndex 双游标 Buffer避免 erase(0, n) 造成的频繁内存搬移。4. HTTP 是分隔符 长度字段的混合体先找 \r\n\r\n 定位头部结束再读 Content-Length 决定 body 长度如果是 Transfer-Encoding: chunked则按块解析直到 0\r\n\r\n。五、IO 模型演进从阻塞到 epoll Reactor5.1 五种 IO 模型前四种都属于同步 IO因为真正的数据拷贝内核缓冲区 → 用户缓冲区阶段仍然是阻塞的。5.2 select / poll / epoll 对比epoll 高效的核心在于把注册和等待分离了epoll_ctl 一次性把 fd 挂到内核红黑树上网卡中断触发时内核通过回调把就绪 fd 塞进就绪链表epoll_wait 只需要看链表空不空。所以它的性能不随监听 fd 总数增长而下降只和活跃连接数有关。这正是 C10K/C1000K 问题的答案。5.3 LT 与 ET• LT水平触发默认只要缓冲区还有数据可读每次 epoll_wait 都会返回该 fd。编程简单、不容易丢事件可以一次只读一部分。• ET边缘触发只在状态发生变化有新数据到达时通知一次。必须一次性把数据读干净循环 read 直到返回 EAGAIN否则剩余数据将永远收不到通知。ET 必须配合非阻塞 fd否则最后一次 read 会阻塞住整个事件循环。5.4 epoll 非阻塞 Reactor 骨架// epoll_server.cpp g epoll_server.cpp -o eserver #include sys/epoll.h #include sys/socket.h #include netinet/in.h #include fcntl.h #include unistd.h #include cstdio #include cstring #include unordered_map static void setNonBlocking(int fd) { int flags fcntl(fd, F_GETFL, 0); fcntl(fd, F_SETFL, flags | O_NONBLOCK); } int main() { int listenFd socket(AF_INET, SOCK_STREAM, 0); int on 1; setsockopt(listenFd, SOL_SOCKET, SO_REUSEADDR, on, sizeof(on)); sockaddr_in addr{}; addr.sin_family AF_INET; addr.sin_addr.s_addr htonl(INADDR_ANY); addr.sin_port htons(8888); bind(listenFd, (sockaddr*)addr, sizeof(addr)); listen(listenFd, 1024); setNonBlocking(listenFd); int epfd epoll_create1(EPOLL_CLOEXEC); epoll_event ev{}; ev.events EPOLLIN; // 监听 fd 用 LT避免惊群下漏 accept ev.data.fd listenFd; epoll_ctl(epfd, EPOLL_CTL_ADD, listenFd, ev); constexpr int kMaxEvents 1024; epoll_event events[kMaxEvents]; while (true) { int n epoll_wait(epfd, events, kMaxEvents, -1); if (n 0) { if (errno EINTR) continue; perror(epoll_wait); break; } for (int i 0; i n; i) { int fd events[i].data.fd; if (fd listenFd) { // 循环 accept把已完成队列一次取空 while (true) { int connFd accept4(listenFd, nullptr, nullptr, SOCK_NONBLOCK); if (connFd 0) { if (errno EAGAIN || errno EWOULDBLOCK) break; // 取完了 if (errno EINTR) continue; break; } epoll_event cev{}; cev.events EPOLLIN | EPOLLET; // 连接 fd 用 ET减少唤醒次数 cev.data.fd connFd; epoll_ctl(epfd, EPOLL_CTL_ADD, connFd, cev); } continue; } if (events[i].events (EPOLLHUP | EPOLLERR)) { epoll_ctl(epfd, EPOLL_CTL_DEL, fd, nullptr); close(fd); continue; } if (events[i].events EPOLLIN) { bool closed false; char buf[65536]; // ET 模式必须读到 EAGAIN 为止 while (true) { ssize_t r read(fd, buf, sizeof(buf)); if (r 0) { write(fd, buf, r); // 生产环境应写入输出缓冲并关注 EPOLLOUT } else if (r 0) { closed true; break; // 对端关闭 } else { if (errno EINTR) continue; if (errno EAGAIN || errno EWOULDBLOCK) break; // 读干净了 closed true; break; } } if (closed) { epoll_ctl(epfd, EPOLL_CTL_DEL, fd, nullptr); close(fd); } } } } close(listenFd); return 0; }5.5 Reactor 模式与主从多线程模型上面的单线程 epoll 就是最简单的 Reactor。工业级方案Netty、muduo、nginx通常采用 主从 Reactor 多线程核心原则是 one loop per thread每个连接固定归属一个 SubReactor 线程处理读写和状态变更都在同一线程内完成从而天然避免了对连接状态加锁。这是高性能网络库最重要的设计决策也是用架构消除锁竞争的经典案例。