epoll实现多人聊天室

佚名 6年前 (2019-04-11) 随笔 884人围观抢沙发百度已收录

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; // clients_list save all the clients's socket
list<int> clients_list;//存储服务端的在线客户 /**********************   macro defintion **************************/
// server ip
#define SERVER_IP "127.0.0.1"//使用本机环路测试地址作为连接socket绑定的ip地址 // server port
#define SERVER_PORT 8888//使用8888作为连接socket绑定的端口 //epoll size
#define EPOLL_SIZE 5000//epoll事件表最大事件数量 //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!" /**********************   some function **************************/
/**
* @param sockfd: socket descriptor
* @return 0
**/
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 )//将感兴趣的fd挂到epollfd指向的事件表
{
    struct epoll_event ev;
    ev.data.fd = fd;
    ev.events = EPOLLIN;
    if( enable_et )
        ev.events = EPOLLIN | EPOLLET;
    epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &ev);
    setnonblocking(fd);
    printf("fd added to epoll!\n\n");
} /**
* @param clientfd: socket descriptor
* @return : len
**/
int sendBroadcastmessage(int clientfd)//广播消息
{
    // buf[BUF_SIZE] receive new chat message
    // message[BUF_SIZE] save format message
    char buf[BUF_SIZE], message[BUF_SIZE];
    bzero(buf, BUF_SIZE);
    bzero(message, BUF_SIZE);     // receive message
    printf("read from client(clientID = %d)\n", clientfd);
    int len = recv(clientfd, buf, BUF_SIZE, 0);//接收clientfd号聊天室的数据     if(len == 0) // len = 0 means the client closed connection
    {
        close(clientfd);
        clients_list.remove(clientfd); //server remove the client
        printf("ClientID = %d closed.\n now there are %d client in the char room\n", clientfd, (int)clients_list.size());     }
    else //broadcast message
    {
        if(clients_list.size() == 1) { // this means There is only one int the char room
            send(clientfd, CAUTION, strlen(CAUTION), 0);
            return len;
        }
        // format message to broadcast
        sprintf(message, SERVER_MESSAGE, clientfd, buf);//SERVER_MESSAGE="ClientID %d say >> %s",以此格式把数据发送給所有聊天室         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

客户端.cpp

SRE实战互联网时代守护先锋，助力企业售后服务体系运筹帷幄！一键直达领取阿里云限量特价优惠。

客户端要处理的事件有2种，1.来自服务器的数据或者断开连接 .2.处理用户输入的数据。所以为了高效使用并发编程，父进程处理服务器相关的操作，子进程处理用户输入

代码中isClientwork状态用来标记该聊天室是否工作，当使用了fork后，父子进程共享数据，但是数据会写时更新，即无论父子进程修改这个变量都不会影响对方，因为修改这个值后会为子进程分配新的数据空间

EPOLLIN：表示对应的文件描述符可以读（包括对端SOCKET正常关闭）；

再来看一个问题，代码中父进程结束有2种方式，一是服务器关闭连接，此时socke即connect有事件发生，对端关闭。二是用户输入EXIT，子进程结束，写管道关闭，因为读管道文件描述符被挂在epoll注册的事件表中，它的对端关闭了，所以有事件发生。两者

读取数据都为0，判定为对端正常关闭。

如果输入了EXIT，子进程关闭，然后父进程关闭。

如果服务器关闭，那么父进程关闭，但是子进程并没有退出，它变成了孤儿进程，它的父进程为init（1号进程），当你向终端写数据时，因为父进程关闭了读管道，所以写入失败，exit(-1)，子进程退出

#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 sock = socket(PF_INET, SOCK_STREAM, 0);
    if(sock < 0) { perror("sock error"); exit(-1); }
    // 连接服务端
    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;
}