引言本篇参考书目《Linux高性能服务器编程》这是我第三次实现网络聊天室这一次是在内存的方面进行了极大的改进并且本篇文章主要使用c11来实现的将书本里面的c语言重写了一遍。代码已经上传到了github上面fengyue05/Linux-: 本仓库里面会有关于《Linux高性能服务器编程》的一些主要代码实现代码在开始真正代码的实现前面我们需要准备一些变量首先是缓存的大小用户的限制事件的限制进程数量的限制。我们这里主要是通过主进程来接受连接广播消息其实相当于我们的服务器端而主进程创建出来的子进程作用是和客户端连接。所以我们客户端的数据结构体里面包含了其地址对应的文件描述符对应的进程还有最重要的就是一个管道我们的子进程只能和父进程通信所以我们通过管道来通信我们通过socketpair把管道两边封装socket然后父进程使用pipefd[0]子进程使用pipefd[1]然后父进程作为服务器还需要处理信号的到来这里我们也使用管道来通信处理。最后我们是通过共享内存的方式进行信息的交流需要一个指针。为了方便我们的代码我们特意开辟了两个数组一个数组是存用户的数据一个数组是存进程方便后面使用。最后我们需要强调一个volatile这个变量的值可能在程序“看不到的地方”被改变所以每次使用它时都应该重新从内存读取不要过度优化。sig_atomic_t 强调的是原子性特别是不会被信号打断。constexpr int USER_LIMIT 5; constexpr int BUFFER_SIZE 1024; constexpr int FD_LIMIT 65535; constexpr int MAX_EVENT_NUMBER 1024; constexpr int PROCESS_LIMIT 65535; constexpr const char* SHM_NAME /my_shm; struct ClientData { sockaddr_in address; int connfd; pid_t pid; // 处理这个连接的子进程的PID std::arrayint, 2 pipefd; // 和父进程通信的管道 }; std::arrayint, 2 m_signalPipe{{-1, -1}}; int m_epollfd -1; int m_listened -1; char* m_sharedMemory nullptr; // 客户连接数组。进程用客户连接的编号来索引这个数组即可取得相关的客户连接数据 std::arrayClientData, USER_LIMIT 1 m_users; std::arrayint, PROCESS_LIMIT m_subProcess; int m_userCount 0; // 当程序正在正常执行时即使突然被信号打断读取或写入这个变量也不会出现“写到一半”的情况。 volatile sig_atomic_t m_stopChild 0;接下来是一个比较常规的操作如果有疑问可以看一下我之前的博客int setNoBlocking(int fd) { const int old_option ::fcntl(fd, F_GETFL); const int new_option old_option | O_NONBLOCK; fcntl(fd, F_SETFL, new_option); return old_option; } void addFd (int epollfd, int fd) { epoll_event event; event.data.fd fd; event.events EPOLLIN | EPOLLET; assert(::epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, event) ! -1); setNoBlocking(fd); } void removeFd(int epollfd, int fd) { if (fd 0) { return; } ::epoll_ctl(epollfd, EPOLL_CTL_DEL, fd, nullptr); ::close(fd); } void signalHandler(int sig) { const int saveErrno errno; const char message static_castchar(sig); ::send(m_signalPipe[1], message, 1, 0); errno saveErrno; } void addSignal (int sig, void (*handler)(int), bool restart true) { struct sigaction sa; sa.sa_handler handler; if (restart) { sa.sa_flags SA_RESTART; } ::sigfillset(sa.sa_mask); assert (::sigaction(sig, sa, nullptr) ! -1); } // 停止一个子进程 void childTermHandler(int) { m_stopChild 1; } void closeFd(int fd) { if (fd 0) { ::close(fd); fd -1; } }这个函数是主进程里面最后销毁所有内存的函数首先我们需要关闭所有存储的文件描述符其实主要就是关闭[0]号管道因为主进程只靠[0]号管道来和对应的进程通信而[1]号管道是在进程创建的时候就在主进程里面关闭了。还有一个重要的就是我们释放共享内存的函数munmap这个和mmap是一对函数一个是创建共享内存一个是释放内存。最后因为我们是通过shm_open打开POSIX共享内存的对象所以我们需要关闭unlinkvoid cleanResouces() { for (int i 0; i m_userCount; i) { // 是在父进程中执行的而父进程保存的是每个 socketpair 的 pipefd[0]与对应子进程通信 closeFd(m_users[i].pipefd[0]); } closeFd(m_signalPipe[0]); closeFd(m_signalPipe[1]); closeFd(m_listened); closeFd(m_epollfd); if (m_sharedMemory ! nullptr m_sharedMemory ! MAP_FAILED) { ::munmap(m_sharedMemory, USER_LIMIT * BUFFER_SIZE); m_sharedMemory nullptr; } ::shm_unlink(SHM_NAME); }我们子进程的作用是和客户端建立联系并且通信所以我们有一个子进程专门运行的函数。我们传入的参数是客户端对应的index从而通过我们存储的数据索引index找到对应的socket等。每一个进程都有一个epollfd因为这个是专门和用户进行监听的同时加入了一个退出的信号以便后面终端的退出。对于一个子进程来说继承了父进程的资源所以也会有users的数据在while循环里面当监听到了事件之后循环处理每一个事件我们对每一个事件有分类处理当事件的sockfd和存储的connfd相同的时候就说明是该进程对应的客户来消息了这个时候就要recv消息不过我们是一个共享内存所以我们不是开辟内存而是直接利用共享内存每个进程对应的共享内存的地方是确定的地址是m_sharedMemory index * BUFFER_SIZE然后把接受到的数据放在这里面。在接受完信息了之后发送消息给主进程服务器端send的是client的编号这样子主进程服务器端就可以知道是哪一个进程也就是对应的客户端。可能有人会疑问那为什么不发送接收到客户端的消息呢原因就是这个是共享内存我们只需要知道发消息对应进程所属的地址这个样子直接可以取数据了所以我们传递的都是index。当事件的sockfd和父进程通信的管道一样的时候说明是接收到了父进程的消息也就是广播这个时候我们接受到了client的index然后就可以找到对应的共享内存位置然后发送给客户端。// 子进程运行的函数参数index指出该子进程处理的客户连接的编号 void runChild(int index) { std::arrayepoll_event, MAX_EVENT_NUMBER events; const int childEpollfd ::epoll_create(1); assert(childEpollfd ! -1); const int connfd m_users[index].connfd; const int pipefd m_users[index].pipefd[1]; // 子进程的socket addSignal(SIGTERM, childTermHandler, false); while(!m_stopChild) { const int number ::epoll_wait(childEpollfd, events.data(), MAX_EVENT_NUMBER, -1); if (number 0 errno ! EINTR) { std::cerr child epoll_wait failed: std::strerror(errno) std::endl; break; } for (int i 0; i number; i) { const int sockfd events[i].data.fd; // 本子进程负责的客户连接有数据到达写到本客户端对应的共享内存操作 // 再把客户编号通知给父进程 if (sockfd connfd (events[i].events EPOLLIN)) { char* buffer m_sharedMemory index * BUFFER_SIZE; std::memset(buffer, 0, BUFFER_SIZE); int received ::recv(connfd, buffer, BUFFER_SIZE - 1, 0); if (received 0) { if (errno ! EAGAIN errno ! EWOULDBLOCK) { m_stopChild 1; } } else if(received 0) { m_stopChild 1; } else { // 传递第几个客户 ::send(pipefd, index, sizeof(index), 0); } } // 父进程通知本进程通过管道将第client个客户的数据发送到本进程负责的客户端 else if (sockfd pipefd (events[i].events EPOLLIN)) { int client -1; int received ::recv(pipefd, client, sizeof(client), 0); if (received 0) { if (errno ! EAGAIN errno ! EWOULDBLOCK) { m_stopChild 1; } } else if(received 0) { m_stopChild 1; } else if (received static_castssize_t(sizeof(client)) client 0 client USER_LIMIT) { // 直接使用共享内存发送给对应的客户端 ::send(connfd, m_sharedMemory client * BUFFER_SIZE, BUFFER_SIZE, 0); } } } } ::close(connfd); ::close(pipefd); ::close(childEpollfd); }当客户端关闭了之后我们删除客户端的一切数据。这个是在父进程里面调用的所以要移除与子进程的通信socket。并且我们的数组排列一定要是连续的否则多次关闭连接之后会浪费很多的内存。所以当我们删除了中间一个index那么我们要把最后一个填补到我们删除的地方包括pidvoid eraseUser (int index) { if (index 0 || index m_userCount) { return; } // 移除父进程为每个子进程保留的通信管道 removeFd(m_epollfd, m_users[index].pipefd[0]); const int last m_userCount - 1; if (index ! last) { m_users[index] m_users[last]; const pid_t movedPid m_users[index].pid; if (movedPid 0 movedPid PROCESS_LIMIT) { m_subProcess[static_castint(movedPid)] index; } } m_userCount--; }对于共享内存我们的一系列操作就是先shm_open打开共享内存的对象其返回的文件描述符很关键我们可以通过这个文件描述符重构内存的大小通过ftruncate函数最后通过mmap创建我们的共享内存并返回这块内存的指针。一切结束了之后就可以关闭这个打开共享内存的文件描述符std::arrayepoll_event, MAX_EVENT_NUMBER events; m_epollfd epoll_create(1); assert(m_epollfd -1); addFd(m_epollfd, m_listened); assert (::socketpair(AF_UNIX, SOCK_STREAM, 0, m_signalPipe.data())); setNoBlocking(m_signalPipe[1]); addFd(m_epollfd, m_signalPipe[0]); addSignal(SIGCHLD, signalHandler); addSignal(SIGTERM, signalHandler); addSignal(SIGINT, signalHandler); addSignal(SIGPIPE, SIG_IGN); bool stopServer false; bool terminating false; const int shmfd ::shm_open(SHM_NAME, O_CREAT | O_RDWR, 0666); assert(shmfd ! -1); // 通过文件描述符修改文件或共享内存对象的大小 assert(::ftruncate(shmfd, USER_LIMIT * BUFFER_SIZE) ! -1); m_sharedMemory static_castchar*(::mmap(nullptr, USER_LIMIT * BUFFER_SIZE, PROT_READ | PROT_WRITE, MAP_SHARED, shmfd, 0)); ::close(shmfd);在while里面我们解决的就是几个问题1、如果有连接发来存储对应的数据创建对应的管道然后创建对应的子进程在子进程里面只保留和父进程通信的socket和客户端通信的socket其他全部关闭如果结束了子进程就要munmap其共享内存的部分。而父进程要做的事情就是存储子进程的信息不过我们进程数量都是有限的所以当我们创建了子进程之后要判断如果超过了这个上限那么就要关闭。2、当有信号到来的时候如果是子进程退出的信号父进程非阻塞回收因为有信号来了一定是可以回收的而非阻塞回收的特点就是高效。如果是父进程被关闭的信号来了那么要关闭所有的子进程。3、通过广播发送信息其实就是传递对应的index让子进程里面运行的函数接受index自己读取共享内存里面的数据。while (!stopServer) { const int number epoll_wait(m_epollfd, events.data(), MAX_EVENT_NUMBER, -1); if (number 0 errno ! EINTR) { std::cerr epoll_wait failed: std::strerror(errno) \n; break; } for (int i 0; i number; i) { const int sockfd events[i].data.fd; if (sockfd m_listened (events[i].events EPOLLIN)) { sockaddr_in clientAddress; socklen_t len sizeof(clientAddress); const int connfd ::accept(m_listened, (sockaddr*)clientAddress, len); if (connfd 0) { if (errno ! EAGAIN errno ! EWOULDBLOCK) { std::cerr accept failed: std::strerror(errno) std::endl; } continue; } if (m_userCount USER_LIMIT) { const std::string message too many users\n; std::cerr message; ::send(connfd, message.data(), message.size(), 0); ::close(connfd); continue; } const int index m_userCount; m_users[index].address clientAddress; m_users[index].connfd connfd; assert(::socketpair(AF_UNIX, SOCK_STREAM, 0, m_users[index].pipefd.data()) ! -1); const pid_t pid ::fork(); // fork失败啦 if (pid 0) { ::close(connfd); ::close(m_users[index].pipefd[0]); ::close(m_users[index].pipefd[1]); continue; } if (pid 0) { ::close(m_epollfd); ::close(m_listened); ::close(m_users[index].pipefd[0]); ::close(m_signalPipe[0]); ::close(m_signalPipe[1]); runChild(index); ::munmap(m_sharedMemory, USER_LIMIT * BUFFER_SIZE); _exit(EXIT_SUCCESS); } ::close(connfd); ::close(m_users[index].pipefd[1]); m_users[index].pipefd[1] -1; addFd (m_epollfd, m_users[index].pipefd[0]); m_users[index].pid pid; if (pid PROCESS_LIMIT) { std::cerr child PID exceeds PROCESS_LIMIT\n; ::kill(pid, SIGTERM); removeFd(m_epollfd, m_users[index].pipefd[0]); continue; } m_subProcess[static_castint(pid)] index; m_userCount; } // 统一信号源从信号管道读取信号编号 else if (sockfd m_signalPipe[0] (events[i].events EPOLLIN)) { std::arraychar, 1024 signals; int received ::recv(m_signalPipe[0], signals.data(), signals.size(), 0); if (received 0) { continue; } for (int j 0; j received; j) { switch (signals[j]) { case SIGCHLD: while(true) { int status 0; const pid_t pid ::waitpid(-1, status, WNOHANG); if (pid 0) { break; } if (pid PROCESS_LIMIT) { const int index m_subProcess[pid]; m_subProcess[pid] -1; eraseUser(index); } } if (terminating m_userCount 0) { stopServer true; } break; case SIGTERM: case SIGINT: if (m_userCount 0) { stopServer true; } else { for (int k 0; k m_userCount; k) { ::kill(m_users[k].pid, SIGTERM); } terminating true; } break; default: break; } } } // 某个子进程写入了共享内存通过广播广播给其他子进程 else if (events[i].events EPOLLIN) { int child -1; const int received ::recv(sockfd, child, sizeof(child), 0); if (received 0) { continue; } for (int j 0; j m_userCount; j) { if (m_users[j].pipefd[0] sockfd) { continue; } ::send(m_users[j].pipefd[0], child, sizeof(child), 0); } } } } cleanResouces();所以共享内存的聊天室最大的优点就是不需要send信息减少了拷贝的开销而是仅仅读取共享内存的数据这个对于性能有很大的提高。总结本篇文章到这里就结束了希望可以帮助大家理解~~~