今天来针对上一节课讲的多路转接知识再进一步进行设计，Reactor是基于epoll的ET模式设计的，在现在的高校和企业中是广泛应用的，今天我们来实现一个简洁版，完整版博主可没那个实力~
目录
基本原理
代码实现
epoll_server.cc
Accepter.hpp
Reactor.hpp
Service.hpp
Sock.hpp
Util.hpp
Makefile
测试结果
基本原理
Reactor负责接收事件并关心对应文件描述符读写事件，当epoll中就绪队列有数据时，把事件派发给连接管理器，Accepter调用对应的读写模块。
代码实现
epoll_server.cc
#include "Reactor.hpp"
#include"Sock.hpp"
#include<stdlib.h>
#include<unistd.h>
#include"Accepter.hpp"
#include"Util.hpp"
static void Usage(std::string proc)
{
std::cout << "Usage: " << "\n\t" << proc << " port" << std::endl;
}
int main(int argc, char* argv[])
{
if(argc != 2)
{
Usage(argv[0]);
exit(1);
}
//1、创建socket, 监听
int listen_sock = Sock::Socket();
SetNonBlock(listen_sock);//listen_sock设置为非阻塞
Sock::Bind(listen_sock, (uint16_t)atoi(argv[1]));
Sock::Listen(listen_sock);
//2、创建Reactor对象
//Reactor 反应堆模式: 通过多路转接方案, 被动的采用事件派发的方式, 去反向的调用对应的回调函数
//1. 检测到事件 -- epoll
//2. 派发事件 - Dispatcher (事件派发 + IO) + 业务处理: 半同步半异步的处理方式
//3. 链接 -- accepter
//4. IO -- recver, sender
Reactor* R = new Reactor();
R->InitReactor();
//3、 给Reactor反应堆中加柴火
//3.1 有柴火
Event* evp = new Event;
evp->sock = listen_sock;
evp->R = R; //指向柴火加的反应堆
//Accepter链接管理器
evp->RegisterCallBack(Accepter, nullptr, nullptr);
//3.2 将准备好的柴火放入反应堆Reactor中
R->InsertEvent(evp, EPOLLIN | EPOLLET); //设置ET模式
//4. 开始进行事件派发
for(; ;)
{
R->Dispatcher(1000); //事件派发
}
return 0;
}Accepter.hpp
#pragma once
#include<iostream>
#include"Reactor.hpp"
#include"Sock.hpp"
#include"Service.hpp"
#include"Util.hpp"
int Accepter(Event* evp)
{
std::cout << "有新的链接到来了, 就绪的sock是: " << evp->sock << std::endl;
//经过这样的死循环，会不断的把链接构建成event结构添加到反应堆中
while(true)
{
int sock = Sock::Accept(evp->sock);
if(sock < 0)
{
std::cout << "Accept Done!" << std::endl;
break;
}
std::cout << "Accept success: " << sock << std::endl;
SetNonBlock(sock);
//获取链接成功, IO socket
Event* other_ev = new Event();
other_ev->sock = sock;
other_ev->R = evp->R; //为什么要让所有的Event指向自己所属的Reactor??
//recver sender error 就是我们代码中的较顶层, 只负责读取!
other_ev->RegisterCallBack(Recver, Sender, Error);
//不能同时设置EPOLLOUT和EPOLLIN, 这样服务器会一直死循环，服务器性能大大降低
evp->R->InsertEvent(other_ev, EPOLLIN|EPOLLET);
}
return 0;
}Reactor.hpp
#pragma once
#include<iostream>
#include<string>
#include<sys/epoll.h>
#include<unordered_map>
#include<stdlib.h>
#include<unistd.h>
//一般处理IO的时候, 我们只有三种接口需要处理
//处理写入
//处理读取
//处理异常
class Event; //声明
class Reactor;
#define SIZE 128
#define NUM 64
typedef int (*callback_t)(Event* ev); //函数指针类型
//需要让epoll管理的基本结点
class Event
{
public:
int sock; //对应的文件描述符
std::string inbuffer;//对应的sock, 对应的输入缓冲区->粘包问题解决
std::string outbuffer; //对应的sock, 对应的输入缓冲区->epoll发送出去
//sock设置回调
callback_t recver;
callback_t sender;
callback_t errorer;
//设置Event回指Reactor的指针
Reactor* R;
public:
Event()
{
sock = -1;
recver = nullptr;
sender = nullptr;
errorer = nullptr;
R = nullptr;
}
~Event()
{}
//设置回调
void RegisterCallBack(callback_t _recver, callback_t _sender, callback_t _errorer)
{
recver = _recver;
sender = _sender;
errorer = _errorer;
}
};
//不需要关心任何sock的类型(listen, 读, 写)
//如何进行使用该类, 对Event进行管理
class Reactor
{
private:
int epfd;
std::unordered_map<int, Event*> events; //我的Epoll类管理的所有的Event的集合
public:
Reactor()
:epfd(-1)
{}
~Reactor()
{}
void InitReactor()
{
epfd = epoll_create(SIZE);
if(epfd < 0)
{
std::cerr << "epoll_create error" << std::endl;
exit(2);
}
std::cout << "InitReactor success" << std::endl;
}
//增加
bool InsertEvent(Event* evp, uint32_t evs)
{
//1、将sock中的sock插入到epoll中
struct epoll_event ev;
ev.events = evs;
ev.data.fd = evp->sock;
if(epoll_ctl(epfd, EPOLL_CTL_ADD, evp->sock, &ev) < 0) //-1失败, 0成功
{
std::cerr << "epoll_ctl add event failed" << std::endl;
return false;
}
//2、将ev本身插入到unordered_map中
events.insert({evp->sock, evp});//evp是维护的一个Event结点
return true;
}
void DeleteEvent(Event* evp)
{
int sock = evp->sock;
auto iter = events.find(sock);
if(iter != events.end())//找到了
{
//1、将sock中的sock从epoll中删除
epoll_ctl(epfd, EPOLL_CTL_DEL, sock, nullptr);//删除不关心events事件,所以设置成nullptr
//2. 把特定的ev 从unordered_map中移除
events.erase(iter);
//3、close
close(sock);
//4、删除event结点 -->它是new出来的, 要手动释放
delete evp;
}
}
//关于修改, 也是最后看, 使能读写
bool EnableRW(int sock, bool enbread, bool enbwrite)
{
struct epoll_event ev;
ev.events = EPOLLET | (enbread ? EPOLLIN : 0) | (enbwrite ? EPOLLOUT : 0);
ev.data.fd = sock;
if(epoll_ctl(epfd, EPOLL_CTL_MOD, sock, &ev) < 0)
{
std::cerr << "epoll_ctl mod event failed" << std::endl;
return false;
}
return true;
}
//判断sock是否是合法的
bool IsSockOK(int sock)
{
auto iter = events.find(sock);
return iter != events.end(); //找到就合法
}
//就绪事件的派发逻辑
void Dispatcher(int timeout) //将就绪事件派发给sock
{
struct epoll_event revs[NUM];
int n = epoll_wait(epfd, revs, NUM, timeout);
for(int i = 0; i < n; i++) //n 表示就绪的fd个数, 维护在了数组里面
{
int sock = revs[i].data.fd;
uint32_t revents = revs[i].events; //本次文件描述符就绪的事件
//代表差错处理, 将所有的错误问题全部转化成为让IO函数去解决
if(revents & EPOLLERR)
{
revents |= (EPOLLIN | EPOLLOUT); //设置读写
}
if(revents & EPOLLHUP) //对端链接关闭
{
revents |= (EPOLLIN | EPOLLOUT); //设置读写
}
//读数据就绪, 可能有bug,后面解决
if(revents & EPOLLIN)
{
//直接调用回调方法, 执行对应的读取
if(IsSockOK(sock) && events[sock]->recver) //recever是回调
{
events[sock]->recver(events[sock]);
}
}
if(revents & EPOLLOUT)
{
//直接调用回调方法, 执行对应的读取
if(IsSockOK(sock) && events[sock]->sender)
{
events[sock]->sender(events[sock]);
}
}
}
}
};Service.hpp
#pragma once
#include "Reactor.hpp"
#include <vector>
#include <errno.h>
#include"Util.hpp"
#define ONCE_SIZE 128
//1: 本轮读取全部完成
//-1: 读取出错
//0: 对端关闭链接
static int RecverCore(int sock, std::string &inbuffer)
{
while (true)
{
char buffer[ONCE_SIZE];
ssize_t s = recv(sock, buffer, ONCE_SIZE - 1, 0);
if (s > 0)
{
//读取成功
buffer[s] = 0;
inbuffer += buffer;
}
else if (s < 0)
{
if(errno == EINTR)
{
//IO被信号打断, 概率特别低
continue;
}
//errno会被自动设置
if (errno == EAGAIN || errno == EWOULDBLOCK)
{
// 1、读完, 底层没数据了
return 1; //success
}
// 2、真的出错了
return -1;
}
else // s == 0
{
return 0;
}
}
}
int Recver(Event *evp)
{
std::cout << "Recver been called" << std::endl;
// 1、真正的读取
int result = RecverCore(evp->sock, evp->inbuffer);
if(result <= 0)
{
//差错处理
if(evp->errorer)
{
evp->errorer(evp);
}
return -1;
}
//1+2X2+3X5+6X
// 2、分包 -- 一个或者多个报文 -- 解决粘包问题
std::vector<std::string> tokens;
std::string sep = "X";
SplitSegment(evp->inbuffer, &tokens, sep); //从缓冲区读取字节到tokens
// 3、反序列化 -- 针对一个报文 提取有效参与计算或者存储的信息
for (auto &seg : tokens)
{
std::string data1, data2;
if(Deserialize(seg, &data1, &data2)) //就是和业务强相关了
{
// 4、业务逻辑 -- 得到结果
int x = atoi(data1.c_str());
int y = atoi(data2.c_str());
int z = x + y;
// 5、构建响应 -- 添加到evp->outbuffer!!
// 1+2X ---> 1+2=3X
std::string res = data1;
res += "+";
res += data2;
res += "=";
res += std::to_string(z);
res += sep;
//send?? 不能直接send!!!
evp->outbuffer += res; //把数据写入自己维护的缓冲区
}
}
// 6、尝试直接/间接进行发送 -- 后续说明
//必须条件成熟了(写事件就绪), 你才能发送呢??
//一般只要将报文处理完毕, 才需要发送
//写事件一般都是就绪的, 但是用户不一定是就绪的!
//对于写事件, 我们通常是按需设置!!
if(!(evp->outbuffer).empty())
{
//写打开的时候, 默认就是就绪的, 即便是发送缓冲区已经满了
//epoll 只要用户重新设置了OUT事件, EPOLLOUT至少会触发一次
evp->R->EnableRW(evp->sock, true, true); //读写使能开启
}
return 0;
}
//1: 数据全部发完
//0: 数据没有发完, 但是不能再发了
//-1:发送失败
int SenderCore(int sock, std::string& outbuffer)
{
while(true)
{
int total = 0; //本轮累计发送的数据量
const char* start = outbuffer.c_str();
int size = outbuffer.size();
ssize_t curr = send(sock, start + total, size - total, 0);
if(curr > 0)
{
total += curr;
if(total == size)
{
//全部将数据发送完成
outbuffer.clear();
return 1;
}
}
else
{
//数据没有发送完成, 但是不能在发送了
if(errno == EINTR) //IO信号中断
{
continue;
}
if(errno == EAGAIN || errno == EWOULDBLOCK) //发送缓冲区满了
{
outbuffer.erase(0, total);
return 0;
}
return -1;
}
}
}
int Sender(Event *evp)
{
std::cout << "Sender been called" << std::endl;
//1: 数据全部发完
//0: 数据没有发完, 但是不能再发了
//-1:发送失败
int result = SenderCore(evp->sock, evp->outbuffer);
if(result == 1)
{
evp->R->EnableRW(evp->sock, true, false); //按需设置
}
else if(result == 0)
{
//可以什么也不做
evp->R->EnableRW(evp->sock, true, false);
}
else
{
if(evp->errorer)
{
evp->errorer(evp); //errorer统一处理差错
}
}
return 0;
}
int Error(Event *evp)
{
std::cout << "Error been called" << std::endl;
evp->R->DeleteEvent(evp);
return 0;
}Sock.hpp
#pragma once
#include <iostream>
#include <string>
#include <string.h>
#include <stdlib.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
using namespace std;
class Sock
{
public:
static int Socket()
{
int sock = socket(AF_INET, SOCK_STREAM, 0); // UDP
if (sock < 0)
{
cerr << "socket err" << endl;
exit(2);
}
return sock;
}
static void Bind(int sock, uint16_t port)
{
struct sockaddr_in local;
memset(&local, 0, sizeof(local));
local.sin_family = AF_INET;
local.sin_port = htons(port);
local.sin_addr.s_addr = INADDR_ANY; //服务端,ip地址
if (bind(sock, (struct sockaddr *)&local, sizeof(local)) < 0)
{
cerr << "bind error" << endl;
exit(3);
}
}
static void Listen(int sock)
{
if (listen(sock, 5) < 0)
{
cerr << "listen error" << endl;
exit(4);
}
}
static int Accept(int sock)
{
struct sockaddr_in peer;
socklen_t len = sizeof(peer);
int fd = accept(sock, (struct sockaddr *)&peer, &len);
if(fd >= 0)
{
return fd;
}
return -1;
}
static void Connect(int sock, string ip, uint16_t port)
{
struct sockaddr_in server;
memset(&server, 0, sizeof(server));
server.sin_family = AF_INET;
server.sin_port = htons(port);
server.sin_addr.s_addr = inet_addr(ip.c_str());//字符串->整型,大小端解决了
//inet_ntoa 整型->字符串
if(connect(sock, (struct sockaddr*)&server, sizeof(server)) == 0)
{
cout << "Connect Success" << endl;
}
else
{
cout << "Connect Failed" << endl;
exit(5);
}
}
};Util.hpp
#pragma once
#include<iostream>
#include<unistd.h>
#include<fcntl.h>
//工具类
//设置一个sock成为非阻塞
void SetNonBlock(int sock)
{
int f1 = fcntl(sock, F_GETFL);
if(f1 < 0)
{
std::cerr << "fcntl failed" << std::endl;
return;
}
fcntl(sock, F_SETFL, f1|O_NONBLOCK);//设置非阻塞
}
//1+2X2+3X5+6X
void SplitSegment(std::string& inbuffer, std::vector<std::string>* tokens, std::string sep)
{
while(true)
{
std::cout << "inbuffer: " << inbuffer << std::endl; //查看缓冲区里还有什么
auto pos = inbuffer.find(sep);
if(pos == std::string::npos) //没有找到
{
break;
}
std::string sub = inbuffer.substr(0, pos); //[ )
tokens->push_back(sub);
inbuffer.erase(0, pos + sep.size()); //从0开始, 移除pos+sep.size()个
}
}
bool Deserialize(const std::string& seg, std::string* out1, std::string* out2) //就是和业务强相关
{
//1+2
std::string op = "+";
auto pos = seg.find("+");
if(pos == std::string::npos) //没找到
{
return false;
}
*out1 = seg.substr(0, pos);
*out2 = seg.substr(pos+op.size());
return true;
}Makefile
epoll_server:epoll_server.cc
g++ -o $@ $^ -std=c++11
.PHONY:clean
clean:
rm -f epoll_server测试结果
看到这里, 给博主点个赞吧~
