二、Socket 编程 TCP

Socket 编程 TCP一、TCP 编程整体认识TCP 是面向连接的可靠传输协议。和 UDP 不同UDP 可以直接 sendto/recvfrom 收发数据而 TCP 通信之前必须先建立连接。TCP 服务端基本流程socket() - bind() - listen() - accept() - read/write - close()TCP 客户端基本流程socket() - connect() - write/read - close()服务端中有两个非常重要的 socket_listenSock监听 socket只负责获取新连接 sockfd通信 socket由 accept 返回负责和某个客户端通信。accept() 是 TCP 服务端非常关键的接口。监听 socket 并不直接和客户端通信它只是负责等待连接。真正与客户端通信的是 accept() 返回的新 socket。客户端一般不需要显式 bind()因为客户端不需要固定端口。当客户端调用 connect() 时操作系统会自动为客户端分配本地 IP 和临时端口。二、TCP 常用接口说明1. socketint socket(int domain, int type, int protocol);TCP 使用socket(AF_INET, SOCK_STREAM, 0);含义AF_INETIPv4 SOCK_STREAM面向字节流也就是 TCP 0使用默认协议。2. bindint bind(int sockfd, const struct sockaddr *addr, socklen_t addrlen);服务器需要绑定固定 IP 和端口。否则客户端不知道连接哪里。常见写法local.sin_family AF_INET; local.sin_port htons(port); local.sin_addr.s_addr htonl(INADDR_ANY);INADDR_ANY 表示绑定本机任意 IP。3. listenint listen(int sockfd, int backlog);listen() 把 socket 设置为监听状态。只有调用 listen() 后服务器才可以接收客户端连接。4. acceptint accept(int sockfd, struct sockaddr *addr, socklen_t *addrlen);accept() 从监听队列中获取一个已经建立好的连接并返回新的通信 socket。5. connectint connect(int sockfd, const struct sockaddr *addr, socklen_t addrlen);客户端通过 connect() 连接服务器。connect() 填的是服务器地址不是自己的地址。6. read/writeTCP 连接建立后可以像读写文件一样读写 socketread(sockfd, buffer, sizeof(buffer)); write(sockfd, data.c_str(), data.size());但是 TCP 是面向字节流的不保证一次 write() 对应一次完整 read()。这就是后面要解决的粘包问题。三、服务器版本演进V1单进程版本主进程 accept() 一个连接后直接调用 Service() 处理客户端。优点是简单适合理解 TCP 基本流程。缺点是一次只能服务一个客户端。如果当前客户端不退出服务器就无法继续处理其他客户端。V2多进程版本父进程只负责 accept()每来一个客户端就创建子进程处理。优点是多个客户端可以并发处理。缺点是进程创建成本较高并且需要处理子进程回收问题否则会产生僵尸进程。V3多线程版本主线程只负责 accept()每来一个客户端就创建一个线程处理。优点是比多进程更轻量。缺点是客户端太多时线程数量会快速增加服务器压力变大。V4线程池版本提前创建固定数量的工作线程主线程负责接收连接然后把任务投递到线程池。优点是避免频繁创建和销毁线程也能控制并发数量。缺点是如果每个连接都是长连接一个连接会长期占用一个线程。更高并发场景还需要继续学习 select/poll/epoll。公共代码下面这些文件四个版本都可以共用。Comm.hpp#pragma once #include sys/types.h #include sys/socket.h enum { Usage_Err 1, Socket_Err, Bind_Err, Listen_Err, Connect_Err }; // 把 sockaddr_in* 转成 sockaddr* #define CONV(addr_ptr) ((struct sockaddr *)(addr_ptr))nocopy.hpp#pragma once class nocopy { public: nocopy() default; ~nocopy() default; nocopy(const nocopy ) delete; const nocopy operator(const nocopy ) delete; };Log.hpp#pragma once #include cstdio #include ctime #include cstdarg enum LogLevel { Debug 0, Info, Warning, Error, Fatal }; class Log { public: void LogMessage(LogLevel level, const char *format, ...) { const char *levelString[] { Debug, Info, Warning, Error, Fatal}; char timeBuffer[64]; time_t curr time(nullptr); struct tm *tm localtime(curr); snprintf(timeBuffer, sizeof(timeBuffer), %04d-%02d-%02d %02d:%02d:%02d, tm-tm_year 1900, tm-tm_mon 1, tm-tm_mday, tm-tm_hour, tm-tm_min, tm-tm_sec); printf([%s][%s] , timeBuffer, levelString[level]); va_list args; va_start(args, format); vprintf(format, args); va_end(args); } }; static Log lg;InetAddr.hpp#pragma once #include string #include cstdint #include netinet/in.h class InetAddr { public: InetAddr(); explicit InetAddr(const struct sockaddr_in addr); std::string Ip() const; uint16_t Port() const; std::string ToString() const; const struct sockaddr_in Addr() const; private: std::string _ip; uint16_t _port; struct sockaddr_in _addr; };InetAddr.cc#include InetAddr.hpp #include cstring #include arpa/inet.h InetAddr::InetAddr() : _ip(0.0.0.0), _port(0) { memset(_addr, 0, sizeof(_addr)); } InetAddr::InetAddr(const struct sockaddr_in addr) : _addr(addr) { char ipBuffer[64]; // inet_ntop 是线程安全版本比 inet_ntoa 更推荐 inet_ntop(AF_INET, _addr.sin_addr, ipBuffer, sizeof(ipBuffer)); _ip ipBuffer; _port ntohs(_addr.sin_port); } std::string InetAddr::Ip() const { return _ip; } uint16_t InetAddr::Port() const { return _port; } std::string InetAddr::ToString() const { return _ip : std::to_string(_port); } const struct sockaddr_in InetAddr::Addr() const { return _addr; }V1 单进程 Echo ServerTcpServer.hpp#pragma once #include cstdint #include nocopy.hpp #include InetAddr.hpp const static int defaultBacklog 6; class TcpServer : public nocopy { public: explicit TcpServer(uint16_t port); ~TcpServer(); void Init(); void Start(); private: void Service(int sockfd, InetAddr peer); private: uint16_t _port; int _listenSock; bool _isRunning; };TcpServer.cc#include TcpServer.hpp #include iostream #include cstring #include cerrno #include cstdlib #include unistd.h #include signal.h #include sys/socket.h #include netinet/in.h #include Comm.hpp #include Log.hpp TcpServer::TcpServer(uint16_t port) : _port(port), _listenSock(-1), _isRunning(false) { } void TcpServer::Init() { // 防止向已经关闭的连接写入时进程被 SIGPIPE 杀掉 signal(SIGPIPE, SIG_IGN); // 1. 创建监听 socket _listenSock socket(AF_INET, SOCK_STREAM, 0); if (_listenSock 0) { lg.LogMessage(Fatal, socket error, errno: %d, %s\n, errno, strerror(errno)); exit(Socket_Err); } // 2. 设置端口复用方便服务器重启 int opt 1; setsockopt(_listenSock, SOL_SOCKET, SO_REUSEADDR, opt, sizeof(opt)); // 3. 填写服务器本地地址 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); // 4. 绑定 IP 和端口 if (bind(_listenSock, CONV(local), sizeof(local)) 0) { lg.LogMessage(Fatal, bind error, errno: %d, %s\n, errno, strerror(errno)); exit(Bind_Err); } // 5. 设置监听状态 if (listen(_listenSock, defaultBacklog) 0) { lg.LogMessage(Fatal, listen error, errno: %d, %s\n, errno, strerror(errno)); exit(Listen_Err); } lg.LogMessage(Info, server init success, port: %d\n, _port); } void TcpServer::Start() { _isRunning true; while (_isRunning) { struct sockaddr_in peer; socklen_t len sizeof(peer); // accept 返回的是通信 socket int sockfd accept(_listenSock, CONV(peer), len); if (sockfd 0) { lg.LogMessage(Warning, accept error, errno: %d, %s\n, errno, strerror(errno)); continue; } InetAddr addr(peer); lg.LogMessage(Info, new connection: %s, sockfd: %d\n, addr.ToString().c_str(), sockfd); // V1直接在主进程中提供服务 Service(sockfd, addr); close(sockfd); } } void TcpServer::Service(int sockfd, InetAddr peer) { char buffer[1024]; while (true) { ssize_t n read(sockfd, buffer, sizeof(buffer) - 1); if (n 0) { buffer[n] \0; std::cout client[ peer.ToString() ] say# buffer std::endl; std::string echo server echo# ; echo buffer; write(sockfd, echo.c_str(), echo.size()); } else if (n 0) { lg.LogMessage(Info, client quit: %s\n, peer.ToString().c_str()); break; } else { lg.LogMessage(Error, read error, errno: %d, %s\n, errno, strerror(errno)); break; } } } TcpServer::~TcpServer() { if (_listenSock 0) { close(_listenSock); } }V2 多进程 Echo ServerV2 的核心变化accept() 新连接后创建子进程处理客户端。TcpServer.hpp#pragma once #include cstdint #include nocopy.hpp #include InetAddr.hpp const static int defaultBacklog 6; class TcpServer : public nocopy { public: explicit TcpServer(uint16_t port); ~TcpServer(); void Init(); void Start(); private: void Service(int sockfd, InetAddr peer); void ProcessConnection(int sockfd, const struct sockaddr_in peer); private: uint16_t _port; int _listenSock; bool _isRunning; };TcpServer.cc#include TcpServer.hpp #include iostream #include cstring #include cerrno #include cstdlib #include unistd.h #include signal.h #include sys/socket.h #include sys/wait.h #include netinet/in.h #include Comm.hpp #include Log.hpp TcpServer::TcpServer(uint16_t port) : _port(port), _listenSock(-1), _isRunning(false) { } void TcpServer::Init() { signal(SIGPIPE, SIG_IGN); _listenSock socket(AF_INET, SOCK_STREAM, 0); if (_listenSock 0) { lg.LogMessage(Fatal, socket error, errno: %d, %s\n, errno, strerror(errno)); exit(Socket_Err); } int opt 1; setsockopt(_listenSock, SOL_SOCKET, SO_REUSEADDR, opt, sizeof(opt)); 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); if (bind(_listenSock, CONV(local), sizeof(local)) 0) { lg.LogMessage(Fatal, bind error, errno: %d, %s\n, errno, strerror(errno)); exit(Bind_Err); } if (listen(_listenSock, defaultBacklog) 0) { lg.LogMessage(Fatal, listen error, errno: %d, %s\n, errno, strerror(errno)); exit(Listen_Err); } lg.LogMessage(Info, server init success, port: %d\n, _port); } void TcpServer::Start() { _isRunning true; while (_isRunning) { struct sockaddr_in peer; socklen_t len sizeof(peer); int sockfd accept(_listenSock, CONV(peer), len); if (sockfd 0) { lg.LogMessage(Warning, accept error, errno: %d, %s\n, errno, strerror(errno)); continue; } ProcessConnection(sockfd, peer); } } void TcpServer::ProcessConnection(int sockfd, const struct sockaddr_in peer) { InetAddr addr(peer); pid_t id fork(); if (id 0) { lg.LogMessage(Error, fork error, errno: %d, %s\n, errno, strerror(errno)); close(sockfd); return; } else if (id 0) { // 子进程不需要监听 socket close(_listenSock); Service(sockfd, addr); close(sockfd); exit(0); } else { // 父进程不负责通信关闭自己的通信 socket close(sockfd); // 非阻塞回收避免僵尸进程堆积 while (waitpid(-1, nullptr, WNOHANG) 0) { } } } void TcpServer::Service(int sockfd, InetAddr peer) { char buffer[1024]; while (true) { ssize_t n read(sockfd, buffer, sizeof(buffer) - 1); if (n 0) { buffer[n] \0; std::cout client[ peer.ToString() ] say# buffer std::endl; std::string echo server echo# ; echo buffer; write(sockfd, echo.c_str(), echo.size()); } else if (n 0) { lg.LogMessage(Info, client quit: %s\n, peer.ToString().c_str()); break; } else { lg.LogMessage(Error, read error, errno: %d, %s\n, errno, strerror(errno)); break; } } } TcpServer::~TcpServer() { if (_listenSock 0) { close(_listenSock); } }V3 多线程 Echo ServerV3 的核心变化每个客户端连接交给一个线程处理。TcpServer.hpp#pragma once #include cstdint #include pthread.h #include nocopy.hpp #include InetAddr.hpp const static int defaultBacklog 6; class TcpServer : public nocopy { public: explicit TcpServer(uint16_t port); ~TcpServer(); void Init(); void Start(); private: class ThreadData { public: ThreadData(TcpServer *server, int sockfd, const struct sockaddr_in peer); public: TcpServer *_server; int _sockfd; InetAddr _peer; }; private: static void *ThreadRoutine(void *args); void Service(int sockfd, InetAddr peer); void ProcessConnection(int sockfd, const struct sockaddr_in peer); private: uint16_t _port; int _listenSock; bool _isRunning; };TcpServer.cc#include TcpServer.hpp #include iostream #include cstring #include cerrno #include cstdlib #include unistd.h #include signal.h #include sys/socket.h #include netinet/in.h #include Comm.hpp #include Log.hpp TcpServer::ThreadData::ThreadData(TcpServer *server, int sockfd, const struct sockaddr_in peer) : _server(server), _sockfd(sockfd), _peer(peer) { } TcpServer::TcpServer(uint16_t port) : _port(port), _listenSock(-1), _isRunning(false) { } void TcpServer::Init() { signal(SIGPIPE, SIG_IGN); _listenSock socket(AF_INET, SOCK_STREAM, 0); if (_listenSock 0) { lg.LogMessage(Fatal, socket error, errno: %d, %s\n, errno, strerror(errno)); exit(Socket_Err); } int opt 1; setsockopt(_listenSock, SOL_SOCKET, SO_REUSEADDR, opt, sizeof(opt)); 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); if (bind(_listenSock, CONV(local), sizeof(local)) 0) { lg.LogMessage(Fatal, bind error, errno: %d, %s\n, errno, strerror(errno)); exit(Bind_Err); } if (listen(_listenSock, defaultBacklog) 0) { lg.LogMessage(Fatal, listen error, errno: %d, %s\n, errno, strerror(errno)); exit(Listen_Err); } lg.LogMessage(Info, server init success, port: %d\n, _port); } void TcpServer::Start() { _isRunning true; while (_isRunning) { struct sockaddr_in peer; socklen_t len sizeof(peer); int sockfd accept(_listenSock, CONV(peer), len); if (sockfd 0) { lg.LogMessage(Warning, accept error, errno: %d, %s\n, errno, strerror(errno)); continue; } ProcessConnection(sockfd, peer); } } void TcpServer::ProcessConnection(int sockfd, const struct sockaddr_in peer) { pthread_t tid; ThreadData *td new ThreadData(this, sockfd, peer); int n pthread_create(tid, nullptr, ThreadRoutine, td); if (n ! 0) { lg.LogMessage(Error, pthread_create error\n); close(sockfd); delete td; } } void *TcpServer::ThreadRoutine(void *args) { pthread_detach(pthread_self()); ThreadData *td static_castThreadData *(args); td-_server-Service(td-_sockfd, td-_peer); close(td-_sockfd); delete td; return nullptr; } void TcpServer::Service(int sockfd, InetAddr peer) { char buffer[1024]; while (true) { ssize_t n read(sockfd, buffer, sizeof(buffer) - 1); if (n 0) { buffer[n] \0; std::cout client[ peer.ToString() ] say# buffer std::endl; std::string echo server echo# ; echo buffer; write(sockfd, echo.c_str(), echo.size()); } else if (n 0) { lg.LogMessage(Info, client quit: %s\n, peer.ToString().c_str()); break; } else { lg.LogMessage(Error, read error, errno: %d, %s\n, errno, strerror(errno)); break; } } } TcpServer::~TcpServer() { if (_listenSock 0) { close(_listenSock); } }V4 线程池 Echo ServerThreadPool.hpp#pragma once #include queue #include mutex #include thread #include vector #include condition_variable template class Task class ThreadPool { public: static ThreadPoolTask *GetInstance() { static ThreadPoolTask instance; return instance; } void Start() { std::lock_guardstd::mutex lock(_startMutex); if (_isRunning) { return; } _isRunning true; for (int i 0; i _threadNum; i) { _threads.emplace_back([this]() { this-ThreadRun(); }); } for (auto thread : _threads) { thread.detach(); } } void Push(const Task task) { { std::lock_guardstd::mutex lock(_mutex); _tasks.push(task); } _cond.notify_one(); } private: ThreadPool(int threadNum 5) : _threadNum(threadNum), _isRunning(false) { } ThreadPool(const ThreadPool ) delete; ThreadPool operator(const ThreadPool ) delete; void ThreadRun() { while (true) { Task task; { std::unique_lockstd::mutex lock(_mutex); _cond.wait(lock, [this]() { return !_tasks.empty(); }); task _tasks.front(); _tasks.pop(); } task(); } } private: int _threadNum; bool _isRunning; std::vectorstd::thread _threads; std::queueTask _tasks; std::mutex _mutex; std::mutex _startMutex; std::condition_variable _cond; };TcpServer.hpp#pragma once #include cstdint #include nocopy.hpp #include InetAddr.hpp const static int defaultBacklog 6; class TcpServer : public nocopy { public: explicit TcpServer(uint16_t port); ~TcpServer(); void Init(); void Start(); private: void Service(int sockfd, InetAddr peer); void ProcessConnection(int sockfd, const struct sockaddr_in peer); private: uint16_t _port; int _listenSock; bool _isRunning; };TcpServer.cc#include TcpServer.hpp #include iostream #include cstring #include cerrno #include cstdlib #include unistd.h #include signal.h #include functional #include sys/socket.h #include netinet/in.h #include Comm.hpp #include Log.hpp #include ThreadPool.hpp TcpServer::TcpServer(uint16_t port) : _port(port), _listenSock(-1), _isRunning(false) { } void TcpServer::Init() { signal(SIGPIPE, SIG_IGN); _listenSock socket(AF_INET, SOCK_STREAM, 0); if (_listenSock 0) { lg.LogMessage(Fatal, socket error, errno: %d, %s\n, errno, strerror(errno)); exit(Socket_Err); } int opt 1; setsockopt(_listenSock, SOL_SOCKET, SO_REUSEADDR, opt, sizeof(opt)); 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); if (bind(_listenSock, CONV(local), sizeof(local)) 0) { lg.LogMessage(Fatal, bind error, errno: %d, %s\n, errno, strerror(errno)); exit(Bind_Err); } if (listen(_listenSock, defaultBacklog) 0) { lg.LogMessage(Fatal, listen error, errno: %d, %s\n, errno, strerror(errno)); exit(Listen_Err); } using task_t std::functionvoid(); ThreadPooltask_t::GetInstance()-Start(); lg.LogMessage(Info, server init success, port: %d\n, _port); } void TcpServer::Start() { _isRunning true; while (_isRunning) { struct sockaddr_in peer; socklen_t len sizeof(peer); int sockfd accept(_listenSock, CONV(peer), len); if (sockfd 0) { lg.LogMessage(Warning, accept error, errno: %d, %s\n, errno, strerror(errno)); continue; } ProcessConnection(sockfd, peer); } } void TcpServer::ProcessConnection(int sockfd, const struct sockaddr_in peer) { using task_t std::functionvoid(); InetAddr addr(peer); // 把连接封装成任务交给线程池处理 task_t task std::bind(TcpServer::Service, this, sockfd, addr); ThreadPooltask_t::GetInstance()-Push(task); } void TcpServer::Service(int sockfd, InetAddr peer) { char buffer[1024]; while (true) { ssize_t n read(sockfd, buffer, sizeof(buffer) - 1); if (n 0) { buffer[n] \0; std::cout client[ peer.ToString() ] say# buffer std::endl; std::string echo server echo# ; echo buffer; write(sockfd, echo.c_str(), echo.size()); } else if (n 0) { lg.LogMessage(Info, client quit: %s\n, peer.ToString().c_str()); break; } else { lg.LogMessage(Error, read error, errno: %d, %s\n, errno, strerror(errno)); break; } } close(sockfd); } TcpServer::~TcpServer() { if (_listenSock 0) { close(_listenSock); } }客户端源码四个版本都可以使用这个客户端测试。TcpClient.cc#include iostream #include string #include cstring #include cstdlib #include unistd.h #include sys/socket.h #include netinet/in.h #include arpa/inet.h #include Comm.hpp static void Usage(const std::string process) { std::cout Usage: process server_ip server_port std::endl; } int main(int argc, char *argv[]) { if (argc ! 3) { Usage(argv[0]); return Usage_Err; } std::string serverIp argv[1]; uint16_t serverPort std::stoi(argv[2]); // 1. 创建 socket int sockfd socket(AF_INET, SOCK_STREAM, 0); if (sockfd 0) { std::cerr socket error std::endl; return Socket_Err; } // 2. 填写服务器地址 struct sockaddr_in server; memset(server, 0, sizeof(server)); server.sin_family AF_INET; server.sin_port htons(serverPort); if (inet_pton(AF_INET, serverIp.c_str(), server.sin_addr) 0) { std::cerr inet_pton error std::endl; close(sockfd); return Connect_Err; } // 3. 连接服务器 // 客户端通常不需要显式 bindconnect 时系统会自动绑定本地端口 if (connect(sockfd, CONV(server), sizeof(server)) 0) { std::cerr connect error std::endl; close(sockfd); return Connect_Err; } std::cout connect server success std::endl; // 4. 通信 while (true) { std::string line; std::cout Please Enter# ; std::getline(std::cin, line); if (line quit || line exit) { break; } ssize_t n write(sockfd, line.c_str(), line.size()); if (n 0) { std::cerr write error std::endl; break; } char buffer[1024]; ssize_t m read(sockfd, buffer, sizeof(buffer) - 1); if (m 0) { buffer[m] \0; std::cout buffer std::endl; } else if (m 0) { std::cout server close connection std::endl; break; } else { std::cerr read error std::endl; break; } } close(sockfd); return 0; }服务端启动入口ServerMain.cc#include iostream #include cstdlib #include TcpServer.hpp #include Comm.hpp static void Usage(const std::string process) { std::cout Usage: process port std::endl; } int main(int argc, char *argv[]) { if (argc ! 2) { Usage(argv[0]); return Usage_Err; } uint16_t port std::stoi(argv[1]); TcpServer server(port); server.Init(); server.Start(); return 0; }Makefile.PHONY: all clean CXXg CXXFLAGS-stdc17 -Wall -Wextra -pthread all: tcp_server tcp_client tcp_server: ServerMain.cc TcpServer.cc InetAddr.cc $(CXX) $(CXXFLAGS) -o $ $^ tcp_client: TcpClient.cc $(CXX) $(CXXFLAGS) -o $ $^ clean: rm -f tcp_server tcp_client运行测试编译make启动服务端./tcp_server 8080启动客户端./tcp_client 127.0.0.1 8080客户端输入hello服务器返回server echo# hello输入quit客户端退出。最后总结TCP 编程的主线是连接。服务端先创建监听 socket然后绑定端口接着调用 listen() 进入监听状态。之后通过 accept() 获取客户端连接。accept() 返回的新 socket 才是真正用于通信的 socket。单进程版本适合理解流程多进程版本可以支持并发多线程版本比多进程更轻量线程池版本进一步减少线程频繁创建销毁的开销。不过这些版本目前都还是简单 Echo Server还没有解决 TCP 粘包问题。因为 TCP 是字节流协议不保留消息边界。真正写业务服务器时还需要设计应用层协议例如报文长度 报文内容后面继续学习协议设计、序列化反序列化、线程池任务处理、IO 多路复用时TCP 服务器的结构就会越来越完整