epoll实现高并发聊天室

深藏阁楼爱情的钟 2022-09-26 02:08 207阅读 0赞

### 关于epoll，请看上一篇文章: [epoll详解][epoll] ###

本代码包含三个文件:  
utility.h  
client.cpp  
server.cpp  
可实现局域网内多主机异步通信

--------------------

## 下面是效果图: ##

## **1. 开启服务端：** ##

![这里写图片描述][20160915165725383]

## **2. 开启两个客户端:** ##

客户端1:  
![这里写图片描述][20160915170225173]  
客户端2:  
![这里写图片描述][20160915170241720]  
服务端的提示消息:  
![这里写图片描述][20160915170331361]

## **3. 消息传送：** ##

客户端1:  
![这里写图片描述][20160915173200123]  
客户端2:  
![这里写图片描述][20160915173217889]  
服务端的提示消息并广播:  
![这里写图片描述][20160915173233327]

代码:

//utility.h
    #ifndef UTILITY_H_INCLUDED
    #define UTILITY_H_INCLUDED
    
    #include <iostream>
    #include <list>
    #include <sys/types.h>
    #include <sys/socket.h>
    #include <netinet/in.h>
    #include <arpa/inet.h>
    #include <sys/epoll.h>
    #include <fcntl.h>
    #include <errno.h>
    #include <unistd.h>
    #include <stdio.h>
    #include <stdlib.h>
    #include <string.h>
    
    using namespace std;
    
    // 选用list方便删除sockfd
    list<int> clients_list;
    
    
    // ip
    #define SERVER_IP "192.168.1.193"
    
    //桥接后在同一网段即可选用以下地址
    //#define CLIENT_IP "192.168.1.XXX"
    
    // server port
    #define SERVER_PORT 8888
    
    //epoll 支持的最大并发量
    #define EPOLL_SIZE 5000
    
    //message buffer size
    #define BUF_SIZE 0xFFFF
    
    #define SERVER_WELCOME "Welcome you join  to the chat room! Your chat ID is: Client #%d"
    
    #define SERVER_MESSAGE "ClientID %d say >> %s"
    
    // exit
    #define EXIT "EXIT"
    
    #define CAUTION "There is only one int the char room!"
    
    
    
    //设置sockfd,pipefd非阻塞
    int setnonblocking(int sockfd)
    {
        fcntl(sockfd, F_SETFL, fcntl(sockfd, F_GETFD, 0)| O_NONBLOCK);
        return 0;
    }
    
    /**
      * @param epollfd: epoll handle
      * @param fd: socket descriptor
      * @param enable_et : enable_et = true, epoll use ET; otherwise LT
    **/
    void addfd( int epollfd, int fd, bool enable_et )
    {
        struct epoll_event ev;
        ev.data.fd = fd;
        ev.events = EPOLLIN;     //输入触发epoll-event
        if( enable_et )
            ev.events = EPOLLIN | EPOLLET;
        epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &ev);
        setnonblocking(fd);
     //   printf("fd added to epoll!\n\n");
    }
    
    //将服务器收到的clientfd的消息进行广播
    int sendBroadcastmessage(int clientfd)
    {
    
        char buf[BUF_SIZE], message[BUF_SIZE];
        //清零
        bzero(buf, BUF_SIZE);
        bzero(message, BUF_SIZE);
    
        printf("read from client(clientID = %d)\n", clientfd);
        int len = recv(clientfd, buf, BUF_SIZE, 0);
    
        if(len == 0)  // len = 0 client关闭了连接
        {
            close(clientfd);
            clients_list.remove(clientfd); //list删除fd
            printf("ClientID = %d closed.\n now there are %d client in the char room\n", clientfd, (int)clients_list.size());
    
        }
        else  //进行广播
        {
            if(clients_list.size() == 1) {
                send(clientfd, CAUTION, strlen(CAUTION), 0);
                return len;
            }
    
            sprintf(message, SERVER_MESSAGE, clientfd, buf);
    
            list<int>::iterator it;
            for(it = clients_list.begin(); it != clients_list.end(); ++it) {
               if(*it != clientfd){
                    if( send(*it, message, BUF_SIZE, 0) < 0 ) { perror("error"); exit(-1);}
               }
            }
        }
        return len;
    }
    #endif // UTILITY_H_INCLUDED

//server.cpp
    #include "utility.h"
    
    int main(int argc, char *argv[])
    {
        //服务器IP + port
        struct sockaddr_in serverAddr;
        serverAddr.sin_family = PF_INET;
        serverAddr.sin_port = htons(SERVER_PORT);
        serverAddr.sin_addr.s_addr = inet_addr(SERVER_IP);
        //创建监听socket
        int listener = socket(PF_INET, SOCK_STREAM, 0);
        if(listener < 0) { perror("listener"); exit(-1);}
        printf("listen socket created \n");
        //绑定地址
        if( bind(listener, (struct sockaddr *)&serverAddr, sizeof(serverAddr)) < 0) {
            perror("bind error");
            exit(-1);
        }
        //监听
        int ret = listen(listener, 5);
        if(ret < 0) { perror("listen error"); exit(-1);}
        printf("%s Start to listen\n", SERVER_IP);
        //在内核中创建事件表
        int epfd = epoll_create(EPOLL_SIZE);
        if(epfd < 0) { perror("epfd error"); exit(-1);}
        printf("epoll created, epollfd = %d\n", epfd);
        static struct epoll_event events[EPOLL_SIZE];
        //往内核事件表里添加事件
        addfd(epfd, listener, true);
        //主循环
        while(1)
        {
            //epoll_events_count表示就绪事件的数目
            int epoll_events_count = epoll_wait(epfd, events, EPOLL_SIZE, -1);
            if(epoll_events_count < 0) {
                perror("epoll failure");
                break;
            }
    
            printf("epoll_events_count = %d\n", epoll_events_count);
            //处理这epoll_events_count个就绪事件
            for(int i = 0; i < epoll_events_count; ++i)
            {
                int sockfd = events[i].data.fd;
                //新用户连接
                if(sockfd == listener)
                {
                    struct sockaddr_in client_address;
                    socklen_t client_addrLength = sizeof(struct sockaddr_in);
                    int clientfd = accept( listener, ( struct sockaddr* )&client_address, &client_addrLength );
    
                    printf("client connection from: %s : % d(IP : port), clientfd = %d \n",
                    inet_ntoa(client_address.sin_addr),
                    ntohs(client_address.sin_port),
                    clientfd);
    
                    addfd(epfd, clientfd, true);
    
                    // 服务端用list保存用户连接
                    clients_list.push_back(clientfd);
                    printf("Add new clientfd = %d to epoll\n", clientfd);
                    printf("Now there are %d clients int the chat room\n", (int)clients_list.size());
    
                    // 服务端发送欢迎信息
                 //   printf("welcome message\n");
                    char message[BUF_SIZE];
                    bzero(message, BUF_SIZE);
                    sprintf(message, SERVER_WELCOME, clientfd);
                    int ret = send(clientfd, message, BUF_SIZE, 0);
                    if(ret < 0) { perror("send error"); exit(-1); }
                }
                //处理用户发来的消息，并广播，使其他用户收到信息
                else
                {
                    int ret = sendBroadcastmessage(sockfd);
                    if(ret < 0) { perror("error");exit(-1); }
                }
            }
        }
        close(listener); //关闭socket
        close(epfd);    //关闭内核
        return 0;
    }

//client.cpp
    #include "utility.h"
    
    int main(int argc, char *argv[])
    {
        printf("this client ip is %s\n",CLIENT_IP);
        //用户连接的服务器 IP + port
        struct sockaddr_in serverAddr;
        serverAddr.sin_family = PF_INET;
        serverAddr.sin_port = htons(SERVER_PORT);
        serverAddr.sin_addr.s_addr = inet_addr(CLIENT_IP);
    
        // 创建socket
        int sock = socket(PF_INET, SOCK_STREAM, 0);
        if(sock < 0) { perror("sock error"); exit(-1); }
    
        serverAddr.sin_addr.s_addr = inet_addr(SERVER_IP);
        // 连接服务端
        if(connect(sock, (struct sockaddr *)&serverAddr, sizeof(serverAddr)) < 0) {
            perror("connect error");
            exit(-1);
        }
    
        // 创建管道，其中fd[0]用于父进程读，fd[1]用于子进程写
        int pipe_fd[2];
        if(pipe(pipe_fd) < 0) { perror("pipe error"); exit(-1); }
    
        // 创建epoll
        int epfd = epoll_create(EPOLL_SIZE);
        if(epfd < 0) { perror("epfd error"); exit(-1); }
        static struct epoll_event events[2];
        //将sock和管道读端描述符都添加到内核事件表中
        addfd(epfd, sock, true);
        addfd(epfd, pipe_fd[0], true);
        // 表示客户端是否正常工作
        bool isClientwork = true;
    
        // 聊天信息缓冲区
        char message[BUF_SIZE];
    
        // Fork
        int pid = fork();
        if(pid < 0) { perror("fork error"); exit(-1); }
        else if(pid == 0)      // 子进程
        {
            //子进程负责写入管道，因此先关闭读端
            close(pipe_fd[0]);
            printf("Please input 'exit' to exit the chat room\n");
    
            while(isClientwork){
                bzero(&message, BUF_SIZE);
                fgets(message, BUF_SIZE, stdin);
    
                // 客户输出exit,退出
                if(strncasecmp(message, EXIT, strlen(EXIT)) == 0){
                    isClientwork = 0;
                }
                // 子进程将信息写入管道
                else {
                    if( write(pipe_fd[1], message, strlen(message) - 1 ) < 0 )
                     { perror("fork error"); exit(-1); }
                }
            }
        }
        else  //pid > 0 父进程
        {
            //父进程负责读管道数据，因此先关闭写端
            close(pipe_fd[1]);
    
            // 主循环(epoll_wait)
            while(isClientwork) {
                int epoll_events_count = epoll_wait( epfd, events, 2, -1 );
                //处理就绪事件
                for(int i = 0; i < epoll_events_count ; ++i)
                {
                    bzero(&message, BUF_SIZE);
    
                    //服务端发来消息
                    if(events[i].data.fd == sock)
                    {
                        //接受服务端消息
                        int ret = recv(sock, message, BUF_SIZE, 0);
    
                        // ret= 0 服务端关闭
                        if(ret == 0) {
                            printf("Server closed connection: %d\n", sock);
                            close(sock);
                            isClientwork = 0;
                        }
                        else printf("%s\n", message);
    
                    }
                    //子进程写入事件发生，父进程处理并发送服务端
                    else {
                        //父进程从管道中读取数据
                        int ret = read(events[i].data.fd, message, BUF_SIZE);
    
                        // ret = 0
                        if(ret == 0) isClientwork = 0;
                        else{   // 将信息发送给服务端
                          send(sock, message, BUF_SIZE, 0);
                        }
                    }
                }//for
            }//while
        }
    
        if(pid){
           //关闭父进程和sock
            close(pipe_fd[0]);
            close(sock);
        }else{
            //关闭子进程
            close(pipe_fd[1]);
        }
        return 0;
    }

[epoll]: http://blog.csdn.net/axuan_k/article/details/52529420
[20160915165725383]: /images/20220717/650c218c830342708a6b8cdb760078c8.png
[20160915170225173]: /images/20220717/6f87530431464c748e3c99e74c12c945.png
[20160915170241720]: /images/20220717/3a2805c29ca5418a88ae3f7bd0e6e616.png
[20160915170331361]: /images/20220717/b26b78491f174c0fa34a85502141e889.png
[20160915173200123]: /images/20220717/5f7de69eb8064b7393002fe74e23302f.png
[20160915173217889]: /images/20220717/269bed0456f64d2ba78895ae143544da.png
[20160915173233327]: /images/20220717/f319aa8bec704b4e9f1f390c80a322b9.png