epoll聊天室的实现

1.服务端

a. 支持多个用户接入，实现聊天室的基本功能

b. 使用epoll机制实现并发，增加效率

2. 客户端

a. 支持用户输入聊天消息

b. 显示其他用户输入的信息

c. 使用fork创建两个进程

子进程有两个功能：

等待用户输入聊天信息
将聊天信息写到管道（pipe），并发送给父进程

父进程有两个功能

使用epoll机制接受服务端发来的信息，并显示给用户，使用户看到其他用户的聊天信息
将子进程发给的聊天信息从管道（pipe）中读取, 并发送给服务端

C/S模型

服务端和客户端采用经典的C/S模型，并且使用TCP连接.

TCP服务端通信的常规步骤

(1)使用socket()创建TCP套接字（socket）

(2)将创建的套接字绑定到一个本地地址和端口上（Bind）

(3)将套接字设为监听模式，准备接收客户端请求（listen）

(4)等待客户请求到来: 当请求到来后，接受连接请求，返回一个对应于此次连接的新的套接字（accept）

(5)用accept返回的套接字和客户端进行通信（使用write()/send()或send()/recv() )

(6)返回，等待另一个客户请求

(7)关闭套接字

TCP客户端通信的常规步骤

(1)创建套接字（socket）

(2)使用connect()建立到达服务器的连接（connect)

(3)客户端进行通信（使用write()/send()或send()/recv())

(4)使用close()关闭客户连接

阻塞与非阻塞socket

通常的，对一个文件描述符指定的文件或设备, 有两种工作方式: 阻塞与非阻塞方式。

（1）. 阻塞方式是指：当试图对该文件描述符进行读写时，如果当时没有数据可读，或者暂时不可写，程序就进入等待状态，直到有东西可读或者可写为止。

（2）. 非阻塞方式是指：如果没有数据可读，或者不可写，读写函数马上返回，而不会等待。

阻塞方式和非阻塞方式唯一的区别：是否立即返回。本项目采用更高效的做法，所以应该将socket设置为非阻塞方式。这样能充分利用服务器资源，效率得到了很大提高。

epoll

当服务端的在线人数越来越多，会导致系统资源吃紧，I/O效率越来越慢，这时候就应该考虑epoll了。epoll是Linux内核为处理大批句柄而作改进的poll，是Linux特有的I/O函数。其特点如下：

epoll是Linux下多路复用IO接口select/poll的增强版本。其实现和使用方式与select/poll有很多不同，epoll通过一组函数来完成有关任务，而不是一个函数。

epoll之所以高效，是因为epoll将用户关心的文件描述符放到内核里的一个事件表中，而不是像select/poll每次调用都需要重复传入文件描述符集或事件集。比如当一个事件发生（比如说读事件），epoll无须遍历整个被侦听的描述符集，只要遍历那些被内核IO事件异步唤醒而加入就绪队列的描述符集合就行了。

epoll有两种工作方式，LT(level triggered)：水平触发和ET(edge-triggered)：边沿触发。LT是select/poll使用的触发方式，比较低效；而ET是epoll的高速工作方式（本项目使用epoll的ET方式）。

epoll 共3个函数, 如下:

1、int epoll_create(int size)
创建一个epoll句柄，参数size用来告诉内核监听的数目，size为epoll所支持的最大句柄数

2、int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)
函数功能： epoll事件注册函数
　参数epfd为epoll的句柄，即epoll_create返回值
　参数op表示动作，用3个宏来表示：
　　 EPOLL_CTL_ADD(注册新的fd到epfd)，
　EPOLL_CTL_MOD(修改已经注册的fd的监听事件)，
　　 EPOLL_CTL_DEL(从epfd删除一个fd)；
　　其中参数fd为需要监听的标示符；
　参数event告诉内核需要监听的事件，event的结构如下：
struct epoll_event {
__uint32_t events; //Epoll events
epoll_data_t data; //User data variable
};
其中介绍events是宏的集合，本项目主要使用EPOLLIN(表示对应的文件描述符可以读，即读事件发生)

3、 int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout)
等待事件的产生，函数返回需要处理的事件数目（该数目是就绪事件的数目，就是前面所说漂亮女孩的个数N）

因此服务端使用epoll的时候，步骤如下：

调用epoll_create函数在Linux内核中创建一个事件表；
然后将文件描述符（监听套接字listener）添加到所创建的事件表中；
在主循环中，调用epoll_wait等待返回就绪的文件描述符集合；
分别处理就绪的事件集合，本项目中一共有两类事件：新用户连接事件和用户发来消息事件

服务端实现

utility.h完整源码

  1 #ifndef UTILITY_H_INCLUDED
  2 #define UTILITY_H_INCLUDED
  3  
  4 #include <iostream>
  5 #include <list>
  6 #include <sys/types.h>
  7 #include <sys/socket.h>
  8 #include <netinet/in.h>
  9 #include <arpa/inet.h>
 10 #include <sys/epoll.h>
 11 #include <fcntl.h>
 12 #include <errno.h>
 13 #include <unistd.h>
 14 #include <stdio.h>
 15 #include <stdlib.h>
 16 #include <string.h>
 17  
 18 using namespace std;
 19  
 20 // clients_list save all the clients's socket
 21 list<int> clients_list;
 22  
 23 /**********************   macro defintion **************************/
 24 // server ip
 25 #define SERVER_IP "127.0.0.1"
 26  
 27 // server port
 28 #define SERVER_PORT 8888
 29  
 30 //epoll size
 31 #define EPOLL_SIZE 5000
 32  
 33 //message buffer size
 34 #define BUF_SIZE 0xFFFF
 35  
 36 #define SERVER_WELCOME "Welcome you join  to the chat room! Your chat ID is: Client #%d"
 37  
 38 #define SERVER_MESSAGE "ClientID %d say >> %s"
 39  
 40 // exit
 41 #define EXIT "EXIT"
 42  
 43 #define CAUTION "There is only one int the char room!"
 44  
 45 /**********************   some function **************************/
 46 /**
 47   * @param sockfd: socket descriptor
 48   * @return 0
 49 **/
 50 int setnonblocking(int sockfd)
 51 {
 52     fcntl(sockfd, F_SETFL, fcntl(sockfd, F_GETFD, 0)| O_NONBLOCK);
 53     return 0;
 54 }
 55  
 56 /**
 57   * @param epollfd: epoll handle
 58   * @param fd: socket descriptor
 59   * @param enable_et : enable_et = true, epoll use ET; otherwise LT
 60 **/
 61 void addfd( int epollfd, int fd, bool enable_et )
 62 {
 63     struct epoll_event ev;
 64     ev.data.fd = fd;
 65     ev.events = EPOLLIN;
 66     if( enable_et )
 67         ev.events = EPOLLIN | EPOLLET;
 68     epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &ev);
 69     setnonblocking(fd);
 70     printf("fd added to epoll!

");
 71 }
 72  
 73 /**
 74   * @param clientfd: socket descriptor
 75   * @return : len
 76 **/
 77 int sendBroadcastmessage(int clientfd)
 78 {
 79     // buf[BUF_SIZE] receive new chat message
 80     // message[BUF_SIZE] save format message
 81     char buf[BUF_SIZE], message[BUF_SIZE];
 82     bzero(buf, BUF_SIZE);
 83     bzero(message, BUF_SIZE);
 84  
 85     // receive message
 86     printf("read from client(clientID = %d)
", clientfd);
 87     int len = recv(clientfd, buf, BUF_SIZE, 0);
 88  
 89     if(len == 0)  // len = 0 means the client closed connection
 90     {
 91         close(clientfd);
 92         clients_list.remove(clientfd); //server remove the client
 93         printf("ClientID = %d closed.
 now there are %d client in the char room
", clientfd, (int)clients_list.size());
 94  
 95     }
 96     else  //broadcast message 
 97     {
 98         if(clients_list.size() == 1) { // this means There is only one int the char room
 99             send(clientfd, CAUTION, strlen(CAUTION), 0);
100             return len;
101         }
102         // format message to broadcast
103         sprintf(message, SERVER_MESSAGE, clientfd, buf);
104  
105         list<int>::iterator it;
106         for(it = clients_list.begin(); it != clients_list.end(); ++it) {
107            if(*it != clientfd){
108                 if( send(*it, message, BUF_SIZE, 0) < 0 ) { perror("error"); exit(-1);}
109            }
110         }
111     }
112     return len;
113 }
114 #endif // UTILITY_H_INCLUDED

服务端完整源码

 1 #include "utility.h"
 2  
 3 int main(int argc, char *argv[])
 4 {
 5     //服务器IP + port
 6     struct sockaddr_in serverAddr;
 7     serverAddr.sin_family = PF_INET;
 8     serverAddr.sin_port = htons(SERVER_PORT);
 9     serverAddr.sin_addr.s_addr = inet_addr(SERVER_IP);
10     //创建监听socket
11     int listener = socket(PF_INET, SOCK_STREAM, 0);
12     if(listener < 0) { perror("listener"); exit(-1);}
13     printf("listen socket created 
");
14     //绑定地址
15     if( bind(listener, (struct sockaddr *)&serverAddr, sizeof(serverAddr)) < 0) {
16         perror("bind error");
17         exit(-1);
18     }
19     //监听
20     int ret = listen(listener, 5);
21     if(ret < 0) { perror("listen error"); exit(-1);}
22     printf("Start to listen: %s
", SERVER_IP);
23     //在内核中创建事件表
24     int epfd = epoll_create(EPOLL_SIZE);
25     if(epfd < 0) { perror("epfd error"); exit(-1);}
26     printf("epoll created, epollfd = %d
", epfd);
27     static struct epoll_event events[EPOLL_SIZE];
28     //往内核事件表里添加事件
29     addfd(epfd, listener, true);
30     //主循环
31     while(1)
32     {
33         //epoll_events_count表示就绪事件的数目
34         int epoll_events_count = epoll_wait(epfd, events, EPOLL_SIZE, -1);
35         if(epoll_events_count < 0) {
36             perror("epoll failure");
37             break;
38         }
39  
40         printf("epoll_events_count = %d
", epoll_events_count);
41         //处理这epoll_events_count个就绪事件
42         for(int i = 0; i < epoll_events_count; ++i)
43         {
44             int sockfd = events[i].data.fd;
45             //新用户连接
46             if(sockfd == listener)
47             {
48                 struct sockaddr_in client_address;
49                 socklen_t client_addrLength = sizeof(struct sockaddr_in);
50                 int clientfd = accept( listener, ( struct sockaddr* )&client_address, &client_addrLength );
51  
52                 printf("client connection from: %s : % d(IP : port), clientfd = %d 
",
53                 inet_ntoa(client_address.sin_addr),
54                 ntohs(client_address.sin_port),
55                 clientfd);
56  
57                 addfd(epfd, clientfd, true);
58  
59                 // 服务端用list保存用户连接
60                 clients_list.push_back(clientfd);
61                 printf("Add new clientfd = %d to epoll
", clientfd);
62                 printf("Now there are %d clients int the chat room
", (int)clients_list.size());
63  
64                 // 服务端发送欢迎信息  
65                 printf("welcome message
");                
66                 char message[BUF_SIZE];
67                 bzero(message, BUF_SIZE);
68                 sprintf(message, SERVER_WELCOME, clientfd);
69                 int ret = send(clientfd, message, BUF_SIZE, 0);
70                 if(ret < 0) { perror("send error"); exit(-1); }
71             }
72             //处理用户发来的消息，并广播，使其他用户收到信息
73             else
74             {   
75                 int ret = sendBroadcastmessage(sockfd);
76                 if(ret < 0) { perror("error");exit(-1); }
77             }
78         }
79     }
80     close(listener); //关闭socket
81     close(epfd);    //关闭内核
82     return 0;
83 }

客户端实现

子进程和父进程的通信

通过调用int pipe(int fd[2])函数创建管道, 其中fd[0]用于父进程读， fd[1]用于子进程写。

通过int pid = fork()函数，创建子进程，当pid < 0 错误；当pid = 0, 说明是子进程；当pid > 0说明是父进程。根据pid的值，我们可以父子进程，从而实现对应的功能！

客户端完整源码

  1 #include "utility.h"
  2  
  3 int main(int argc, char *argv[])
  4 {
  5     //用户连接的服务器 IP + port
  6     struct sockaddr_in serverAddr;
  7     serverAddr.sin_family = PF_INET;
  8     serverAddr.sin_port = htons(SERVER_PORT);
  9     serverAddr.sin_addr.s_addr = inet_addr(SERVER_IP);
 10  
 11     // 创建socket
 12     int sock = socket(PF_INET, SOCK_STREAM, 0);
 13     if(sock < 0) { perror("sock error"); exit(-1); }
 14     // 连接服务端
 15     if(connect(sock, (struct sockaddr *)&serverAddr, sizeof(serverAddr)) < 0) {
 16         perror("connect error");
 17         exit(-1);
 18     }
 19  
 20     // 创建管道，其中fd[0]用于父进程读，fd[1]用于子进程写
 21     int pipe_fd[2];
 22     if(pipe(pipe_fd) < 0) { perror("pipe error"); exit(-1); }
 23  
 24     // 创建epoll
 25     int epfd = epoll_create(EPOLL_SIZE);
 26     if(epfd < 0) { perror("epfd error"); exit(-1); }
 27     static struct epoll_event events[2]; 
 28     //将sock和管道读端描述符都添加到内核事件表中
 29     addfd(epfd, sock, true);
 30     addfd(epfd, pipe_fd[0], true);
 31     // 表示客户端是否正常工作
 32     bool isClientwork = true;
 33  
 34     // 聊天信息缓冲区
 35     char message[BUF_SIZE];
 36  
 37     // Fork
 38     int pid = fork();
 39     if(pid < 0) { perror("fork error"); exit(-1); }
 40     else if(pid == 0)      // 子进程
 41     {
 42         //子进程负责写入管道，因此先关闭读端
 43         close(pipe_fd[0]); 
 44         printf("Please input 'exit' to exit the chat room
");
 45  
 46         while(isClientwork){
 47             bzero(&message, BUF_SIZE);
 48             fgets(message, BUF_SIZE, stdin);
 49  
 50             // 客户输出exit,退出
 51             if(strncasecmp(message, EXIT, strlen(EXIT)) == 0){
 52                 isClientwork = 0;
 53             }
 54             // 子进程将信息写入管道
 55             else {
 56                 if( write(pipe_fd[1], message, strlen(message) - 1 ) < 0 )
 57                  { perror("fork error"); exit(-1); }
 58             }
 59         }
 60     }
 61     else  //pid > 0 父进程
 62     {
 63         //父进程负责读管道数据，因此先关闭写端
 64         close(pipe_fd[1]); 
 65  
 66         // 主循环(epoll_wait)
 67         while(isClientwork) {
 68             int epoll_events_count = epoll_wait( epfd, events, 2, -1 );
 69             //处理就绪事件
 70             for(int i = 0; i < epoll_events_count ; ++i)
 71             {
 72                 bzero(&message, BUF_SIZE);
 73  
 74                 //服务端发来消息
 75                 if(events[i].data.fd == sock)
 76                 {
 77                     //接受服务端消息
 78                     int ret = recv(sock, message, BUF_SIZE, 0);
 79  
 80                     // ret= 0 服务端关闭
 81                     if(ret == 0) {
 82                         printf("Server closed connection: %d
", sock);
 83                         close(sock);
 84                         isClientwork = 0;
 85                     }
 86                     else printf("%s
", message);
 87  
 88                 }
 89                 //子进程写入事件发生，父进程处理并发送服务端
 90                 else { 
 91                     //父进程从管道中读取数据
 92                     int ret = read(events[i].data.fd, message, BUF_SIZE);
 93  
 94                     // ret = 0
 95                     if(ret == 0) isClientwork = 0;
 96                     else{   // 将信息发送给服务端
 97                       send(sock, message, BUF_SIZE, 0);
 98                     }
 99                 }
100             }//for
101         }//while
102     }
103  
104     if(pid){
105        //关闭父进程和sock
106         close(pipe_fd[0]);
107         close(sock);
108     }else{
109         //关闭子进程
110         close(pipe_fd[1]);
111     }
112     return 0;
113 }