目录
服务端
创建套接字(socket)
服务端绑定(bind)
服务端监听(listen)
服务器接收(accept)
服务端处理(read & write)
客户端
创建套接字(socket)
客户端连接(connect)
客户端处理(send & recv)
多进程代码测试
捕获忽略SIGCHLD信号
子进程创建子进程
多线程代码测试
线程池代码测试
服务端
创建套接字(socket)
我们继续将TCP服务器封装成一个类,TCP创建套接字只有第二个参数不同,所需要的服务类型是SOCK_STREAM,这就是一个可靠的、需要链接的、全双工的、面向字节流的协议。
class TcpServer { public: TcpServer(uint16_t port, std::string ip = "") :_port(port) ,_ip(ip) ,_sock(-1) {} void initServer() { // 创建socket _sock = socket(AF_INET, SOCK_STREAM, 0); if (_sock < 0) { logMessage(FATAL, "%d:%s", errno, strerror(errno)); exit(2); } logMessage(NORMAL, "sock: %d", _sock); } ~TcpServer() { if (_sock > 0) close(_sock); } private: uint16_t _port; std::string _ip; int _sock; };
服务端绑定(bind)
绑定的操作和UDP是差不多的,想要详细了解可以翻看上一篇。
// 创建套接字 // ... // 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 = _ip.empty() ? INADDR_ANY : inet_addr(_ip.c_str()); if (bind(_sock, (struct sockaddr*)&local, sizeof(local)) < 0) { logMessage(FATAL, "bind error, %d:%s", errno, strerror(errno)); exit(3); }
服务端监听(listen)
至此,创建套接字和绑定与UDP没有太大的差别,而TCP是面向连接的,客户端发送数据要先与TCP建立连接,然后才能通信。
所以TCP要注意,随时都有可能从客户端发来连接请求,此时就需要将TCP服务器创建的套接字设置为监听状态,用到的函数就是listen。
参数:
- sockfd:需要设置为监听状态的套接字文件描述符
- backlog:全连接队列的最大长度,具体是什么后面再说,一般设置为5、10或20就可以
返回值:监听成功返回0,失败返回-1,错误码被设置。
至此,我的服务器初始化任务已经完成了。
const static int g_backlog = 20; void initServer() { // 创建socket _sock = socket(AF_INET, SOCK_STREAM, 0); if (_sock < 0) { logMessage(FATAL, "%d:%s", errno, strerror(errno)); exit(2); } logMessage(NORMAL, "sock: %d", _sock); // 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 = _ip.empty() ? INADDR_ANY : inet_addr(_ip.c_str()); if (bind(_sock, (struct sockaddr*)&local, sizeof(local)) < 0) { logMessage(FATAL, "bind error, %d:%s", errno, strerror(errno)); exit(3); } logMessage(NORMAL, "bind success ..."); // tcp面向连接的,通信前要先建立连接,设置为监听状态,listen if (listen(_sock, g_backlog) < 0) { logMessage(FATAL, "listen error, %d:%s", errno, strerror(errno)); exit(4); } logMessage(NORMAL, "listen success ..."); logMessage(NORMAL, "init server success"); }
服务器接收(accept)
TCP服务器在与客户端进行通信之前,服务器需要先获取到客户端的连接请求,获取的函数就是accept。
参数:
- sockfd:从该套接字中获取连接
- addr:输出型参数,获取对端的网络属性信息
- addrlen:输入输出型参数,返回实际读到的addr结构体的长度、
返回值:获取连接成功的对端套接字的文件描述符,失败返回-1,错误吗被设置。
accept返回值 vs sockfd:
- sockfd是我们绑定的监听套接字,它只负责帮我们把获取的连接拿过来。
- 而真正执行IO的就是accept返回的对端的套接字。
所以我们上述的_sock也叫做监听套接字listensock(_sock->listensock)。
void start() { while (true) { // 获取连接accept struct sockaddr_in src; // 客户端addr socklen_t len = sizeof(src); int servicesock = accept(listensock, (struct sockaddr *)&src, &len); if (servicesock < 0) { logMessage(ERROR, "accept error, %d:%s", errno, strerror(errno)); // 接收失败也不影响,继续接收 continue; } // 获取连接成功 uint16_t client_port = ntohs(src.sin_port); std::string client_ip = inet_ntoa(src.sin_addr); logMessage(NORMAL, "link success, servicesock: %d | %s : %d|\n", servicesock, client_ip.c_str(), client_port); // 处理任务 close(servicesock); } }
服务端处理(read & write)
static void service(int sock, const std::string& clientip, const uint16_t& clientport) { // echo char buffer[1024]; while (true) { // read和write可以直接使用 // 读取数据 ssize_t s = read(sock, buffer, sizeof(buffer) - 1); if (s > 0) { buffer[s] = 0; // 将发过来的消息当做字符串 std::cout << clientip << " : " << clientport << "# " << buffer << std::endl; } else if (s == 0) { // 代表对端关闭连接 logMessage(NORMAL, "%s:%d shutdown , me too", clientip.c_str(), clientport); break; // 跳出循环结束通信 } else { logMessage(ERROR, "read sock error, %d:%s", errno, strerror(errno)); break; } // 写回数据 write(sock, buffer, sizeof(buffer)); } } // 获取连接成功 // ... service(servicesock, client_ip, client_port); close(servicesock);
直接使用read和write就可以进行IO了,但是就有一个问题,当一个连接被接收到了,这个函数就是一个死循环,对端不退出会一直执行,那么如果此时再来一个连接那就会被阻塞。
那我们就可以使用多进程,子进程负责处理接收的文件描述符,父进程负责接收描述符。
void start() { signal(SIGCHLD, SIG_IGN); // 子进程退出向父进程发送SIGCHLD,直接忽略这个信号,子进程退出就自动释放僵尸状态 while (true) { // 获取连接accept // ... // 获取连接成功 // ... pid_t id = fork(); assert(id != -1); if (id == 0) { // 子进程 // 子进程也会继承父进程的文件与文件fd // 子进程提供服务,他不用管监听的事,所以要关闭 close(listensock); service(servicesock, client_ip, client_port); exit(0); } // 父进程 // 父进程只需要接收sock,所以要关闭服务sock close(servicesock); } }
客户端
创建套接字(socket)
初始化客户端,而客户端唯一要做的就是创建套接字,它不需要bind,也不需要listen,因为没有人会去连接客户端,那就更不需要accept接收了,它只需要知道服务端的IP和PORT就可以了。
// 1.创建套接字 int sock = socket(AF_INET, SOCK_STREAM, 0); if (sock < 0) { std::cerr << "socket error" << std::endl; exit(2); }
客户端连接(connect)
向服务端发起连接请求,使用的函数就是connect。
参数:
- sockfd:通过该套接字发起连接请求
- addr:服务端网络属性信息
- addrlen:传入的addr结构体的长度
返回值:连接成功返回0,失败返回-1,错误码被设置。
客户端不需要bind,当客户端向服务端发起连接请求,操作系统会自动给客户端分配端口号。
// 不需要显示绑定bind,OS自动选择port // 2.向服务端发起建立连接的请求connect struct sockaddr_in server; memset(&server, 0, sizeof(server)); server.sin_family = AF_INET; server.sin_port = htons(serverport); server.sin_addr.s_addr = inet_addr(serverip.c_str()); if (connect(sock, (struct sockaddr*)&server, sizeof(server)) < 0) { std::cerr << "connect error" << std::endl; exit(3); }
客户端处理(send & recv)
send和recv是TCP用来通信的接口,但是read和write也可以使用。
作用:发送数据到对端
ssize_t send(int sockfd, const void *buf, size_t len, int flags);
参数:
- sockfd:向创建的套接字写入
- buf:要把哪个缓冲区中的数据写入
- len:缓冲区的长度
- flags:一般默认为0
返回值:成功返回发送的数据个数,失败返回-1,错误码被设置。
作用:接收对端的数据
ssize_t recv(int sockfd, void *buf, size_t len, int flags);
参数:
- sockfd:从哪个套接字中接收数据
- buf:将接收的数据放到的缓冲区
- len:缓冲区的长度
- flags:一般设置为0
返回值:成功返回接收的个数,失败返回-1,错误码被设置。
while (true) { std::string line; std::cout << "请输入: "; std::getline(std::cin, line); send(sock, line.c_str(), line.size(), 0); char buffer[1024]; ssize_t s = recv(sock, buffer, sizeof(buffer) - 1, 0); if (s > 0) { buffer[s] = 0; std::cout << "server echo: " << buffer << std::endl; } else if (s == 0) { // 对端关闭 break; } else { break; } }
多进程代码测试
捕获忽略SIGCHLD信号
// tcp_server.hpp #pragma once #include <iostream> #include <string> #include <cerrno> #include <cstdio> #include <cstring> #include <cassert> #include <signal.h> #include <unistd.h> #include <sys/types.h> #include <sys/socket.h> #include <netinet/in.h> #include <arpa/inet.h> #include "log.hpp" static void service(int sock, const std::string& clientip, const uint16_t& clientport) { // echo char buffer[1024]; while (true) { // read和write可以直接使用 // 读取数据 ssize_t s = read(sock, buffer, sizeof(buffer) - 1); if (s > 0) { buffer[s] = 0; // 将发过来的消息当做字符串 std::cout << clientip << " : " << clientport << "# " << buffer << std::endl; } else if (s == 0) { // 代表对端关闭连接 logMessage(NORMAL, "%s:%d shutdown , me too", clientip.c_str(), clientport); break; // 跳出循环结束通信 } else { logMessage(ERROR, "read sock error, %d:%s", errno, strerror(errno)); break; } // 写回数据 write(sock, buffer, strlen(buffer)); } } class TcpServer { const static int g_backlog = 20; public: TcpServer(uint16_t port, std::string ip = "") : _port(port), _ip(ip), listensock(-1) {} void initServer() { // 创建socket listensock = socket(AF_INET, SOCK_STREAM, 0); if (listensock < 0) { logMessage(FATAL, "%d:%s", errno, strerror(errno)); exit(2); } logMessage(NORMAL, "sock: %d", listensock); // 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 = _ip.empty() ? INADDR_ANY : inet_addr(_ip.c_str()); if (bind(listensock, (struct sockaddr *)&local, sizeof(local)) < 0) { logMessage(FATAL, "bind error, %d:%s", errno, strerror(errno)); exit(3); } logMessage(NORMAL, "bind success ..."); // tcp面向连接的,通信前要先建立连接,设置为监听状态,listen if (listen(listensock, g_backlog) < 0) { logMessage(FATAL, "listen error, %d:%s", errno, strerror(errno)); exit(4); } logMessage(NORMAL, "listen success ..."); logMessage(NORMAL, "init server success"); } void start() { signal(SIGCHLD, SIG_IGN); // 子进程退出向父进程发送SIGCHLD,直接忽略这个信号,子进程退出就自动释放僵尸状态 while (true) { // 获取连接accept struct sockaddr_in src; // 客户端addr socklen_t len = sizeof(src); int servicesock = accept(listensock, (struct sockaddr *)&src, &len); if (servicesock < 0) { logMessage(ERROR, "accept error, %d:%s", errno, strerror(errno)); // 接收失败也不影响,继续接收 continue; } // 获取连接成功 uint16_t client_port = ntohs(src.sin_port); std::string client_ip = inet_ntoa(src.sin_addr); logMessage(NORMAL, "link success, servicesock: %d | %s : %d|\n", servicesock, client_ip.c_str(), client_port); // v1 -- 单进程循环版 // service(servicesock, client_ip, client_port); // v2 -- 多进程版 pid_t id = fork(); assert(id != -1); if (id == 0) { // 子进程 // 子进程也会继承父进程的文件与文件fd // 子进程提供服务,他不用管监听的事,所以要关闭 // 这是不是很像匿名管道呢 close(listensock); service(servicesock, client_ip, client_port); exit(0); } // 父进程 // 父进程只需要接收sock,所以要关闭服务sock close(servicesock); } } ~TcpServer() { if (listensock > 0) close(listensock); } private: uint16_t _port; std::string _ip; int listensock; };
// tcp_server.cc #include "tcp_server.hpp" #include <memory> void usage(std::string proc) { std::cout << "\nUsage: " << proc << " port\n" << std::endl; } // ./tcp_server port int main(int argc, char* argv[]) { if (argc != 2) { usage(argv[0]); exit(1); } uint16_t port = atoi(argv[1]); std::unique_ptr<TcpServer> svr(new TcpServer(port)); svr->initServer(); svr->start(); return 0; }
// tcp_client.cc #include <iostream> #include <unistd.h> #include <string> #include <cstring> #include <sys/types.h> #include <sys/socket.h> #include <netinet/in.h> #include <arpa/inet.h> void usage(std::string proc) { std::cout << "\nUsage: " << proc << " ip port\n" << std::endl; } int main(int argc, char* argv[]) { if (argc != 3) { usage(argv[0]); exit(1); } std::string serverip = argv[1]; uint16_t serverport = atoi(argv[2]); // 1.创建套接字 int sock = socket(AF_INET, SOCK_STREAM, 0); if (sock < 0) { std::cerr << "socket error" << std::endl; exit(2); } // 不需要显示绑定bind,OS自动选择port // 2.向服务端发起建立连接的请求connect struct sockaddr_in server; memset(&server, 0, sizeof(server)); server.sin_family = AF_INET; server.sin_port = htons(serverport); server.sin_addr.s_addr = inet_addr(serverip.c_str()); if (connect(sock, (struct sockaddr*)&server, sizeof(server)) < 0) { std::cerr << "connect error" << std::endl; exit(3); } // 连接成功 std::cout << "connect success" << std::endl; while (true) { std::string line; std::cout << "请输入: "; std::getline(std::cin, line); send(sock, line.c_str(), line.size(), 0); char buffer[1024]; ssize_t s = recv(sock, buffer, sizeof(buffer) - 1, 0); if (s > 0) { buffer[s] = 0; std::cout << "server echo: " << buffer << std::endl; } else if (s == 0) { // 对端关闭 break; } else { break; } } close(sock); return 0; }
子进程创建子进程
其实多进程还有一种写法,我们再来看看这种写法是怎么实现的,这种写法没有用子进程退出向父进程发送SIGCHLD信号的特性。
void start() { while (true) { // 获取连接accept // ... // 获取连接成功 // ... pid_t id = fork(); assert(id != -1); if (id == 0) { // 子进程 close(listensock); if (fork() > 0) exit(0); // 子进程fork创建孙子进程,子进程exit退出了 // 孙子进程就变成了孤儿进程,孤儿进程会被系统领养,执行完后由操作系统释放 service(servicesock, client_ip, client_port); exit(0); } // 父进程 close(servicesock); waitpid(id, nullptr, 0); //阻塞式等待子进程退出 } }
多线程代码测试
创建进程的成本是很高的,创建进程时需要创建该进程对应task_struct、进程地址空间(mm_struct)等数据结构。而创建线程的成本就会会小得多,线程本质是在进程地址空间内运行,创建出来的线程会共享该进程的大部分资源,因此在实现多执行流的服务器时最好采用多线程进行实现。
class ThreadData { public: int sock; uint16_t port; std::string ip; }; class TcpServer { static void *threadRoutine(void *args) { pthread_detach(pthread_self()); // 线程分离,就不用join了 ThreadData *td = (ThreadData *)args; service(td->sock, td->ip, td->port); delete td; return nullptr; } private: void start() { while (true) { // 获取连接accept // ... // 获取连接成功 // ... ThreadData *td = new ThreadData(); td->sock = servicesock; td->ip = client_ip; td->port = client_port; pthread_t tid; // 多线程就不用关闭listensock文件描述符了,线程共享这些文件描述符 pthread_create(&tid, nullptr, threadRoutine, (void *)td); close(servicesock); } } };
再查看一下线程信息。
线程池代码测试
当有新连接到来时,服务端的主线程都会重新为该客户端创建新线程,当服务结束后,将该线程销毁,在短时间内创建与销毁线程,这样做不仅麻烦,而且效率低,创建线程和销毁线程也是有成本的。
所以为了解决这样的问题我们就可以引入线程池,客户端有发来任务,线程就为他服务,服务完后可以让线程休眠,再有任务来时再将线程唤醒。创建的线程也不能太多,要不然CPU的压力也会过大。如果这一批线程都在服务客户,再来一个客户端的连接就不能创建线程,应该让这个新来的连接请求在连接队列中排队,有空闲线程后再获取请求的连接为它提供服务。
当有新的任务到来的时候,就Push到线程池中的任务队列中,多线程就不断检测任务队列中是否有任务,通过封装的任务类,调用仿函数执行相应的操作。
这次我们想要实现一个网络词典的功能。
class TcpServer { public: TcpServer(uint16_t port, std::string ip = "") : _port(port) , _ip(ip) , listensock(-1) , _threadpool_ptr(ThreadPool<Task>::getThreadPool()) // 获取线程池 {} void start() { _threadpool_ptr->run(); while (true) { // 获取连接accept // ... // 获取连接成功 // ... Task t(servicesock, client_ip, client_port, dictOnline); _threadpool_ptr->pushTask(t); } } private: // ... std::unique_ptr<ThreadPool<Task>> _threadpool_ptr; // 线程池指针 };
添加成员变量_threadpool_ptr,用智能指针维护,在构造函数的初始化列表中调用getThreadPool获取线程池,因为线程池用了单例模式。当TCP服务端初始化完成后,调用start函数,首先就要启动线程池,当accept接受到对端的sock文件描述符后就可以制作任务,并push到任务队列中。
// Task.hpp typedef std::function<void (int, const std::string&, const uint16_t&, const std::string&)> func_t; class Task { public: Task(){} Task(int sock, const std::string& ip, const uint16_t port, func_t func) :_sock(sock) ,_ip(ip) ,_port(port) ,_func(func) {} void operator()(const std::string& name) { _func(_sock, _ip, _port, name); } public: int _sock; std::string _ip; uint16_t _port; func_t _func; };
我们想要实现一个从客户端接收一个英文单词,在单词的map中找,找到了就返回,没有找到就告知客户端词库中没有这个单词。
这次我们采用短连接的方式,这个dictOnline函数就是修改了service函数,service函数使用的是死循环,一个线程对应一个客户端,如果这个客户端已经连接上了,它一直死循环不退出地做某件事,那么服务器接收的连接数就被限制了,所以来一个连接就服务一个,服务完之后,双方都断开连接。
static void dictOnline(int sock, const std::string &clientip, const uint16_t &clientport, const std::string &thread_name) { // 网络词典 static unordered_map<std::string, std::string> dict = { {"apple", "苹果"}, {"banana", "香蕉"}, {"process", "进程"}, {"thread", "线程"}}; // 简易字典 char buffer[1024]; // read和write可以直接使用 // 读取数据 ssize_t s = read(sock, buffer, sizeof(buffer) - 1); if (s > 0) { buffer[s] = 0; // 将发过来的消息当做字符串 std::cout << thread_name << " | " << clientip << " : " << clientport << "# " << buffer << std::endl; std::string message; auto iter = dict.find(buffer); if (iter == dict.end()) message = "词库中无此单词"; else message = iter->second; // 写回数据 write(sock, message.c_str(), message.size()); } else if (s == 0) { // 代表对端关闭连接 logMessage(NORMAL, "%s:%d shutdown , me too", clientip.c_str(), clientport); } else { logMessage(ERROR, "read sock error, %d:%s", errno, strerror(errno)); } close(sock); }
所以采用短连接,客户端可以死循环,每次执行完后就要关闭套接字,再一次连接的时候就要创建套接字。
void usage(std::string proc) { std::cout << "\nUsage: " << proc << " ip port\n" << std::endl; } int main(int argc, char *argv[]) { if (argc != 3) { usage(argv[0]); exit(1); } std::string serverip = argv[1]; uint16_t serverport = atoi(argv[2]); int sock = 0; std::string line; while (true) { // 1.创建套接字 sock = socket(AF_INET, SOCK_STREAM, 0); if (sock < 0) { std::cerr << "socket error" << std::endl; exit(2); } // 不需要显示绑定bind,OS自动选择port // 2.向服务端发起建立连接的请求connect struct sockaddr_in server; memset(&server, 0, sizeof(server)); server.sin_family = AF_INET; server.sin_port = htons(serverport); server.sin_addr.s_addr = inet_addr(serverip.c_str()); if (connect(sock, (struct sockaddr *)&server, sizeof(server)) < 0) { std::cerr << "connect error" << std::endl; exit(3); } // 连接成功 std::cout << "connect success" << std::endl; std::cout << "请输入英文: "; std::getline(std::cin, line); if (line == "quit") break; ssize_t s = send(sock, line.c_str(), line.size(), 0); if (s > 0) { char buffer[1024]; s = recv(sock, buffer, sizeof(buffer) - 1, 0); if (s > 0) { buffer[s] = 0; std::cout << "中文: " << buffer << std::endl; } } close(sock); } return 0; }