I/O多路复用：基于epoll实现Reactor高性能TCP服务器
目录
前言
Linux下的I/O多路复用技术是构建高并发网络服务器的核心，多路复用经历了从select/poll/epoll的发展演进。select和poll采取轮询机制，每次调用都需将文件描述符集合从用户态拷贝到内核态，时间复杂度为O(n)，当连接数增大时性能将线性下降。epoll则解决了select/poll性能这方面的问题，epoll底层通过红黑树来管理已注册的文件描述符，使用就绪链表来存储活跃连接，通过回调机制实现O(1)的事件通知，避免了无效的文件描述符遍历。本文将围绕I/O多路复用展开介绍，并在I/O多路复用的基础上，基于Reactor设计模式，以epoll作为底层事件通知机制，实现一个高性能的TCP服务器。基于epoll的Reactor模式，使得I/O事件与业务处理分离，以事件循环驱动整个服务器，是Linux下高性能网络服务器实现的经典范式。
一、I/O多路复用
1、select
select用于实现多路复用输入/输出模型，select可以监视多个文件描述符的状态变化，调用select后，程序将停在select等待，直到被监视的文件描述符有一个或多个发生了状态改变。
nfds：需要监视的最大文件描述符值+1；
readfds、writefds、exceptfds分别为需要检测的可读文件描述符集合、可写文件描述符集合及异常文件描述符集合；
参数timeout为结构timeval，用来设置select的等待时间：
NULL：则表示select没有timeout，select将一直阻塞，直到某个文件描述符上发生了事件；
0：仅检测文件描述符集合的状态，然后立即返回，并不等待外部事件的发生；
特定的时间值：如果在指定的时间段里没有事件发生，select将超时返回。
（1）fd_set
关于select参数类型fd_set的结构，fd_set为位图结构，通过使用位图中对应的位来表示要监视的文件描述符。通过fd_set的接口，可以比较方便的操作位图。
void FD_CLR(int fd, fd_set *set); // ⽤来清除描述词组set中相关fd 的位
int FD_ISSET(int fd, fd_set *set);// ⽤来测试描述词组set中相关fd 的位是否为真
void FD_SET(int fd, fd_set *set);// ⽤来设置描述词组set中相关fd的位
void FD_ZERO(fd_set *set);// ⽤来清除描述词组set的全部位（2）timeval
timeval结构用于描述一段时间长度，如果在这个时间内，需要监视的描述符没有事件发生则函数返回，返回值为0。
select返回值：
执行成功则返回文件描述符状态已改变的个数；
如果返回0代表在文件描述符状态改变前已超过timeout时间，没有返回；
当有错误发生时则返回-1。
（3）原理
理解select模型的关键在于理解fd_set，为方便说明，取fd_set为1字节，fd_set中的每一个bit对应一个文件描述符fd，则1字节的fd_set对应8个fd。
执行FD_ZERO：
fd_set set;
FD_ZERO(&set);则set位图表示是0000,0000；
若fd=5，执行FD_SET：
FD_SET(fd,&set);set第5位将置为1，则set变为0001,0000；
若再加入fd=2、fd=1，则set变为0001,0011；
select(6,&set,0,0,0)执行select阻塞等待；
若fd=1、fd=2上都发生可读事件，则select返回，此时set将变为0000,0011，没有事件发生的fd=5将被清空。
（4）特点
select可监控的文件描述符个数取决于sizeof(fd_set)的值，将fd加入select监控集的同时，还需使用一个array保存放到select监控集中的fd，array的作用可概括为两个方面：
一是用于在select返回后，array作为源数据和fd_set进行FD_ISSET判断。
二是select返回后会把以前加入的但并无事件发生的fd清空，则每次开始select前都要重新从array取得fd逐一加入，扫描array的同时取得fd最大值maxfd，用于select的第一个参数。
（5）缺点
每次调用select，都需要手动设置fd集合，从接口使用角度来说不是很方便；
每次调用select，都需要把fd集合从用户态拷贝到内核态，这个开销在fd很多时会很大；
同时每次调用select都需要在内核遍历传递进来的所有fd，这个开销在fd很多时也会很大；
select支持的文件描述符数量太小。
（6）实现
下面实现一个基于select的TCP回显服务器，该服务器主要包括了网络通信、日志、线程同步、事件驱动等模块。
<1> common
common模块包含了服务器通用的头文件、错误码定义以及一个禁止拷贝的基类。
common.hpp
#ifndef _COMMON_HPP_
#define _COMMON_HPP_
#include<iostream>
#include<functional>
#include<unistd.h>
#include<string>
#include<cstring>
#include<sys/types.h>
#include<sys/socket.h>
#include<arpa/inet.h>
#include<netinet/in.h>
using namespace std;
enum exitcode
{
OK=0,
USAGE_ERR,
SOCKET_ERR,
BIND_ERR,
LISTEN_ERR,
CONNECT_ERR,
FORK_ERR,
OPEN_ERR
};
class nocopy
{
public:
nocopy(){}
nocopy(const nocopy& )=delete;
const nocopy& operator=(const nocopy& )=delete;
~nocopy(){}
};
#define CONV(addr) ((struct sockaddr*)&addr)
#endifexitcode枚举类型定义了一组程序退出时的状态码，每个枚举值对应一种可能的错误场景，nocopy为不可拷贝基类，通过将拷贝构造、赋值重载函数删除来禁止派生类拷贝，便于后续文件描述符这种唯一资源的管理，CONV为地址转换宏，用于将sockaddr_in强制类型转化为sockaddr。
<2> inetaddr
inetaddr为网络地址封装模块，用于将底层的sockaddr_in结构体与IP字符串和端口号进行统一管理，为上层网络提供便利的地址操作接口。
inetaddr.hpp
#ifndef _INETADDR_HPP_
#define _INETADDR_HPP_
#include"common.hpp"
using namespace std;
class inetaddr
{
public:
inetaddr(){}
inetaddr(struct sockaddr_in& addr)
{
setaddr(addr);
}
inetaddr(const string& ip,uint16_t port)
:_ip(ip)
,_port(port)
{
memset(&_addr,0,sizeof(_addr));
_addr.sin_family=AF_INET;
inet_pton(AF_INET,_ip.c_str(),&_addr.sin_addr);
_addr.sin_port=htons(_port);
}
inetaddr(uint16_t port)
:_port(port)
,_ip()
{
memset(&_addr,0,sizeof(_addr));
_addr.sin_family=AF_INET;
_addr.sin_port=htons(_port);
_addr.sin_addr.s_addr=INADDR_ANY;
}
void setaddr(struct sockaddr_in& addr)
{
_addr=addr;
_port=ntohs(_addr.sin_port);
char ipbuffer[64];
inet_ntop(AF_INET,&_addr.sin_addr,ipbuffer,sizeof(_addr));
_ip=ipbuffer;
}
uint16_t port()
{
return _port;
}
string ip()
{
return _ip;
}
const struct sockaddr_in& netaddr()
{
return _addr;
}
const struct sockaddr* netaddrptr()
{
return CONV(_addr);
}
socklen_t netaddrlen()
{
return sizeof(_addr);
}
bool operator==(const inetaddr& addr)
{
return _ip==addr._ip && _port==addr._port;
}
string StringAddr()
{
return _ip+":"+to_string(_port);
}
~inetaddr()
{}
private:
struct sockaddr_in _addr;
uint16_t _port;
string _ip;
};
#endifinetaddr通过多个构造函数来满足不同场景，无参构造函数用于创建一个空地址对象；接收sockaddr_in的构造函数从系统底层地址结构初始化；接收IP字符串和端口的构造函数用于连接远程服务器时指定目标地址；接收端口的构造函数用于服务器创建监听地址，IP设置为INADDR_ANY用于监听本机所有网络接口。setaddr负责在设置地址结构的同时自动填充字符串形式的IP、端口；port、ip分别返回端口号和IP字符串；netaddr返回sockaddr_in结构的常量引用；netaddrptr返回可用于系统调用的sockaddr指针；内部使用CONV完成类型转换；netaddrlen返回地址结构体长度；StringAddr返回IP：端口格式的字符串，便于日志打印和调试；operator==通过比较IP、端口来判断两个地址是否相同。
<3> mutex
mutex为线程同步模块，实现了对互斥锁的封装。
mutex.hpp
#pragma once
#include<iostream>
#include<pthread.h>
using namespace std;
namespace mutexmodule
{
class mutex
{
public:
mutex()
{
pthread_mutex_init(&_mutex,nullptr);
}
void lock()
{
int n=pthread_mutex_lock(&_mutex);
(void)n;
}
void unlock()
{
int n=pthread_mutex_unlock(&_mutex);
(void)n;
}
~mutex()
{
pthread_mutex_destroy(&_mutex);
}
private:
pthread_mutex_t _mutex;
};
class lockguard
{
public:
lockguard(mutex& mutex):_mutex(mutex)
{
_mutex.lock();
}
~lockguard()
{
_mutex.unlock();
}
private:
mutex& _mutex;
};
}mutex构造函数通过调用pthread_mutex_init对互斥锁进行初始化，lock、unlock分别调用pthread_mutex_lock、pthread_mutex_unlock进行加锁、解锁，析构函数调用pthread_mutex_destroy销毁锁。lockguard构造函数通过接收mutex对象调用lock进行加锁，析构函数自动调用unlock解锁。
<4> log
log是一个支持策略模式、线程安全的日志系统，定义了日志策略的抽象基类，并派生出控制台日志输出策略、文件日志输出策略。
log.hpp
#ifndef _LOG_HPP_
#define _LOG_HPP_
#include<iostream>
#include<string>
#include<cstdio>
#include<filesystem>
#include<fstream>
#include<sstream>
#include<memory>
#include<ctime>
#include<unistd.h>
#include"mutex.hpp"
using namespace std;
namespace logmodule
{
using namespace mutexmodule;
const string gsep ="\r\n";
class logstrategy
{
public:
virtual ~logstrategy()=default;
virtual void synclog(const string& message)=0;
};
class consolelogstrategy:public logstrategy
{
public:
consolelogstrategy(){}
void synclog(const string& message) override
{
lockguard guard(_mutex);
cout<<message<<gsep;
}
~consolelogstrategy(){}
private:
mutex _mutex;
};
const string defaultpath="/var/log/";
const string defaultfile="my.log";
class filelogstrategy:public logstrategy
{
public:
filelogstrategy(const string& path=defaultpath,const string& file=defaultfile)
:_path(path),
_file(file)
{
lockguard guard(_mutex);
if(filesystem::exists(_path))
{
return;
}
try
{
filesystem::create_directories(_path);
}
catch(const filesystem::filesystem_error& e)
{
cerr<<e.what()<<endl;
}
}
void synclog(const string& message) override
{
lockguard guard(_mutex);
string filename=_path+(_path.back()=='/'?"":"/")+_file;
ofstream out(filename,ios::app);
if(!out.is_open())
{
return;
}
out<<message<<gsep;
out.close();
}
~filelogstrategy(){}
private:
string _path;
string _file;
mutex _mutex;
};
enum class loglevel
{
DEBUG,
INFO,
WARNING,
ERROR,
FATAL
};
string levelstr(loglevel lev)
{
switch(lev)
{
case loglevel::DEBUG: return "DEBUG";
case loglevel::INFO: return "INFO";
case loglevel::WARNING: return "WARNING";
case loglevel::ERROR: return "ERROR";
case loglevel::FATAL: return "FATAL";
default: return "UNKNOWN";
}
}
string gettimestamp()
{
time_t t=time(nullptr);
struct tm curr_tm;
localtime_r(&t,&curr_tm);
char timebuffer[128];
snprintf(timebuffer, sizeof(timebuffer),"%4d-%02d-%02d %02d:%02d:%02d",
curr_tm.tm_year+1900,
curr_tm.tm_mon+1,
curr_tm.tm_mday,
curr_tm.tm_hour,
curr_tm.tm_min,
curr_tm.tm_sec
);
return timebuffer;
}
class logger
{
public:
logger():_ptr(nullptr)
{
enableconsolelogstrategy();
}
void enableconsolelogstrategy()
{
_ptr=make_unique<consolelogstrategy>();
}
void enablefilelogstrategy()
{
_ptr=make_unique<filelogstrategy>();
}
class logmessage
{
public:
logmessage(const string& src,loglevel level,int num,logger& logger)
:_pid(getpid())
,_src(src)
,_num(num)
,_curr_time(gettimestamp())
,_level(level)
,_logger(logger)
{
stringstream ss;
ss<<"["<<_curr_time<<"]"<<"["<<levelstr(_level)<<"]"<<"["<<_pid<<"]"<<"["<<_src<<"]"<<"["<<_num<<"]"<<"-";
_loginfo=ss.str();
}
template<class K>
logmessage& operator<<(const K& info)
{
stringstream ss;
ss<<info;
_loginfo+=ss.str();
return *this;
}
~logmessage()
{
if(_logger._ptr)
{
_logger._ptr->synclog(_loginfo);
}
}
private:
string _curr_time;
loglevel _level;
pid_t _pid;
string _src;
int _num;
string _loginfo;
logger& _logger;
};
logmessage operator()(loglevel lev,const string& name,int line)
{
return logmessage(name,lev,line,*this);
}
~logger()
{
}
private:
unique_ptr<logstrategy> _ptr;
};
static logger log;
#define LOG(level) logmodule::log(level,__FILE__,__LINE__)
#define Enable_Console_Log_Strategy() logmodule::log.enableconsolelogstrategy()
#define Enable_File_Log_Strategy() logmodule::log.enablefilelogstrategy()
}
#endifconsolelogstrategy控制台策略在输出时使用互斥锁保证线程安全，filelogstrategy文件策略则在构造时检查并创建日志目录，每次写入时以追加模式打开文件并加锁。日志级别定义了DEBUG、INFO、WARNING、ERROR、FATAL五个等级，并提供了等级转字符串的函数。gettimestamp函数用于获取当前时间的格式化字符串。logger类为日志器，内部持有一个策略对象的智能指针，默认使用控制台策略，也可通过enableconsolelogstrategy和enablefilelogstrategy动态切换策略。logger内部定义了一个嵌套类logmessage，logmessage构造函数通过接收文件名、日志级别、行号、logger引用，在构造时生成带有时间戳、进程ID、文件名、行号等信息的日志前缀；重载operator<<通过流式方式输入日志内容，析构时将日志内容通过logger持有的策略对象输出。logger类还重载了operator()，返回一个临时的logmessage对象，这样就可以实现通过LOG以流的方式输出日志。静态logger、宏，用于日志记录、并根据需要启用控制台、文件输出策略。
<5> socket
socket模块是对套接字的网络封装层，位于socketmodule命名空间中，为上层提供了面向对象的socket操作接口。
#ifndef _SOCKET_HPP_
#define _SOCKET_HPP_
#include<iostream>
#include<string>
#include<unistd.h>
#include<sys/types.h>
#include<sys/socket.h>
#include<netinet/in.h>
#include<arpa/inet.h>
#include<cstdlib>
#include"common.hpp"
#include"log.hpp"
#include"inetaddr.hpp"
using namespace std;
namespace socketmodule
{
using namespace logmodule;
const static int gbacklog=16;
class socket
{
public:
virtual ~socket(){}
virtual void socketordie()=0;
virtual void bindordie(uint16_t port)=0;
virtual void listenordie(int backlog)=0;
virtual void Close()=0;
virtual int recv(string* out)=0;
virtual int send(const string& message)=0;
virtual int accept(inetaddr* addr)=0;
virtual int connect(const string& ip,uint16_t port)=0;
virtual int fd()=0;
public:
void buildtcpsocket(uint16_t port,int backlog=gbacklog)
{
socketordie();
bindordie(port);
listenordie(backlog);
}
void buildclient()
{
socketordie();
}
};
const static int defaultfd=-1;
class tcpsocket:public socket
{
public:
tcpsocket()
:_sockfd(defaultfd)
{}
tcpsocket(int fd)
:_sockfd(fd)
{}
tcpsocket(const string& ip,uint16_t port)
{}
~tcpsocket()
{}
void socketordie() override
{
_sockfd=::socket(AF_INET,SOCK_STREAM,0);
if(_sockfd<0)
{
LOG(loglevel::FATAL)<<"socket error";
exit(SOCKET_ERR);
}
LOG(loglevel::INFO)<<"sock success";
}
void bindordie(uint16_t port) override
{
inetaddr addr(port);
int n=::bind(_sockfd,addr.netaddrptr(),addr.netaddrlen());
if(n<0)
{
LOG(loglevel::FATAL)<<"bind error";
exit(BIND_ERR);
}
LOG(loglevel::INFO)<<"bind success";
}
void listenordie(int backlog) override
{
int n=listen(_sockfd,backlog);
if(n<0)
{
LOG(loglevel::FATAL)<<"listen error";
exit(LISTEN_ERR);
}
LOG(loglevel::INFO)<<"listen success";
}
int accept(inetaddr* addr) override
{
struct sockaddr_in peer;
socklen_t len=sizeof(peer);
int fd=::accept(_sockfd,CONV(peer),&len);
if(fd<0)
{
LOG(loglevel::WARNING)<<"accept error";
return -1;
}
return fd;
}
int recv(string* out) override
{
char buffer[4096*2];
ssize_t n=::recv(_sockfd,buffer,sizeof(buffer)-1,0);
if(n>0)
{
buffer[n]=0;
*out+=buffer;
}
return n;
}
int send(const string& message) override
{
return ::send(_sockfd,message.c_str(),message.size(),0);
}
void Close() override
{
if(_sockfd>=0)
{
::close(_sockfd);
}
}
int connect(const string& ip,uint16_t port) override
{
inetaddr addr(ip,port);
return ::connect(_sockfd,addr.netaddrptr(),addr.netaddrlen());
}
int fd(){return _sockfd;}
private:
int _sockfd;
};
}
#endifsocket为抽象基类，声明了一系列的纯虚函数，socketordie用于创建套接字、bindordie用于绑定端口、listenordie用于监听、Close用于关闭、recv和send用于数据收发、accept用于接受连接、connect用于连接远程服务器、以及fd用于获取文件描述符。buildtcpsocket用于服务器依次执行创建、绑定和监听三步操作，buildclient用于客户端只创建套接字。tcpsocket类继承自socket，在tcpsocket中，socketordie调用系统socket函数创建TCP套接字，bindordie通过inetaddr类将端口转换为网络地址结构后调用bind；listenordie调用listen开始监听，backlog为默认连接数，accept负责接收客户端连接，返回新的套接字文件描述符，recv用于读取数据，将读取到的内容追加到传入的字符串参数中，send通过调用系统send进行消息的发送，Close用于关闭套接字，connect用于客户端连接服务器，通过inetaddr构建地址后调用connect。
<6> selectserver
selectserver模块是一个基于select实现的一个TCP服务器：
6.1 构造函数
#ifndef _SELECTSERVER_HPP_
#define _SELECTSERVER_HPP_
#include<iostream>
#include<unistd.h>
#include<memory>
#include"socket.hpp"
#include"log.hpp"
using namespace socketmodule;
using namespace logmodule;
using namespace std;
class selectserver
{
const static int size=sizeof(fd_set)*8;
const static int defaultfd=-1;
public:
selectserver(int port)
:_listensock(make_unique<tcpsocket>())
,_isrunning(false)
{
_listensock->buildtcpsocket(port);
for(int i=0;i<size;i++)
fdarr[i]=defaultfd;
fdarr[0]=_listensock->fd();
}
void stop()
{
_isrunning=false;
}
~selectserver()
{}
private:
unique_ptr<socketmodule::socket> _listensock;
bool _isrunning;
int fdarr[size];
};
#endif_listensock(make_unique<tcpserver>())，_listensock->buildtcpsocket(port)，构造函数selectserver通过接收一个端口号，通过make_unique创建tcpsocket对象并调用buildtcpsocket完成套接字的创建、绑定和监听。fdarr[i]=defaultfd，遍历fdarr数组将所有槽位设为defaultfd，fdarr[0]=_listensock->fd()，将监听套接字的文件描述符存入数组的第一个位置。
6.2 printfd
void printfd()
{
cout<<"fdarr[]: ";
for(int i=0;i<size;i++)
{
if(fdarr[i]==defaultfd)
continue;
cout<<fdarr[i]<<" ";
}
cout<<"\r\n";
}printfd用于调试输出，通过for循环遍历fdarr数组，if(fdarr[i]==defaultfd) continue，跳过无效槽位，cout<<fdarr[i]<<" "，输出所有有效的文件描述符，cout<<"\r\n"，最后输出换行。
6.3 accepter
void accepter()
{
inetaddr addr;
int sockfd=_listensock->accept(&addr);
if(sockfd>=0)
{
LOG(loglevel::INFO)<<"get a new link,sockfd: "<<sockfd;
int pos=0;
for(;pos<size;pos++)
{
if(fdarr[pos]==defaultfd)
break;
}
if(pos==size)
{
LOG(loglevel::WARNING)<<"server full";
close(sockfd);
}
else
{
fdarr[pos]=sockfd;
}
}
}accepter负责接收新连接，int sockfd=_listensock->accept(&addr)，调用accept接收新连接，返回新的套接字文件描述符，通过for循环从fdarr的起始位置寻找第一个空闲位置，fdarr[pos]=sockfd，并将sockfd存入到fdarr中。
6.4 recver
void recver(int fd,int pos)
{
char buffer[1024];
ssize_t n=recv(fd,buffer,sizeof(buffer)-1,0);
if(n>0)
{
buffer[n]=0;
cout<<"client say@: "<<buffer<<endl;
}
else if(n==0)
{
LOG(loglevel::INFO)<<"client quit...";
fdarr[pos]=defaultfd;
close(fd);
}
else
{
LOG(loglevel::ERROR)<<"recv error";
}
}recver接收文件描述符和它在数组中的索引，char buffer[1024]，buffer为字符缓冲区，ssize_t n=recv(fd,buffer,sizeof(buffer)-1,0)，调用recv读取数据，若n>0，则读到数据，buffer[n]=0，在buffer末尾添加字符串结束符后输出到控制台，若n==0，则表示客户端关闭连接，fdarr[pos]=defaultfd，close(fd)，将数组对应位置的槽位设为defaultfd，并关闭套接字，如果n<0，LOG(loglevel::ERROR)<<"recv error"；记录错误日志。
6.5 handlerevent
void handlerevent(fd_set& fdset)
{
for(int i=0;i<size;i++)
{
if(fdarr[i]==defaultfd)
continue;
if(FD_ISSET(fdarr[i],&fdset))
{
if(fdarr[i]==_listensock->fd())
{
accepter();
}
else
{
recver(fdarr[i],i);
}
}
}
}handlerevent接收一个fd_set引用，用于处理就绪的事件。通过for循环遍历fdarr，跳过无效槽位，FD_ISSET(fdarr[i],&fdset)，对每个有效文件描述符调用FD_ISSET判断是否就绪。fdarr[i]==_listensock->fd()，如果就绪的描述符等于监听套接字的描述符，则调用accepter接收新连接；recver(fdarr[i],i)，else则调用recver处理客户端数据。
6.6 start
void start()
{
_isrunning=true;
while(_isrunning)
{
fd_set fset;
FD_ZERO(&fset);
int maxfd=defaultfd;
for(int i=0;i<size;i++)
{
if(fdarr[i]==defaultfd)
continue;
FD_SET(fdarr[i],&fset);
if(maxfd<fdarr[i])
{
maxfd=fdarr[i];
}
}
printfd();
int n=select(maxfd+1,&fset,nullptr,nullptr,nullptr);
switch(n)
{
case -1:
LOG(loglevel::ERROR)<<"select error";
break;
case 0:
LOG(loglevel::INFO)<<"time out...";
break;
default:
LOG(loglevel::DEBUG)<<"有事件就绪了...,n: "<<n;
handlerevent(fset);
break;
}
}
_isrunning=false;
}start是服务器的主循环，fd_set fset，FD_ZERO(&fset)，声明一个fd_set类型的集合变量fset并调用FD_ZERO清空，通过for循环遍历fdarr，FD_SET(fdarr[i]，&fset)，将有效的文件描述符添加到fset集合中，maxfd=fdarr[i]，同时更新maxfd为当前最大值。printfd()，调用printfd输出当前所有有效的文件描述符。int n=select(maxfd+1,&fset,nullptr,nullptr)，调用select，传入maxfd+1、fset集合，只监听读事件且阻塞等待，select根据返回值n通过switch语句进行处理，当有事件就绪了，n为就绪事件的数量，调用handlerevent对就绪事件进行处理。
<7> main
main模块为整个select服务器的入口点：
#include"selectserver.hpp"
int main(int argc,char*argv[])
{
if(argc!=2)
{
cout<<"usage:"<<argv[0]<<"port"<<endl;
exit(USAGE_ERR);
}
Enable_Console_Log_Strategy();
uint16_t port=stoi(argv[1]);
unique_ptr<selectserver> up=make_unique<selectserver>(port);
up->start();
return 0;
}Enable_Console_Log_Strategy()，启用控制台日志策略，uint16_t port=stoi(argv[1])，将命令行第二个参数转换为端口号，unique_ptr<selectserver> up=make_unique<selectserver>(port)，通过端口号构造selectserver，up->start()，调用start启动服务器。
Makefile:
selectserver:main.cc
g++ -o $@ $^ -std=c++17
.PHONY:clean
clean:
rm -rf selectserver通过Makefile即可实现一键化编译，g++ -o $@ $^ -std=c++17，采用c++17标准，服务器运行如下所示：
Makefile编译通过后，./selectserver 8922，绑定端口号8922，服务器开始运行，打开另一个终端，telnet 127.0.0.1 8922，进行连接，客户端向服务器输入消息，服务器可回显相应内容，结果如下所示：
客户端：
服务器：
2、poll
（1）原理
poll也是Linux I/O多路复用的一种机制，可以用于同时监控多个文件描述符的事件。与select类似，但克服了select的一些限制。
poll接口如下：
fds是一个poll函数监听的结构列表，pollfd结构包含了三部分内容，如下所示：
//pollfd结构
struct pollfd
{
int fd;
short events;
short revents;
};fd为文件描述符，events为监听的事件集合，revents为返回的事件集合。
events、revents的取值如下：
nfds表示fds数组的长度，timeout表示poll函数的超时时间，以毫秒为单位。
poll返回值大于0，表示poll由于监听的文件描述符就绪而返回；
poll返回值等于0，表示poll等待超时；
poll返回值小于0，表示出错。
（2）优点
不同于select使用三个位图来表示三个fdset的方式，poll使用一个pollfd的指针实现；
pollfd结构包含了要监视的event和发生的event，不再使用select参数-值传递的方式，接口使用比select更加方便；
poll并没有文件描述符最大数量限制。
（3）缺点
当poll中监听的文件描述符数目增多时：
和select一样，poll返回后，需要轮询pollfd来获取就绪的文件描述符；
每次调用poll都需要把大量的pollfd结构从用户态拷贝到内核中；
同时连接的大量客户端在同一时刻可能只有很少的处于就绪状态，因此随着监视的文件描述符数量增多，poll的效率也会随之线性下降。
（4）实现
将上面实现的select TCP回显服务器稍加修改，即可改造成一个基于poll的TCP服务器。
<1> common
common.hpp
#ifndef _COMMON_HPP_
#define _COMMON_HPP_
#include<iostream>
#include<functional>
#include<unistd.h>
#include<string>
#include<cstring>
#include<sys/types.h>
#include<sys/socket.h>
#include<arpa/inet.h>
#include<netinet/in.h>
using namespace std;
enum exitcode
{
OK=0,
USAGE_ERR,
SOCKET_ERR,
BIND_ERR,
LISTEN_ERR,
CONNECT_ERR,
FORK_ERR,
OPEN_ERR
};
class nocopy
{
public:
nocopy(){}
nocopy(const nocopy& )=delete;
const nocopy& operator=(const nocopy& )=delete;
~nocopy(){}
};
#define CONV(addr) ((struct sockaddr*)&addr)
#endifcommon模块与上面select服务器实现一致，提供了退出状态码枚举、不可拷贝基类和socket地址转化宏，为整个服务器提供通用的基础设施。
<2> inetaddr
inetaddr.hpp
#ifndef _INETADDR_HPP_
#define _INETADDR_HPP_
#include"common.hpp"
using namespace std;
class inetaddr
{
public:
inetaddr(){}
inetaddr(struct sockaddr_in& addr)
{
setaddr(addr);
}
inetaddr(const string& ip,uint16_t port)
:_ip(ip)
,_port(port)
{
memset(&_addr,0,sizeof(_addr));
_addr.sin_family=AF_INET;
inet_pton(AF_INET,_ip.c_str(),&_addr.sin_addr);
_addr.sin_port=htons(_port);
}
inetaddr(uint16_t port)
:_port(port)
,_ip()
{
memset(&_addr,0,sizeof(_addr));
_addr.sin_family=AF_INET;
_addr.sin_port=htons(_port);
_addr.sin_addr.s_addr=INADDR_ANY;
}
void setaddr(struct sockaddr_in& addr)
{
_addr=addr;
_port=ntohs(_addr.sin_port);
char ipbuffer[64];
inet_ntop(AF_INET,&_addr.sin_addr,ipbuffer,sizeof(_addr));
_ip=ipbuffer;
}
uint16_t port()
{
return _port;
}
string ip()
{
return _ip;
}
const struct sockaddr_in& netaddr()
{
return _addr;
}
const struct sockaddr* netaddrptr()
{
return CONV(_addr);
}
socklen_t netaddrlen()
{
return sizeof(_addr);
}
bool operator==(const inetaddr& addr)
{
return _ip==addr._ip && _port==addr._port;
}
string StringAddr()
{
return _ip+":"+to_string(_port);
}
~inetaddr()
{}
private:
struct sockaddr_in _addr;
uint16_t _port;
string _ip;
};
#endifinetaddr模块与select服务器实现一致，对网络地址进行了封装，支持从IP、端口或仅从端口构造，并能返回原生地址结构和IP字符串。
<3> mutex
mutex.hpp
#pragma once
#include<iostream>
#include<pthread.h>
using namespace std;
namespace mutexmodule
{
class mutex
{
public:
mutex()
{
pthread_mutex_init(&_mutex,nullptr);
}
void lock()
{
int n=pthread_mutex_lock(&_mutex);
(void)n;
}
void unlock()
{
int n=pthread_mutex_unlock(&_mutex);
(void)n;
}
~mutex()
{
pthread_mutex_destroy(&_mutex);
}
private:
pthread_mutex_t _mutex;
};
class lockguard
{
public:
lockguard(mutex& mutex):_mutex(mutex)
{
_mutex.lock();
}
~lockguard()
{
_mutex.unlock();
}
private:
mutex& _mutex;
};
}mutex模块与select服务器实现一致，对互斥锁进行了封装，提供了RAII风格的lockguard类。
<4> log
log.hpp
#ifndef _LOG_HPP_
#define _LOG_HPP_
#include<iostream>
#include<string>
#include<cstdio>
#include<filesystem>
#include<fstream>
#include<sstream>
#include<memory>
#include<ctime>
#include<unistd.h>
#include"mutex.hpp"
using namespace std;
namespace logmodule
{
using namespace mutexmodule;
const string gsep ="\r\n";
class logstrategy
{
public:
virtual ~logstrategy()=default;
virtual void synclog(const string& message)=0;
};
class consolelogstrategy:public logstrategy
{
public:
consolelogstrategy(){}
void synclog(const string& message) override
{
lockguard guard(_mutex);
cout<<message<<gsep;
}
~consolelogstrategy(){}
private:
mutex _mutex;
};
const string defaultpath="/var/log/";
const string defaultfile="my.log";
class filelogstrategy:public logstrategy
{
public:
filelogstrategy(const string& path=defaultpath,const string& file=defaultfile)
:_path(path),
_file(file)
{
lockguard guard(_mutex);
if(filesystem::exists(_path))
{
return;
}
try
{
filesystem::create_directories(_path);
}
catch(const filesystem::filesystem_error& e)
{
cerr<<e.what()<<endl;
}
}
void synclog(const string& message) override
{
lockguard guard(_mutex);
string filename=_path+(_path.back()=='/'?"":"/")+_file;
ofstream out(filename,ios::app);
if(!out.is_open())
{
return;
}
out<<message<<gsep;
out.close();
}
~filelogstrategy(){}
private:
string _path;
string _file;
mutex _mutex;
};
enum class loglevel
{
DEBUG,
INFO,
WARNING,
ERROR,
FATAL
};
string levelstr(loglevel lev)
{
switch(lev)
{
case loglevel::DEBUG: return "DEBUG";
case loglevel::INFO: return "INFO";
case loglevel::WARNING: return "WARNING";
case loglevel::ERROR: return "ERROR";
case loglevel::FATAL: return "FATAL";
default: return "UNKNOWN";
}
}
string gettimestamp()
{
time_t t=time(nullptr);
struct tm curr_tm;
localtime_r(&t,&curr_tm);
char timebuffer[128];
snprintf(timebuffer, sizeof(timebuffer),"%4d-%02d-%02d %02d:%02d:%02d",
curr_tm.tm_year+1900,
curr_tm.tm_mon+1,
curr_tm.tm_mday,
curr_tm.tm_hour,
curr_tm.tm_min,
curr_tm.tm_sec
);
return timebuffer;
}
class logger
{
public:
logger():_ptr(nullptr)
{
enableconsolelogstrategy();
}
void enableconsolelogstrategy()
{
_ptr=make_unique<consolelogstrategy>();
}
void enablefilelogstrategy()
{
_ptr=make_unique<filelogstrategy>();
}
class logmessage
{
public:
logmessage(const string& src,loglevel level,int num,logger& logger)
:_pid(getpid())
,_src(src)
,_num(num)
,_curr_time(gettimestamp())
,_level(level)
,_logger(logger)
{
stringstream ss;
ss<<"["<<_curr_time<<"]"<<"["<<levelstr(_level)<<"]"<<"["<<_pid<<"]"<<"["<<_src<<"]"<<"["<<_num<<"]"<<"-";
_loginfo=ss.str();
}
template<class K>
logmessage& operator<<(const K& info)
{
stringstream ss;
ss<<info;
_loginfo+=ss.str();
return *this;
}
~logmessage()
{
if(_logger._ptr)
{
_logger._ptr->synclog(_loginfo);
}
}
private:
string _curr_time;
loglevel _level;
pid_t _pid;
string _src;
int _num;
string _loginfo;
logger& _logger;
};
logmessage operator()(loglevel lev,const string& name,int line)
{
return logmessage(name,lev,line,*this);
}
~logger()
{
}
private:
unique_ptr<logstrategy> _ptr;
};
static logger log;
#define LOG(level) logmodule::log(level,__FILE__,__LINE__)
#define Enable_Console_Log_Strategy() logmodule::log.enableconsolelogstrategy()
#define Enable_File_Log_Strategy() logmodule::log.enablefilelogstrategy()
}
#endiflog日志模块与select服务器实现一致，采用策略模式，控制台策略、文件策略都通过互斥锁保证线程安全，logger类通过宏提供流式日志输出，每条日志都包含时间戳、日志级别、进程ID、源文件名和行号。
<5> socket
#ifndef _SOCKET_HPP_
#define _SOCKET_HPP_
#include<iostream>
#include<string>
#include<unistd.h>
#include<sys/types.h>
#include<sys/socket.h>
#include<netinet/in.h>
#include<arpa/inet.h>
#include<cstdlib>
#include"common.hpp"
#include"log.hpp"
#include"inetaddr.hpp"
using namespace std;
namespace socketmodule
{
using namespace logmodule;
const static int gbacklog=16;
class socket
{
public:
virtual ~socket(){}
virtual void socketordie()=0;
virtual void bindordie(uint16_t port)=0;
virtual void listenordie(int backlog)=0;
virtual void Close()=0;
virtual int recv(string* out)=0;
virtual int send(const string& message)=0;
virtual int accept(inetaddr* addr)=0;
virtual int connect(const string& ip,uint16_t port)=0;
virtual int fd()=0;
public:
void buildtcpsocket(uint16_t port,int backlog=gbacklog)
{
socketordie();
bindordie(port);
listenordie(backlog);
}
void buildclient()
{
socketordie();
}
};
const static int defaultfd=-1;
class tcpsocket:public socket
{
public:
tcpsocket()
:_sockfd(defaultfd)
{}
tcpsocket(int fd)
:_sockfd(fd)
{}
tcpsocket(const string& ip,uint16_t port)
{}
~tcpsocket()
{}
void socketordie() override
{
_sockfd=::socket(AF_INET,SOCK_STREAM,0);
if(_sockfd<0)
{
LOG(loglevel::FATAL)<<"socket error";
exit(SOCKET_ERR);
}
LOG(loglevel::INFO)<<"sock success";
}
void bindordie(uint16_t port) override
{
inetaddr addr(port);
int n=::bind(_sockfd,addr.netaddrptr(),addr.netaddrlen());
if(n<0)
{
LOG(loglevel::FATAL)<<"bind error";
exit(BIND_ERR);
}
LOG(loglevel::INFO)<<"bind success";
}
void listenordie(int backlog) override
{
int n=listen(_sockfd,backlog);
if(n<0)
{
LOG(loglevel::FATAL)<<"listen error";
exit(LISTEN_ERR);
}
LOG(loglevel::INFO)<<"listen success";
}
int accept(inetaddr* addr) override
{
struct sockaddr_in peer;
socklen_t len=sizeof(peer);
int fd=::accept(_sockfd,CONV(peer),&len);
if(fd<0)
{
LOG(loglevel::WARNING)<<"accept error";
return -1;
}
return fd;
}
int recv(string* out) override
{
char buffer[4096*2];
ssize_t n=::recv(_sockfd,buffer,sizeof(buffer)-1,0);
if(n>0)
{
buffer[n]=0;
*out+=buffer;
}
return n;
}
int send(const string& message) override
{
return ::send(_sockfd,message.c_str(),message.size(),0);
}
void Close() override
{
if(_sockfd>=0)
{
::close(_sockfd);
}
}
int connect(const string& ip,uint16_t port) override
{
inetaddr addr(ip,port);
return ::connect(_sockfd,addr.netaddrptr(),addr.netaddrlen());
}
int fd(){return _sockfd;}
private:
int _sockfd;
};
}
#endifsocket模块与select服务器实现一致，定义了抽象基类socket，tcpsocket实现了具体的socket操作。
<6> pollserver
pollserver模块是整个服务器的核心，实现了一个基于poll的TCP服务器，该模块负责管理监听套接字和所有客户端连接，通过poll系统调用同时监控多个文件描述符上的读事件。
pollserver.hpp
6.1 构造函数
#ifndef _POLLSERVER_HPP_
#define _POLLSERVER_HPP_
#include<iostream>
#include<unistd.h>
#include<memory>
#include<sys/poll.h>
#include"socket.hpp"
#include"log.hpp"
using namespace socketmodule;
using namespace logmodule;
using namespace std;
class pollserver
{
const static int size=4096;
const static int defaultfd=-1;
public:
pollserver(int port)
:_listensock(make_unique<tcpsocket>())
,_isrunning(false)
{
_listensock->buildtcpsocket(port);
for(int i=0;i<size;i++)
{
fds[i].fd=defaultfd;
fds[i].events=0;
fds[i].revents=0;
}
fds[0].fd=_listensock->fd();
fds[0].events=POLLIN;
}
void stop()
{
_isrunning=false;
}
~pollserver()
{}
private:
unique_ptr<socketmodule::socket> _listensock;
bool _isrunning;
struct pollfd fds[size];
};
#endif构造函数pollserver接收一个端口号作为参数，_listensock(make_unique<tcpsocket>())，通过make_unique创建一个tcpsocket对象，_listensock->buildtcpsocket(port)，调用buildtcpsocket完成套接字的创建、绑定和监听操作，通过for循环遍历pollfd，fds[0].fd=_listensock->fd()，fds[0].events=POLLIN，将数组第一个元素的fd设为监听套接字的文件描述符，并设置其events为POLLIN，关心该文件描述符的可读事件。
6.2 printfd
void printfd()
{
cout<<"fds[]: ";
for(int i=0;i<size;i++)
{
if(fds[i].fd==defaultfd)
continue;
cout<<fds[i].fd<<" ";
}
cout<<"\r\n";
}printfd通过for循环遍历pollfd，cout<<"fds[i].fd<<" "；输出所有有效文件描述符的编号，便于观察当前服务器管理的连接情况。
6.3 accepter
void accepter()
{
inetaddr addr;
int sockfd=_listensock->accept(&addr);
if(sockfd>=0)
{
LOG(loglevel::INFO)<<"get a new link,sockfd: "<<sockfd;
int pos=0;
for(;pos<size;pos++)
{
if(fds[pos].fd==defaultfd)
break;
}
if(pos==size)
{
LOG(loglevel::WARNING)<<"server full";
close(sockfd);
}
else
{
fds[pos].fd=sockfd;
fds[pos].events=POLLIN;
fds[pos].revents=0;
}
}
}accepter负责接收客户端连接，int sockfd=_listensock->accept(&addr)，调用_listensock的accept接收连接，通过for循环寻找第一个空闲位置，fds[pos].fd=sockfd，fds[pos].events=POLLIN，fds[pos].revents=0，如果成功找到，则将新的套接字文件描述符放入该槽位，将events设为POLLIN，revents清零；if(pos==size)，如果数组已满，LOG(loglevel::WARNING)<<"server full"；close(sockfd)，则记录日志并关闭新连接。
6.4 recver
void recver(int pos)
{
char buffer[1024];
ssize_t n=recv(fds[pos].fd,buffer,sizeof(buffer)-1,0);
if(n>0)
{
buffer[n]=0;
cout<<"client say@: "<<buffer<<endl;
}
else if(n==0)
{
LOG(loglevel::INFO)<<"client quit...";
fds[pos].fd=defaultfd;
fds[pos].events=0;
fds[pos].revents=0;
close(fds[pos].fd);
}
else
{
LOG(loglevel::ERROR)<<"recv error";
fds[pos].fd=defaultfd;
fds[pos].events=0;
fds[pos].revents=0;
close(fds[pos].fd);
}
}recver通过接收一个数组索引作为参数，表示要处理哪个客户端连接。char buffer[1024]，ssize_t n=recv(fds[pos].fd，buffer，sizeof(buffer)-1,0)，通过调用recv从对应的套接字读取数据。若n>0，表示成功读取到数据，buffer[n]=0，在缓冲区末尾添加字符串结束符后输出到控制台。若n==0，表示客户端主动关闭连接，fds[pos].fd=defaultfd，fds[pos].events=0，fds[pos].revents=0，close(fds[pos].fd)，将对应槽位的fd设为defaultfd，events和revents清零，关闭套接字。如果n<0，表示读取错误，记录错误日志，并执行相同的清理操作。
6.5 handlerevent
void handlerevent()
{
for(int i=0;i<size;i++)
{
if(fds[i].fd==defaultfd)
continue;
if(fds[i].revents & POLLIN)
{
if(fds[i].fd==_listensock->fd())
{
accepter();
}
else
{
recver(i);
}
}
}
}handlerevent用于处理所有就绪的事件，通过for循环遍历pollfd，if(fds[i].fd==defaultfd)，continue跳过无效槽位，if(fds[i].revents & POLLIN)，对每个有效的文件描述符，检查其revents是否包含POLLIN事件。if(fds[i].fd==_listensock->fd())，accepter()，如果当前就绪的文件描述符为监听套接字，则调用accepter接收新连接，recver(i)，else则调用recver处理客户端数据。
6.6 start
void start()
{
int timeout=-1;
_isrunning=true;
while(_isrunning)
{
int n=poll(fds,size,timeout);
switch(n)
{
case -1:
LOG(loglevel::ERROR)<<"poll error";
break;
case 0:
LOG(loglevel::INFO)<<"poll time out...";
break;
default:
LOG(loglevel::DEBUG)<<"有事件就绪了...,n: "<<n;
handlerevent();
break;
}
}
_isrunning=false;
}start是服务器的主循环，int timeout=-1，将超时时间设为-1表示无限等待，通过while循环，int n=poll(fds,size,timeout)，调用poll传入整个fds数组、数组大小和超时时间。通过switch对返回值n进行处理，当n==-1时表示出错，LOG(loglevel::ERROR)<<"poll error"，记录错误日志；n==0表示超时，LOG(loglevel::INFO)<<"poll time out..."，记录超时信息；当n>0时表示有事件就绪，handlerevent()，调用handlerevent处理就绪事件。
<7> main
main模块为整个pollserver服务器的入口点：
#include"pollserver.hpp"
int main(int argc,char*argv[])
{
if(argc!=2)
{
cout<<"usage:"<<argv[0]<<"port"<<endl;
exit(USAGE_ERR);
}
Enable_Console_Log_Strategy();
uint16_t port=stoi(argv[1]);
unique_ptr<pollserver> up=make_unique<pollserver>(port);
up->start();
return 0;
}Enable_Console_Log_Strategy()，启用控制台日志策略，uint16_t port=stoi(argv[1])，将命令行第二个参数转换为端口号，unique_ptr<pollserver> up=make_unique<pollserver>(port)，通过make_unique创建pollserver对象，up->start()，调用start启动服务器。
Makefile:
pollserver:main.cc
g++ -o $@ $^ -std=c++17
.PHONY:clean
clean:
rm -rf pollserver通过Makefile即可实现一键化编译，g++ -o $@ $^ -std=c++17，采取c++17标准，服务器运行如下所示：
Makefile编译通过后，./pollserver 8923，绑定端口号8923，服务器开始运行，打开另一个终端，telnet 127.0.0.1 8923，进行连接，客户端向服务器发送消息，服务器可回显相应内容，结果如下所示：
客户端：
服务器：
3、epoll
epoll被公认为Linux下性能最好的多路I/O就绪通知方法，epoll几乎具备了之前所说的一切优点，epoll是为处理大批量句柄而作了改进的poll。
（1）接口
epoll有3个相关的系统调用：
<1> epoll_create
epoll_create用于创建一个epoll的句柄，从Linux2.6.8之后，size参数被忽略，调用完epoll_create之后，必须调用close()关闭。
<2> epoll_ctl
epoll_ctl为epoll的事件注册函数，它不同于select是在监听事件时告诉内核要监听什么类型的事件，而是在这里先注册要监听的事件类型。
第一个参数epfd是epoll_create的返回值；
第二个参数op表示动作，用三个宏表示，取值如下：
EPOLL_CTL_ADD：注册新的fd到epfd中；
EPOLL_CTL_MOD：修改已经注册的fd的监听事件；
EPOLL_CTL_DEL：从epfd中删除一个fd；
第三个参数fd是需要监听的fd；
第四个参数event告诉内核需要监听什么事件。
struct epoll_event的结构如下：
events的取值如下：
EPOLLIN：表示对应的文件描述符可以读；
EPOLLOUT：表示对应的文件描述符可以写；
EPOLLPRI：表示对应的文件描述符有紧急的数据可读；
EPOLLERR：表示对应的文件描述符发生错误；
EPOLLHUP：表示对应的文件描述符被挂断；
EPOLLET：将EPOLL设为边缘触发模式；
EPOLLONESHOT：只监听一次事件，当监听完这次事件之后，如果还需要继续监听这个socket，需要再次把这个socket加入到EPOLL红黑树里。
<3> epoll_wait
epoll_wait用于收集在epoll监控的事件中已经发送的事件。参数events是分配好的epoll_event结构体数组，epoll将会把发生的事件赋值到events数组中，events不能为空；maxevents用于告诉内核这个events有多大，maxevents不能大于epoll_create的size；参数timeout表示超时时间，以毫秒为单位，0会立即返回，-1表示永久阻塞；如果函数调用成功，返回对应I/O上已准备好的文件描述符数目，返回0表示已超时，返回小于0表示函数调用失败。
（2）原理
当某一进程调用epoll_create时，内核会创建一个eventpoll结构体，这个结构体中有两个成员与epoll的使用方式密切相关。
struct eventpoll{
....
/*红⿊树的根节点，这颗树中存储着所有添加到epoll中的需要监控的事件*/
struct rb_root rbr;
/*双链表中则存放着将要通过epoll_wait返回给⽤⼾的满⾜条件的事件*/
struct list_head rdlist;
....
};每个epoll对象都有一个独立的eventpoll结构体，用于存放通过epoll_ctl向epoll对象中添加进来的事件。这些事件都会被挂载在红黑树中，如此，重复添加的事件就可以通过红黑树而高效的识别出来
而所有添加到epoll中的事件都会与设备、驱动程序建立回调关系，当响应的事件发生时会调用这个回调方法。这个回调方法在内核中称为ep_poll_callback，它会将发生的事件添加到rdlist双链表中。在epoll中，对于每一个事件，都会建立一个epitem结构体。
struct epitem{
struct rb_node rbn;//红⿊树节点
struct list_head rdllink;//双向链表节点
struct epoll_filefd ffd; //事件句柄信息
struct eventpoll *ep; //指向其所属的eventpoll对象
struct epoll_event event; //期待发⽣的事件类型
}当调用epoll_wait检查是否有事件发生时，只需要检查eventpoll对象中的rdlist双链表中是否有epitem元素即可；如果rdlist不为空，则把发生的事件复制到用户态，同时将事件数量返回给用户，时间复杂度为O(1)。
epoll的使用就是三部曲：
调用epoll_create创建一个epoll句柄；
调用epoll_ctl，将要监控的文件描述符进行注册；
调用epoll_wait，等待文件描述符就绪。
（3）优点
接口使用方便：虽然拆分成了三个函数，但使用起来更方便高效，不需要每次循环都设置关注的文件描述符，也做到了输入输出参数分离开；
数据拷贝轻量：只在合适的时候调用EPOLL_CTL_ADD将文件描述符结构拷贝到内核中，这个操作并不频繁，而select/poll都是每次循环都需要进行拷贝；
事件回调机制：避免使用遍历，而是使用回调函数的方式，将就绪的文件描述符结构加入到就绪队列中，epoll_wait返回直接访问就绪队列就知道哪些文件描述符就绪，时间复杂度为O(1)，即使文件描述符数目很多，效率也不会受到影响；
没有数量限制：文件描述符数目无上限。
（4）工作方式
epoll有2种工作方式：水平触发LT和边缘触发ET。
<1> LT
epoll默认状态下就是LT工作模式：
当epoll检测到socket上事件就绪时，可以不立刻处理，或者只处理一部分。
LT支持阻塞读写和非阻塞读写。
<2> ET
将socket添加到epoll文件描述符的时候使用了EPOLLET标志，epoll将进入ET工作模式。
当epoll检测到socket上事件就绪时，必须立刻处理；
ET模式下，文件描述符上的事件就绪后，只有一次处理机会；
ET的性能相比LT更高，epoll_wait的返回次数会少很多；
ET只支持非阻塞的读写；
使用ET模式的epoll，需要将文件描述符设置为非阻塞。
（5）应用
对于多连接，且多连接中只有一部分连接比较活跃时，比较适合使用epoll；
如一个需要处理上万个客户端的服务器，各种互联网APP的入口服务器，这样的服务器适合使用epoll。
二、Reactor
Reactor是一种事件驱动的编程模式，用于高效地处理并发I/O操作。它通过一个或多个事件循环来监听和处理各种事件，从而实现高效的并发处理，而无需为每个连接创建一个线程或进程。
事件循环：事件循环是异步编程的核心，负责监听事件并触发相应的回调函数。它通常是一个单线程的执行模型，通过多路复用技术select/poll/epoll高效地管理多个I/O操作。
三、Epoll+Reactor实现
下面实现一个基于epoll的Reactor高性能TCP服务器，该服务器主要包含了网络通信、日志、线程同步、事件驱动等模块。
1、common
common模块是整个服务器的基础，包含了服务器通用的头文件、错误码以及一个禁止拷贝的基类等。
common.hpp
#ifndef _COMMON_HPP_
#define _COMMON_HPP_
#include<iostream>
#include<functional>
#include<fcntl.h>
#include<unistd.h>
#include<string>
#include<cstring>
#include<sys/types.h>
#include<sys/socket.h>
#include<arpa/inet.h>
#include<netinet/in.h>
using namespace std;
enum exitcode
{
OK=0,
USAGE_ERR,
SOCKET_ERR,
BIND_ERR,
LISTEN_ERR,
CONNECT_ERR,
FORK_ERR,
OPEN_ERR,
EPOLL_CREATE_ERR,
EPOLL_CTL_ERR
};
class nocopy
{
public:
nocopy(){}
nocopy(const nocopy& )=delete;
const nocopy& operator=(const nocopy& )=delete;
~nocopy(){}
};
int defaultport=8915;
void setnonblock(int fd)
{
int fl=fcntl(fd,F_GETFL);
if(fl<0)
{
return;
}
fcntl(fd,F_SETFL,fl | O_NONBLOCK);
}
#define CONV(addr) ((struct sockaddr*)&addr)
#endif程序退出时的状态码枚举exitcode、不可拷贝的基类nocopy、以及sockaddr类型转换宏与前面实现的select/poll服务器一致。int fl=fcntl(fd，F_GETFL)，setnonblock通过fcntl获取当前文件状态标志，fcntl(fd,F_SETFL,fl | O_NONBLOCK)，设置O_NONBLOCK将套接字变为非阻塞模式。
2、inetaddr
inetaddr.hpp
#ifndef _INETADDR_HPP_
#define _INETADDR_HPP_
#include"common.hpp"
using namespace std;
class inetaddr
{
public:
inetaddr(){}
inetaddr(struct sockaddr_in& addr)
{
setaddr(addr);
}
inetaddr(const string& ip,uint16_t port)
:_ip(ip)
,_port(port)
{
memset(&_addr,0,sizeof(_addr));
_addr.sin_family=AF_INET;
inet_pton(AF_INET,_ip.c_str(),&_addr.sin_addr);
_addr.sin_port=htons(_port);
}
inetaddr(uint16_t port)
:_port(port)
,_ip()
{
memset(&_addr,0,sizeof(_addr));
_addr.sin_family=AF_INET;
_addr.sin_port=htons(_port);
_addr.sin_addr.s_addr=INADDR_ANY;
}
void setaddr(struct sockaddr_in& addr)
{
_addr=addr;
_port=ntohs(_addr.sin_port);
char ipbuffer[64];
inet_ntop(AF_INET,&_addr.sin_addr,ipbuffer,sizeof(_addr));
_ip=ipbuffer;
}
uint16_t port()
{
return _port;
}
string ip()
{
return _ip;
}
const struct sockaddr_in& netaddr()
{
return _addr;
}
const struct sockaddr* netaddrptr()
{
return CONV(_addr);
}
socklen_t netaddrlen()
{
return sizeof(_addr);
}
bool operator==(const inetaddr& addr)
{
return _ip==addr._ip && _port==addr._port;
}
string StringAddr()
{
return _ip+":"+to_string(_port);
}
~inetaddr()
{}
private:
struct sockaddr_in _addr;
uint16_t _port;
string _ip;
};
#endifinetaddr模块与前面实现的select/poll服务器一致，对网络地址进行了封装，提供了从IP和端口构造、从已有地址结构构造以及仅从端口构造三种构造函数，还提供了获取端口号、IP字符串、原生地址结构及其指针的方法，重载operator==用于地址比较。
3、mutex
mutex.hpp
#pragma once
#include<iostream>
#include<pthread.h>
using namespace std;
namespace mutexmodule
{
class mutex
{
public:
mutex()
{
pthread_mutex_init(&_mutex,nullptr);
}
void lock()
{
int n=pthread_mutex_lock(&_mutex);
(void)n;
}
void unlock()
{
int n=pthread_mutex_unlock(&_mutex);
(void)n;
}
~mutex()
{
pthread_mutex_destroy(&_mutex);
}
private:
pthread_mutex_t _mutex;
};
class lockguard
{
public:
lockguard(mutex& mutex):_mutex(mutex)
{
_mutex.lock();
}
~lockguard()
{
_mutex.unlock();
}
private:
mutex& _mutex;
};
}mutex模块与前面实现的select/poll服务器一致，对互斥锁进行了封装，mutex用于初始化和销毁互斥锁，lockguard采用RAII机制自动管理锁的生命周期，在构造时加锁，析构时解锁。
4、log
log.hpp
#ifndef _LOG_HPP_
#define _LOG_HPP_
#include<iostream>
#include<string>
#include<cstdio>
#include<filesystem>
#include<fstream>
#include<sstream>
#include<memory>
#include<ctime>
#include<unistd.h>
#include"mutex.hpp"
using namespace std;
namespace logmodule
{
using namespace mutexmodule;
const string gsep ="\r\n";
class logstrategy
{
public:
virtual ~logstrategy()=default;
virtual void synclog(const string& message)=0;
};
class consolelogstrategy:public logstrategy
{
public:
consolelogstrategy(){}
void synclog(const string& message) override
{
lockguard guard(_mutex);
cout<<message<<gsep;
}
~consolelogstrategy(){}
private:
mutex _mutex;
};
const string defaultpath="/var/log/";
const string defaultfile="my.log";
class filelogstrategy:public logstrategy
{
public:
filelogstrategy(const string& path=defaultpath,const string& file=defaultfile)
:_path(path),
_file(file)
{
lockguard guard(_mutex);
if(filesystem::exists(_path))
{
return;
}
try
{
filesystem::create_directories(_path);
}
catch(const filesystem::filesystem_error& e)
{
cerr<<e.what()<<endl;
}
}
void synclog(const string& message) override
{
lockguard guard(_mutex);
string filename=_path+(_path.back()=='/'?"":"/")+_file;
ofstream out(filename,ios::app);
if(!out.is_open())
{
return;
}
out<<message<<gsep;
out.close();
}
~filelogstrategy(){}
private:
string _path;
string _file;
mutex _mutex;
};
enum class loglevel
{
DEBUG,
INFO,
WARNING,
ERROR,
FATAL
};
string levelstr(loglevel lev)
{
switch(lev)
{
case loglevel::DEBUG: return "DEBUG";
case loglevel::INFO: return "INFO";
case loglevel::WARNING: return "WARNING";
case loglevel::ERROR: return "ERROR";
case loglevel::FATAL: return "FATAL";
default: return "UNKNOWN";
}
}
string gettimestamp()
{
time_t t=time(nullptr);
struct tm curr_tm;
localtime_r(&t,&curr_tm);
char timebuffer[128];
snprintf(timebuffer, sizeof(timebuffer),"%4d-%02d-%02d %02d:%02d:%02d",
curr_tm.tm_year+1900,
curr_tm.tm_mon+1,
curr_tm.tm_mday,
curr_tm.tm_hour,
curr_tm.tm_min,
curr_tm.tm_sec
);
return timebuffer;
}
class logger
{
public:
logger():_ptr(nullptr)
{
enableconsolelogstrategy();
}
void enableconsolelogstrategy()
{
_ptr=make_unique<consolelogstrategy>();
}
void enablefilelogstrategy()
{
_ptr=make_unique<filelogstrategy>();
}
class logmessage
{
public:
logmessage(const string& src,loglevel level,int num,logger& logger)
:_pid(getpid())
,_src(src)
,_num(num)
,_curr_time(gettimestamp())
,_level(level)
,_logger(logger)
{
stringstream ss;
ss<<"["<<_curr_time<<"]"<<"["<<levelstr(_level)<<"]"<<"["<<_pid<<"]"<<"["<<_src<<"]"<<"["<<_num<<"]"<<"-";
_loginfo=ss.str();
}
template<class K>
logmessage& operator<<(const K& info)
{
stringstream ss;
ss<<info;
_loginfo+=ss.str();
return *this;
}
~logmessage()
{
if(_logger._ptr)
{
_logger._ptr->synclog(_loginfo);
}
}
private:
string _curr_time;
loglevel _level;
pid_t _pid;
string _src;
int _num;
string _loginfo;
logger& _logger;
};
logmessage operator()(loglevel lev,const string& name,int line)
{
return logmessage(name,lev,line,*this);
}
~logger()
{
}
private:
unique_ptr<logstrategy> _ptr;
};
static logger log;
#define LOG(level) logmodule::log(level,__FILE__,__LINE__)
#define Enable_Console_Log_Strategy() logmodule::log.enableconsolelogstrategy()
#define Enable_File_Log_Strategy() logmodule::log.enablefilelogstrategy()
}
#endiflog为日志模块，与select/poll服务器实现一致，采取策略模式，定义了日志策略的抽象基类，并派生出控制台输出和文件输出两种具体策略。logger为日志器，内部持有策略对象的智能指针，嵌套了logmessage类，构造时将生成带有时间戳、进程ID、文件名和行号的日志前缀，通过operator<<流式输出，析构时将日志内容输出。
5、socket
socket.hpp
#ifndef _SOCKET_HPP_
#define _SOCKET_HPP_
#include<iostream>
#include<string>
#include<unistd.h>
#include<sys/types.h>
#include<sys/socket.h>
#include<netinet/in.h>
#include<arpa/inet.h>
#include<cstdlib>
#include"common.hpp"
#include"log.hpp"
#include"inetaddr.hpp"
using namespace std;
namespace socketmodule
{
using namespace logmodule;
const static int gbacklog=16;
class socket
{
public:
virtual ~socket(){}
virtual void socketordie()=0;
virtual void bindordie(uint16_t port)=0;
virtual void listenordie(int backlog)=0;
virtual void Close()=0;
virtual int recv(string* out)=0;
virtual int send(const string& message)=0;
virtual int accept(inetaddr* addr)=0;
virtual int connect(const string& ip,uint16_t port)=0;
virtual int fd()=0;
public:
void buildtcpsocket(uint16_t port,int backlog=gbacklog)
{
socketordie();
bindordie(port);
listenordie(backlog);
}
void buildclient()
{
socketordie();
}
};
const static int defaultfd=-1;
class tcpsocket:public socket
{
public:
tcpsocket()
:_sockfd(defaultfd)
{}
tcpsocket(int fd)
:_sockfd(fd)
{}
tcpsocket(const string& ip,uint16_t port)
{}
~tcpsocket()
{}
void socketordie() override
{
_sockfd=::socket(AF_INET,SOCK_STREAM,0);
if(_sockfd<0)
{
LOG(loglevel::FATAL)<<"socket error";
exit(SOCKET_ERR);
}
LOG(loglevel::INFO)<<"sock success";
}
void bindordie(uint16_t port) override
{
inetaddr addr(port);
int n=::bind(_sockfd,addr.netaddrptr(),addr.netaddrlen());
if(n<0)
{
LOG(loglevel::FATAL)<<"bind error";
exit(BIND_ERR);
}
LOG(loglevel::INFO)<<"bind success";
}
void listenordie(int backlog) override
{
int n=listen(_sockfd,backlog);
if(n<0)
{
LOG(loglevel::FATAL)<<"listen error";
exit(LISTEN_ERR);
}
LOG(loglevel::INFO)<<"listen success";
}
#define ACCEPT_ERR -3
#define ACCEPT_DONE -1
#define ACCEPT_CONTINUE -2
int accept(inetaddr* addr) override
{
struct sockaddr_in peer;
socklen_t len=sizeof(peer);
int fd=::accept(_sockfd,CONV(peer),&len);
if(fd<0)
{
if(errno==EAGAIN || errno==EWOULDBLOCK)
{
return -1;
}
else if(errno==EINTR)
{
return -2;
}
else
{
LOG(loglevel::WARNING)<<"accept error";
return -3;
}
}
return fd;
}
int recv(string* out) override
{
char buffer[4096*2];
ssize_t n=::recv(_sockfd,buffer,sizeof(buffer)-1,0);
if(n>0)
{
buffer[n]=0;
*out+=buffer;
}
return n;
}
int send(const string& message) override
{
return ::send(_sockfd,message.c_str(),message.size(),0);
}
void Close() override
{
if(_sockfd>=0)
{
::close(_sockfd);
}
}
int connect(const string& ip,uint16_t port) override
{
inetaddr addr(ip,port);
return ::connect(_sockfd,addr.netaddrptr(),addr.netaddrlen());
}
int fd(){return _sockfd;}
private:
int _sockfd;
};
}
#endifsocket模块与select/poll服务器实现一致，对socket套接字进行了封装，定义了抽象基类socket，声明了创建、绑定、监听、关闭、收发、接受连接、连接服务器等纯虚函数。tcpsocket继承socket并实现了所有方法，提供了构建TCP服务器和客户端的辅助方法，是非阻塞模式的基础。
6、epoller
epoller模块是对epoll的完整封装，为上层Reactor实现提供了高效的事件管理接口。
epoller.hpp
（1）构造函数
#pragma once
#include<iostream>
#include<unistd.h>
#include<sys/epoll.h>
#include"common.hpp"
#include"log.hpp"
using namespace std;
using namespace logmodule;
class epoller
{
public:
epoller()
:_epfd(-1)
{
_epfd=epoll_create(128);
if(_epfd<0)
{
LOG(loglevel::FATAL)<<"epoll error";
exit(EPOLL_CREATE_ERR);
}
LOG(loglevel::INFO)<<"epoll success"<<_epfd;
}
private:
int _epfd;
};_epfd=epoll_create(128)，构造函数通过调用epoll_create创建一个epoll实例，返回值由_epfd接收。
（2）modeventhelper
void modeventhelper(int sockfd,uint32_t event,int oper)
{
struct epoll_event ev;
ev.events=event;
ev.data.fd=sockfd;
int n=epoll_ctl(_epfd,oper,sockfd,&ev);
if(n<0)
{
LOG(loglevel::ERROR)<<"epoll_ctl error";
return;
}
LOG(loglevel::INFO)<<"epoll_ctl success: "<<sockfd;
}modeventhelper为核心函数，接收套接字描述符、事件类型和操作类型三个参数，struct epoll_event ev，ev.events=event，ev.data.fd=sockfd，声明一个epoll_event结构体，将events字段设置为传入的事件标志，data.fd设置为套接字文件描述符。int n=epoll_ctl(_epfd,oper,sockfd,&ev)，随后调用epoll_ctl执行操作。
（3）addevent
void addevent(int sockfd,uint32_t events)
{
modeventhelper(sockfd,events,EPOLL_CTL_ADD);
}addevent用于将事件添加到epoll中，modeventhelper(sockfd，events，EPOLL_CTL_ADD)，通过调用modeventhelper并传入EPOLL_CTL_ADD操作码，用于首次将套接字加入epoll事件监听。
（4）delevent
void delevent(int sockfd)
{
int n=epoll_ctl(_epfd,EPOLL_CTL_DEL,sockfd,nullptr);
(void)n;
}delevent用于从epoll中删除事件，int n=epoll_ctl(_epfd,EPOLL_CTL_DEL,sockfd,nullptr)，通过调用epoll_ctl传入EPOLL_CTL_DEL，并将最后一个参数置为nullptr，表示删除该文件描述符的所有事件，这个操作用于关闭连接。
（5）modevent
void modevent(int sockfd,uint32_t event)
{
modeventhelper(sockfd,event,EPOLL_CTL_MOD);
}modevent用于修改已注册的事件，modeventhelper(sockfd,event,EPOLL_CTL_MOD)，通过调用modeventhelper并传入EPOLL_CTL_MOD操作码，用于根据缓冲区状态动态调整监听写事件。
（6）waitevent
int waitevent(struct epoll_event revs[],int maxnum,int timeout)
{
int n=epoll_wait(_epfd,revs,maxnum,timeout);
if(n<0)
{
LOG(loglevel::WARNING)<<"epoll_wait error";
}
else if(n==0)
{
LOG(loglevel::WARNING)<<"epoll wait timeout";
}
else
{
}
return n;
}waitevent对epoll_wait进行了封装，接收事件数组、数组大小和超时时间三个参数，int n=epoll_wait(_epfd,revs,maxnum,timeout)，调用epoll_wait阻塞等待事件发生，n用于接收就绪事件的数量，return n，返回就绪文件描述符的数量。
（7）析构函数
~epoller()
{
if(_epfd>=0)
{
close(_epfd);
}
}析构函数负责清理资源，if(_epfd>=0)，close(_epfd)，如果_epfd有效则调用close关闭epoll实例，释放内核资源。
7、protocol
protocol模块负责协议解码和业务计算。
protocol.hpp
（1）decode
#ifndef _PROTOCOL_HPP_
#define _PROTOCOL_HPP_
#include<iostream>
#include<string>
#include"log.hpp"
using namespace std;
using namespace logmodule;
const string sep="\r\n";
class protocol
{
public:
protocol()
{}
bool decode(string& buffer,string* package)
{
ssize_t pos=buffer.find(sep);
if(pos==string::npos) return false;
*package=buffer.substr(0,pos);
buffer.erase(0,pos+sep.size());
return true;
}
};
#endifdecode负责从缓冲区中提取一个完整的包，ssize_t pos=buffer.find(sep)，首先在缓冲区中查找换行符的位置，if(pos==string::npos) 如果找不到说明缓冲区中没有完整的行，return false，返回false。*package=buffer.substr(0,pos)，如果找到了，就从缓冲区开头截取到换行符之前的部分作为完整的包存入package，buffer.erase(0,pos+sep.size())，然后从原缓冲区中删除这一行包括换行符。
（2） excute
string excute(string& package)
{
cout << "excute: " << package << endl;
size_t pos = string::npos;
char op = 0;
for(char c : {'+','-','*','/','%'}) {
pos = package.find(c);
if(pos != string::npos) {
op = c;
break;
}
}
if(pos == string::npos) return "ERROR\r\n";
int x = stoi(package.substr(0, pos));
int y = stoi(package.substr(pos + 1));
int result = 0;
switch(op)
{
case '+': result = x + y; break;
case '-': result = x - y; break;
case '*': result = x * y; break;
case '/':
if(y == 0) return "ERROR: division by zero\r\n";
result = x / y;
break;
case '%':
if(y == 0) return "ERROR: modulo by zero\r\n";
result = x % y;
break;
}
return to_string(result) + "\r\n";
}excute是业务处理的核心，通过for循环查找表达式中的操作符，pos=package.find(c)，遍历加减乘除取模五个字符，找到第一个出现的位置。找到操作符后，将表达式分割成左右两部分，int x=stoi(package.substr(0，pos))，int y=stoi(package.substr(pos+1))，使用stoi函数转换为整数，通过switch语句根据操作符进行计算。to_string(result)+"\r\n"，计算完成后，将结果转换为字符串并加上换行符返回。
8、connection
connection模块是所有连接类的抽象基类，定义了网络连接的统一接口和公共功能，为多态处理不同类型的连接提供了基础。
connection.hpp
#ifndef _CONNECTION_HPP_
#define _CONNECTION_HPP_
#include<iostream>
#include<string>
#include<functional>
#include"inetaddr.hpp"
using namespace std;
class Reactor;
using handler_t=function<string(string&)>;
class connection
{
public:
connection()
:_events(0)
,_owner(nullptr)
,_sockfd(-1)
{
}
virtual void recver()=0;
virtual void sender()=0;
virtual void excepter()=0;
virtual int getsockfd()=0;
void setevent(const uint32_t& events)
{
_events=events;
}
uint32_t getevent()
{
return _events;
}
void setfd(int sockfd)
{
_sockfd=sockfd;
}
int getfd()
{
return _sockfd;
}
void setowner(Reactor* owner)
{
_owner=owner;
}
Reactor* getowner()
{
return _owner;
}
void appendoutbuffer(const string& out)
{
}
string& inbuffer()
{
return _inbuffer;
}
void registerhandler(handler_t handler)
{
_handler=handler;
}
~connection()
{}
private:
Reactor *_owner;
uint32_t _events;
int _sockfd;
string _inbuffer;
string _outbuffer;
public:
handler_t _handler;
};
#endifconnection为抽象基类，recver、sender、excepter分别对应接收数据、发送数据和异常处理三种核心操作，getsockfd用于获取文件描述符。setevent、getevent用于设置和获取事件标志。setfd、getfd用于设置和获取套接字文件描述符。setowner、getowner用于设置和获取所属的Reactor对象。inbuffer用于返回输入缓冲区的引用，允许外部直接操作缓冲区。registerhandler用于注册业务处理回调函数，这是连接层与业务层解耦的关键。
9、Reactor
Reactor模块是整个服务器的核心，实现了Reactor设计模式，负责事件循环、事件分发和连接管理。Reactor将epoll的事件通知机制与业务处理逻辑完全解耦，是事件的中枢调度器。
Reactor.hpp
（1）构造函数
#ifndef _REACTOR_HPP_
#define _REACTOR_HPP_
#include<iostream>
#include<memory>
#include<unordered_map>
#include"epoller.hpp"
#include"connection.hpp"
#include"log.hpp"
using namespace std;
using namespace logmodule;
class Reactor
{
static const int revsnum=128;
public:
Reactor()
:_up(make_unique<epoller>())
,_isrunning(false)
{}
~Reactor()
{}
private:
unique_ptr<epoller> _up;
unordered_map<int,shared_ptr<connection>> _connections;
bool _isrunning;
struct epoll_event _revs[revsnum];
};
#endifunique_ptr<epoller> _up，_up是一个指向epoller对象的智能指针，封装了epoll的系统调用，unordered_map<int,shared_ptr<connection>> _connections，_connections用于将文件描述符映射到对应的连接，_isrunning为运行状态标志，控制事件循环的启停。_revs为epoll_event数组，用于存放epoll_wait返回的就绪事件列表。_up(make_unique<epoller>())，_isrunning(false)，构造函数通过make_unique初始化epoller对象并将运行标志设为false。
（2） 查找连接
#ifndef _REACTOR_HPP_
#define _REACTOR_HPP_
#include<iostream>
#include<memory>
#include<unordered_map>
#include"epoller.hpp"
#include"connection.hpp"
#include"log.hpp"
using namespace std;
using namespace logmodule;
class Reactor
{
static const int revsnum=128;
private:
bool func(int sockfd)
{
auto iter=_connections.find(sockfd);
if(iter==_connections.end())
{
return false;
}
else
{
return true;
}
}
public:
Reactor()
:_up(make_unique<epoller>())
,_isrunning(false)
{}
~Reactor()
{}
private:
unique_ptr<epoller> _up;
unordered_map<int,shared_ptr<connection>> _connections;
bool _isrunning;
struct epoll_event _revs[revsnum];
};
#endifauto iter=_connections.find(sockfd)，func用于根据文件描述符在_connections中查找是否存在对应连接，if(iter==_connections.end())，表示查找失败，return false，else查找成功，return true。
（3）connectionexist
bool connectionexist(const shared_ptr<connection>& conn)
{
return func(conn->getsockfd());
}
bool connectionexist(int sockfd)
{
return func(sockfd);
}connectionexist有两个重载版本，func(conn->getsockfd())，func(sockfd)，分别接收共享指针和文件描述符，调用func完成存在性检查。
（4）connectionempty
bool connectionempty()
{
return _connections.empty();
}return _connections.empty()，connectionempty用于判断连接映射表_connections是否为空。
（5） looponce
int looponce(int timeout)
{
return _up->waitevent(_revs,revsnum,timeout);
}_up->waitevent(_revs,revsnum,timeout)，looponce通过调用epoller的waitevent，等待事件发生并返回就绪事件数量。
（6） dispatcher
void dispatcher(int n)
{
for(int i=0;i<n;i++)
{
int sockfd=_revs[i].data.fd;
uint32_t revents=_revs[i].events;
if(revents & EPOLLERR)
{
revents|=(EPOLLIN | EPOLLOUT);
}
if(revents & EPOLLHUP)
{
revents|=(EPOLLIN | EPOLLOUT);
}
if(revents & EPOLLIN)
{
if(connectionexist(sockfd))
_connections[sockfd]->recver();
}
if(revents & EPOLLOUT)
{
if(connectionexist(sockfd))
_connections[sockfd]->sender();
}
}
}dispatcher是核心的事件分发函数，int sockfd=_revs[i].data.fd，uint32_t revents=_revs[i].events，通过for循环遍历就绪事件数组，revents & EPOLLERR，revents & EPOLLHUP，对于错误和挂起事件，revents |=(EPOLLIN | EPOLLOUT)，会主动添加读写事件标记。revents & EPOLLIN，对于读事件，_connections[sockfd]->recver()，调用连接的recver方法；revents & EPOLLOUT，对于写事件，_connections[sockfd]->sender()，调用连接的sender方法。
（7） printconnection
void printconnection()
{
for(auto& conn:_connections)
{
cout<<conn.second->getsockfd()<<" ";
}
cout<<"\r\n";
}cout<< conn.second->getsockfd()<<" "，printconnection通过范围for输出当前所有连接的文件描述符列表。
（8） loop
void loop()
{
if(connectionempty())
return;
_isrunning=true;
int timeout=-1;
while(_isrunning)
{
printconnection();
int n=looponce(timeout);
dispatcher(n);
}
_isrunning=false;
}loop是Reactor的主事件循环，int timeout=-1，timeout为超时时间，设为-1表示无限等待，通过while循环，printconnection()，int n=looponce(timeout)，dispatcher(n)，每次循环调用printconnection输出当前所有连接的文件描述符，然后调用looponce等待事件，最后调用dispatcher处理就绪事件。
（9） addconnection
void addconnection(shared_ptr<connection>& conn)
{
if(connectionexist(conn))
{
LOG(loglevel::WARNING)<<"connection exist"<<conn->getsockfd();
return;
}
uint32_t events=conn->getevent();
int sockfd=conn->getsockfd();
_up->addevent(sockfd,events);
conn->setowner(this);
_connections[sockfd]=conn;
}addconnection用于添加新连接，uint32_t events=conn->getevent()，int sockfd=conn->getsockfd()，获取连接关心的事件类型和文件描述符，_up->addevent(sockfd,events)，调用epoller的addevent将事件加入epoll监听，conn->setowner(this)，_connections[sockfd]=conn，接着设置连接的所有者为当前Reactor对象，最后将连接存入_connections中。
（10） enablereadwrite
void enablereadwrite(int sockfd,bool read,bool write)
{
if(!connectionexist(sockfd))
{
LOG(loglevel::WARNING)<<"sockfd exist"<<sockfd;
return;
}
uint32_t newevent=(EPOLLET | (read?EPOLLIN:0) | (write?EPOLLOUT:0));
_connections[sockfd]->setevent(newevent);
_up->modevent(sockfd,newevent);
}enablereadwrite用于动态修改某个连接的事件监听状态，uint32_t newevent=(EPOLLET | (read?EPOLLIN:0) | (write?EPOLLOUT:0))，根据read和write参数组合新的事件标志，_connections[sockfd]->setevent(newevent)，_up->modevent(sockfd,newevent)，最后更新连接的事件标志并调用epoller的modevent修改监听事件。
（11） delconnection
void delconnection(int sockfd)
{
_up->delevent(sockfd);
_connections.erase(sockfd);
close(sockfd);
}delconnection用于删除连接，_up->delevent(sockfd)，_connections.erase(sockfd)，close(sockfd)，首先从epoll删除该文件描述符的事件，然后从_connections中移除，最后关闭套接字文件描述符。
（12） stop
void stop()
{
_isrunning=false;
}_isrunning=false，stop通过将_isrunning运行标志设为false，从而终止事件循环。
10、channel
channel模块继承自connection抽象基类，用于封装一个非阻塞TCP连接的具体行为。
channel.hpp
（1）构造函数
#ifndef _CHANNEL_HPP_
#define _CHANNEL_HPP_
#include"Reactor.hpp"
#include<iostream>
#include<string>
#include<sys/types.h>
#include<sys/socket.h>
#include<memory>
#include<functional>
#include"common.hpp"
#include"connection.hpp"
#include"log.hpp"
#include"inetaddr.hpp"
using namespace std;
using namespace logmodule;
#define SIZE 1024
class channel:public connection
{
public:
channel(int sockfd,const inetaddr& addr)
:_sockfd(sockfd)
,_addr(addr)
{
setnonblock(_sockfd);
}
private:
int _sockfd;
inetaddr _addr;
string _inbuffer;
string _outbuffer;
};
#include"Reactor.hpp"
#endif构造函数通过接收套接字文件描述符sockfd和地址对象addr，setnonblock(_sockfd)，调用setnonblock将套接字设置为非阻塞模式。
（2）recver
void recver() override
{
char buffer[SIZE];
while(true)
{
buffer[0]=0;
ssize_t n=recv(_sockfd,buffer,sizeof(buffer)-1,0);
if(n>0)
{
buffer[n]=0;
_inbuffer+=buffer;
}
else if(n==0)
{
excepter();
return;
}
else
{
if(errno==EAGAIN || errno==EWOULDBLOCK)
{
break;
}
else if(errno==EINTR)
{
continue;
}
else
{
excepter();
return;
}
}
}
if(!_inbuffer.empty() && _handler)
{
_outbuffer+=_handler(_inbuffer);
_inbuffer.clear();
}
if(!_outbuffer.empty())
{
sender();
if(!_outbuffer.empty())
{
getowner()->enablereadwrite(_sockfd,true,true);
}
}
}ssize_t n=recv(_sockfd,buffer,sizeof(buffer)-1,0)，recver通过while循环调用recv读取数据直到缓冲区无新数据或遇到EAGAIN错误，buffer[n]=0，_inbuffer+=buffer，将读取到的内容追加到_inbuffer中，if(!_inbuffer.empty() && _handler)，如果输入缓冲区非空且存在回调处理函数_handler，_outbuffer+=_handler(_inbuffer)，_inbuffer.clear()，则调用_handler处理输入数据，将生成的响应内容存入_outbuffer并清空输入缓冲区，sender()，随后调用sender发送响应数据，if(!_outbuffer.empty())，如果输出缓冲区不为空，getowner()->enablereadwrite(_sockfd,true,true)，则调整Reactor监听读写事件。
（3）sender
void sender() override
{
while(true)
{
ssize_t n=send(_sockfd,_outbuffer.c_str(),_outbuffer.size(),0);
if(n>0)
{
_outbuffer.erase(0,n);
if(_outbuffer.empty())
break;
}
else if(n==0)
{
break;
}
else
{
if(errno==EAGAIN || errno==EWOULDBLOCK)
{
break;
}
if(errno==EINTR)
{
continue;
}
else
{
excepter();
return;
}
}
}
if(!_outbuffer.empty())
{
getowner()->enablereadwrite(_sockfd,true,true);
}
else
{
getowner()->enablereadwrite(_sockfd,true,false);
}
}
}ssize_t n=send(_sockfd,_outbuffer.c_str(),_outbuffer.size(),0)，sender通过while循环调用send发送_outbuffer中的数据，直到全部发完或发送缓冲区满，if(!_outbuffer.empty())，发送完毕后根据是否还有待发数据，通过getowner()->enablereadwrite动态调整Reactor监听的事件类型，getowner()->enablereadwrite(_sockfd,true,true)，如果还有数据未发完就同时监听读写事件，
getowner()->enablereadwrite(_sockfd,true,false)，else只监听读事件。
（4）excepter
void excepter() override
{
getowner()->delconnection(_sockfd);
}excepter()用于关闭连接，当连接异常、对端关闭或发生错误时，getowner()->delconnection(_sockfd)，excepter会调用Reactor的delconnection移除并关闭连接。
（5）getsockfd
int getsockfd() override
{
return _sockfd;
}return _sockfd，getsockfd用于获取_sockfd套接字文件描述符。
（6）inbuffer
string& inbuffer()
{
return _inbuffer;
}return _inbuffer，inbuffer用于返回输入缓冲区的内容。
（7）appendoutbuffer
void appendoutbuffer(const string& out)
{
_outbuffer+=out;
}_outbuffer+=out，appendoutbuffer用于实现将数据追加到输出缓冲区_outbuffer中。
11、listener
listener模块继承自connection基类，用于监听端口并接受新的客户端连接。
listener.hpp
（1）构造函数
#ifndef _LISTENER_HPP_
#define _LISTENER_HPP_
#include<iostream>
#include<memory>
#include"epoller.hpp"
#include"socket.hpp"
#include"connection.hpp"
#include"common.hpp"
#include"channel.hpp"
using namespace std;
using namespace socketmodule;
class listener:public connection
{
public:
listener(int port=defaultport)
:_port(port)
,_listensock(make_unique<tcpsocket>())
{
_listensock->buildtcpsocket(_port);
setevent(EPOLLIN | EPOLLET);
setnonblock(_listensock->fd());
setfd(_listensock->fd());
}
private:
int _port;
unique_ptr<socketmodule::socket> _listensock;
};
#include"Reactor.hpp"
#endif_listensock(make_unique<tcpsocket>())，通过make_unique创建tcpsocket对象，_listensock->buildtcpsocket(_port)，调用buildtcpsocket绑定_port端口开始监听，setevent(EPOLLIN | EPOLLET)，调用setevent将套接字设置为非阻塞边缘触发，setfd(_listensock->fd())，通过setfd将文件描述符记录到基类中。
（2）recver
void recver() override
{
inetaddr addr;
while(true)
{
int sockfd=_listensock->accept(&addr);
if(sockfd==ACCEPT_ERR)
break;
else if(sockfd==ACCEPT_CONTINUE)
continue;
else if(sockfd==ACCEPT_DONE)
break;
else
{
shared_ptr<connection> sp=make_shared<channel>(sockfd,addr);
sp->setevent(EPOLLIN | EPOLLET);
if(_handler!=nullptr)
sp->registerhandler(_handler);
getowner()->addconnection(sp);
}
}
}int sockfd=_listensock->accept(&addr)，recver通过while循环调用accept尽可能多地接收连接，shared_ptr<connection> sp=make_shared<channel>(sockfd,addr)，每成功接收一个新的客户端套接字，就创建一个channel对象来封装这个连接，sp->setevent(EPOLLIN | EPOLLET)，设置其事件为边缘可读，if(_handler!=nullptr)，sp->registerhandler(_handler)，如果外部通过registerhandler设置了回调函数则一并传递给新连接，getowner()->addconnection(sp)，最后通过getowner将新连接注册到Reactor中统一管理。
（3）getsockfd
int getsockfd() override
{
return _listensock->fd();
}return _listensock->fd()，getsockfd用于获取_listensock对应的文件描述符fd。
12、main
main模块实现了将该服务器的所有组件串联起来，启动了一个基于Reactor模式的事件驱动服务器。
main.cc
#include<iostream>
#include<string>
#include"connection.hpp"
#include"listener.hpp"
#include"channel.hpp"
#include"log.hpp"
#include"common.hpp"
#include"protocol.hpp"
using namespace std;
static void usage(string proc)
{
cerr<<"usage:"<<proc<<"port"<<endl;
}
int main(int argc,char*argv[])
{
if(argc!=2)
{
usage(argv[0]);
exit(USAGE_ERR);
}
Enable_Console_Log_Strategy();
uint16_t port=stoi(argv[1]);
shared_ptr<protocol> pp=make_shared<protocol>();
shared_ptr<connection> sp=make_shared<listener>(port);
sp->registerhandler([pp](string& inbuffer)->string{
cout<<"handler called"<<endl;
string repstr;
while(true)
{
string package;
if(!pp->decode(inbuffer,&package))
break;
repstr+=pp->excute(package);
}
return repstr;
});
unique_ptr<Reactor> rp=make_unique<Reactor>();
rp->addconnection(sp);
rp->loop();
return 0;
}Enable_Console_Log_Strategy()，启动控制台日志输出，shared_ptr<protocol> pp=make_shared<protocol>()，通过make_shared创建protocol对象用来处理协议的编解码和执行逻辑，uint16_t port=stoi(argv[1])，shared_ptr<connection> sp=make_shared<listener>(port)，通过make_shared创建listener对象并绑定到指定端口。通过调用sp->registerhandler为listener注册一个回调函数，当有数据到达时，这个回调函数将被调用，if(!pp->decode(inbuffer，&package))，通过while循环调用protocol的decode从输入缓冲区中提取完整的数据包，repstr+=pp->excute(package)，每提取出一个包就调用excute执行相应的业务逻辑并将结果拼接起来，return repstr，最后返回响应字符串。unique_ptr<Reactor> rp=make_unique<Reactor>()，rp->addconnection(sp)，rp->loop()，通过make_unique创建一个Reactor对象，将listener作为第一个连接添加进去，最后调用loop启动主循环。由于listener本身也是connection的子类，Reactor会监听它的读事件，每当有新客户端连接进来时，listener的recver会被触发，它会接受新连接并自动创建对应的channel对象添加到Reactor中，同时把listener上的回调函数也注册到这些新channel上，从而实现对所有客户端连接的统一处理。通过一个监听器加上一个Reactor事件循环，调用回调函数将协议处理逻辑注入到连接中。
Makefile：
Reactorserver:main.cc
g++ -o $@ $^ -std=c++17
.PHONY:clean
clean:
rm -rf Reactorserver通过Makefile即可实现一键化编译，g++ -o $@ $^ -std=c++17，采用c++17标准，服务器运行如下所示：
Makefile编译通过后，./Reactorserver 8925，绑定端口号8925，服务器开始运行，打开另一个终端，telnet 127.0.0.1 8925，进行连接，客户端向服务器发送消息，服务器可执行相应的计算功能，并将结果发送给客户端，结果如下所示：
客户端：
服务器：
结语
I/O多路复用的核心思想就是让一个线程同时监控多个文件描述符，只有当文件描述符就绪时才去处理，而不是每个文件描述符都阻塞等待。select/poll/epoll是Linux下经典的I/O多路复用方式，select有文件描述符数量上限，且每次调用都需要重新传递整个文件描述符集合，效率随文件描述符数量增多而线性下降。poll突破了数量限制，但仍需线性扫描所有文件描述符找出就绪者。epoll是高效的I/O复用方式，采用事件驱动机制，通过内核维护一个就绪列表，用户只需要索取就绪事件而不需要每次都重新传递全部文件描述符，同时也支持边缘触发和水平触发两种模式，非常适合大规模高并发场景。Reactor是一种事件驱动的编程模式，通过一个或多个事件循环来监听和处理各种事件，从而实现高效的并发处理。在基于epoll的Reactor服务器实现过程中，从最底层的epoll事件驱动机制出发，构建了一个非阻塞的事件处理框架，多路复用使得用少量的线程就能管理大量的连接，而Reactor模式则将连接的建立、读写、异常处理等操作抽象为对事件的响应，这种设计模式改变了程序的执行流程，不再是主动调用函数去读写，而是等待事件到来触发处理逻辑，这种反转正是高并发服务器实现的核心所在。
更多推荐
10
0- 0
已为社区贡献6条内容
所有评论(0)