高性能服务器模型之Reactor（单线程版本）

🕗 发布于 2024-11-23 05:32 服务器 c++ linux 网络

一、概念

Reactor模型是一种事件驱动的设计模式，用于处理并发I/O操作。其核心思想是将I/O事件的监听和实际的I/O操作分离开来，由事件循环（Event Loop）负责监听I/O事件，当事件发生时，将事件分发给相应的事件处理器（Event Handler）进行处理。

单线程版本如下图：

其中各个组件的任务如下：

Reactor：这是事件循环的核心，负责监听和分发事件。它等待I/O事件（如连接请求、数据到达等）的发生，并将这些事件分发给相应的处理器。
Acceptor：Acceptor是Reactor模式中的一个特殊处理器，专门用于处理新的连接请求。当Reactor监听到新的连接请求时，它会将这个事件交给Acceptor处理。Acceptor负责接受新的连接，并将新的连接分配给适当的资源或处理器。
Clients：在图中，客户端（client）代表与服务器建立连接的实体。在Reactor模式中，客户端可以是任何发起连接请求的实体。
Dispatch：Dispatch是Reactor中的一个组件，它负责将事件分发给相应的处理器。在图中，Dispatch连接了Acceptor和一系列的处理步骤。
Read：读取数据的步骤。一旦连接建立，服务器需要从客户端读取数据。
Decode：解码数据的步骤。读取的数据可能需要解码，以转换为服务器可以理解的格式。
Compute：计算或处理数据的步骤。服务器对解码后的数据进行业务逻辑处理。
Encode：编码数据的步骤。处理完的数据可能需要编码，以转换为客户端可以理解的格式。
Send：发送数据的步骤。服务器将编码后的数据发送回客户端。

二、代码实现

代码模块：

Socket：负责管理sockfd的创建和传参。

InetAddress：负责处理ip地址，端口号等的大小端转换处理等。

Acceptor：由Socket和InetAddress组合而成，ready函数负责完成服务器前期的基本操作，达到监听状态，accept负责获取连接的netfd。

TcpConnect：负责和客户端和读写操作，使用Acceptor的accept获得的文件描述符初始化，进行信息传递。

EevenLoop：负责事件循环中的操作，封装了epoll的创建，增加，监听和删除操作，TCP连接三个事件的处理，接受用户输入的回调函数，并处理但由于EvenLoop本身没有发送消息的能力，所以EvenLoop将回调函数传入TcpConnect中注册，再使用TcpConnect的回调函数。

Server：封装Acceptor和EvenLoop的一些操作，简化使用。

1、Socket

Socket.h

#pragma once

class Socket
{
public:
    Socket();
    ~Socket();
    int getSockfd();
private:
    int _fd;
};

Socket.cc

#include"Socket.h"
#include<kk.h>
Socket::Socket(){
    _fd=socket(AF_INET,SOCK_STREAM,0);
}

Socket::~Socket(){
    close(_fd);
}

int Socket::getSockfd(){
    return _fd;
}

2、InetAddress

InetAddress.h

#pragma once
#include<kk.h>
#include<iostream>
using std::string;

class InetAddress
{
public:
    InetAddress(const string &ip,const string &port);
    InetAddress(struct sockaddr_in clientAddr);
    ~InetAddress();
    struct sockaddr_in *getInetAddr();
    string getIp();
    string getPort();
private:
    struct sockaddr_in _serverAddr;
};

InetAddress.cc

#include"InetAddress.h"
InetAddress::InetAddress(const string & ip,const string & port){
    _serverAddr.sin_family=AF_INET;
    _serverAddr.sin_port=htons(atoi(port.c_str()));
    _serverAddr.sin_addr.s_addr=inet_addr(ip.c_str());
}
InetAddress::InetAddress(struct sockaddr_in clientAddr){
    
}
InetAddress::~InetAddress(){
    
}
struct sockaddr_in *InetAddress::getInetAddr(){
    return &_serverAddr;
}
    string getIp();
    string getPort();

3、Acceptor

Acceptor.h

#pragma once
#include"InetAddress.h"
#include"Socket.h"

class Acceptor
{
public:
    Acceptor(const string &ip,const string &port);
    ~Acceptor();
    void bind();
    void listen();
    int accept();
    void setReuse();
    void ready();
    int getSockfd();
private:
    Socket _sock;
    InetAddress _addr;
};

Acceptor.cc

#include"Acceptor.h"
Acceptor::Acceptor(const string &ip,const string &port)
     :_sock()
     ,_addr(ip,port){}

Acceptor::~Acceptor(){

}

void Acceptor::bind(){
    int ret=::bind(_sock.getSockfd(),(struct sockaddr *)_addr.getInetAddr(),sizeof(struct sockaddr_in));
    if(ret==-1){
        perror("bind");
        return;
    }
}

void Acceptor::listen(){
    int ret=::listen(_sock.getSockfd(),50);
    if(ret==-1){
        perror("listen");
        return;
    }
}

int Acceptor::accept(){
    int netfd=::accept(_sock.getSockfd(),NULL,NULL);
    if(netfd==-1){
        perror("accept");
        return -1;
    }
    return netfd;
}

void Acceptor::setReuse(){
    int reuse=1;
    setsockopt(_sock.getSockfd(),SOL_SOCKET,SO_REUSEADDR,&reuse,sizeof(reuse));
    int reuse2=1;
    setsockopt(_sock.getSockfd(),SOL_SOCKET,SO_REUSEPORT,&reuse2,sizeof(reuse2));
}

void Acceptor::ready(){
    setReuse();
    bind();
    listen();
}
int Acceptor::getSockfd(){
    return _sock.getSockfd();
}

4、TcpConnect

TcpConnect.h

#pragma once
#include<iostream>
#include<functional>
#include<memory>
#include<kk.h>
using std::string;
class TcpConnect;
using ConnectPtr=std::shared_ptr<TcpConnect>;
using TcpConnectCallback=std::function<void(const ConnectPtr &)>;
class TcpConnect
:public std::enable_shared_from_this<TcpConnect>
{
public:
    TcpConnect(int netfd);
    ~TcpConnect();
    string recv();
    int send(const string &msg);

    void setNewConnectCb(const TcpConnectCallback &cb);
    void setCloseCb(const TcpConnectCallback &cb);
    void setMessageCb(const TcpConnectCallback &cb);

    void handleNewConnectCb();
    void handleMessageCb();
    void handCloseCb();

    bool isClose();
private:
    int readn(char *buf,int len);
    int writen(char *buf,int len);
    int readline(char *buf,int len);

    TcpConnectCallback _onNewConnectCb;
    TcpConnectCallback _onCloseCb;
    TcpConnectCallback _onMessageCb;

    int _netfd;
};

TcpConnect.cc

#include"TcpConnect.h"

TcpConnect::TcpConnect(int netfd)
    :_netfd(netfd)
{

}

TcpConnect::~TcpConnect(){
    close(_netfd);
}

string TcpConnect::recv(){
     char buf[65535]={0}; 
     readline(buf,sizeof(buf)); 
     return string(buf); 
}

int TcpConnect::send(const string &msg){
    ::send(_netfd,msg.c_str(),msg.length(),MSG_NOSIGNAL);
}
    
int TcpConnect::readn(char *buf,int len){
    int left=len;
    char *pstr=buf;
    int ret=0;
    while(left>0){
        ret=read(_netfd,pstr,left);
        if(ret==-1&&errno==EINTR){
            continue;
        }
        else if(ret==-1){
            perror("read error -1");
            return -1;
        }
        else if(ret==0){
            break;
        }
        else{
            pstr+=ret;
            left-=ret;
        }
    }
    return len-left;
}
    
int TcpConnect::readline(char *buf,int len){
    int left=len-1;
    char *pstr=buf;
    int ret=0,total=0;
    while(left>0){
        ret=::recv(_netfd,pstr,left,MSG_PEEK);
        if(-1==ret&&errno==EINTR){
            continue;
        }
        else if(-1==ret){
            perror("readLine error -1");
            return -1;
        }
        else if(0==ret){
            break;
        }
        else{
            for(int i=0;i<ret;i++){
                if(pstr[i]=='\n'){
                    int sz=i+1;
                    readn(pstr,sz);
                    pstr+=sz;
                    *pstr='\0';
                    return total+sz;
                }
            }
            readn(pstr,ret);
            total+=ret;
            pstr+=ret;
            left-=ret;
        }
    }
    return total;
}
void TcpConnect::setNewConnectCb(const TcpConnectCallback &cb){
    _onNewConnectCb=std::move(cb);
}
void TcpConnect::setCloseCb(const TcpConnectCallback &cb){
    _onCloseCb=std::move(cb);
}
void TcpConnect::setMessageCb(const TcpConnectCallback &cb){
    _onMessageCb=std::move(cb);
}
void TcpConnect::handleNewConnectCb(){
    if(_onNewConnectCb){
        _onNewConnectCb(shared_from_this());
    }
    else{
        std::cout<<"_onNewConnectCb==nullptr"<<std::endl;
    }
}
void TcpConnect::handleMessageCb(){
    if(_onMessageCb){
        _onMessageCb(shared_from_this());
    }
    else{
        std::cout<<"_onMessageCb==nullptr"<<std::endl;
    }
}
void TcpConnect::handCloseCb(){
    if(_onMessageCb){
        _onCloseCb(shared_from_this());
    }
    else{
        std::cout<<"_onCloseCb"<<std::endl;
    }
}


bool TcpConnect::isClose(){
    char buf[10]={0};
    int ret=::recv(_netfd,buf,sizeof(buf),MSG_PEEK);
    return (0==ret);
}

5、EvenLoop

EvenLoop.h

#pragma once
#include<vector>
#include<memory>
#include<map>
#include<utility>
#include<kk.h>
#include<functional>
#include"Acceptor.h"
#include"TcpConnect.h"
using std::vector;
using std::shared_ptr;
using std::map;

using ConnectPtr=shared_ptr<TcpConnect>;
using TcpConnectCallback=std::function<void(const ConnectPtr &)>;

class EvenLoop
{
public:
    EvenLoop(Acceptor &acceptor);
    ~EvenLoop();
    int createEpoll();
    void addEpoll(int fd);
    void delEpoll(int fd);
    void loop();
    void unloop();
    void waitEpoll();
    void handleNewConnect();
    void handleMessage(int netfd);
    

    void setNewConnectCb(const TcpConnectCallback &cb);
    void setCloseCb(const TcpConnectCallback &cb);
    void setMessageCb(const TcpConnectCallback &cb);

private:
    int _epfd;
    vector<struct epoll_event> _readyArr;
    bool _isLooping;
    Acceptor &_acceptor;
    map<int,ConnectPtr> _conns;
    
    TcpConnectCallback _onNewConnectCb;
    TcpConnectCallback _onCloseCb;
    TcpConnectCallback _onMessageCb;
};

EvenLoop.cc

#include"EvenLoop.h"
#include<iostream>
using std::cerr;
using std::endl;
using std::cout;
EvenLoop::EvenLoop(Acceptor &acceptor)
    :_epfd(createEpoll())
    ,_readyArr(1024)
    ,_isLooping(false)
    ,_acceptor(acceptor)
    ,_conns()
{
    addEpoll(_acceptor.getSockfd());
}

EvenLoop::~EvenLoop(){
    close(_epfd);
}

int EvenLoop::createEpoll(){
    int ret=epoll_create(1);
    if(ret<0){
        perror("epoll_create");
        return ret;
    }
    return ret;
}

void EvenLoop::addEpoll(int fd){
    struct epoll_event event;
    event.events=EPOLLIN;
    event.data.fd=fd;
    int ret=epoll_ctl(_epfd,EPOLL_CTL_ADD,fd,&event);
    if(ret<0){
        perror("epoll_ctr_add");
        return;
    }
}

void EvenLoop::delEpoll(int fd){
    epoll_ctl(_epfd,EPOLL_CTL_DEL,fd,nullptr);
}

void EvenLoop::loop(){
    _isLooping=true;
    while(_isLooping==true){
        waitEpoll();
    }
}
    
void EvenLoop::unloop(){
    _isLooping=false;
}

void EvenLoop::waitEpoll(){
    int readyNum=0;
    do{
        readyNum=epoll_wait(_epfd,&*_readyArr.begin(),_readyArr.size(),3000);

    }while(readyNum==-1&&errno==EINTR);
    if(readyNum==-1){
        cerr<<"readyNum==-1"<<endl;
    }
    else if(readyNum==0){
        cout<<"----timeout!!!----"<<endl;
    }
    else{
        if(readyNum==(int)_readyArr.size()){
            _readyArr.reserve(2*readyNum);
        }
        for(int i=0;i<readyNum;i++){
            int fd=_readyArr[i].data.fd;
            if(fd==_acceptor.getSockfd()){
                handleNewConnect();
            }
            else{
                handleMessage(fd);
            }
        }
    }
}
void EvenLoop::handleNewConnect(){
    int netfd=_acceptor.accept();
    if(netfd<0){
        perror("handleNewConnect");
        return ;
    }
    addEpoll(netfd);
    ConnectPtr conn(new TcpConnect(netfd));
    conn->setNewConnectCb(_onNewConnectCb);
    conn->setMessageCb(_onMessageCb);
    conn->setCloseCb(_onCloseCb);
    _conns.insert(std::make_pair(netfd,conn));
    conn->handleNewConnectCb();
}

void EvenLoop::handleMessage(int netfd){
    auto it=_conns.find(netfd);
    //存在该此连接
    if(it!=_conns.end()){
       bool flag=it->second->isClose(); 
       if(flag==true){//已断开为true
            it->second->handCloseCb();
            _conns.erase(it);
            delEpoll(netfd);
            return ;
       }
       else{
           it->second->handleMessageCb();
       }
    }
    else{
        std::cout<<"连接不存在"<<std::endl;
        return;
    }
}

void EvenLoop::setNewConnectCb(const TcpConnectCallback &cb){
    _onNewConnectCb=cb;
}
void EvenLoop::setCloseCb(const TcpConnectCallback &cb){
    _onCloseCb=cb;
}
void EvenLoop::setMessageCb(const TcpConnectCallback &cb){
    _onMessageCb=cb;
}

6、TcpServer

TcpServer.h

#pragma once
#include"EvenLoop.h"
class TcpServer
{
public:
    TcpServer(const string &ip,const string &port);
    ~TcpServer();
    void start();
    void stop();
    void setAllFunction(const TcpConnectCallback &newCallBack,
                        const TcpConnectCallback &msgCallBack
                        ,const TcpConnectCallback &closeCallBack);

private:
    Acceptor _acceptor;
    EvenLoop _loop;
};

TcpServer.cc

#include"TcpServer.h"

TcpServer::TcpServer(const string &ip,const string &port)
    :_acceptor(ip,port)
     ,_loop(_acceptor){

}
TcpServer::~TcpServer(){

}

void TcpServer::start(){
    _acceptor.ready();
    _loop.loop();
}
void TcpServer::stop(){
    _loop.unloop();
}
void TcpServer::setAllFunction(const TcpConnectCallback &newCallBack,
                        const TcpConnectCallback &msgCallBack
                        ,const TcpConnectCallback &closeCallBack){
    _loop.setNewConnectCb(newCallBack);
    _loop.setMessageCb(msgCallBack);
    _loop.setCloseCb(closeCallBack);
}

三、测试

test.c

#include <iostream>
#include"Socket.h"
#include"Acceptor.h"
#include"InetAddress.h"
#include"TcpConnect.h"
#include"EvenLoop.h"
#include"TcpServer.h"
#include"ThreadPool.h"


using std::cout;
using std::endl;

void onNew(const ConnectPtr& con){
    cout<<"---a client has connected!!---"<<endl;
}
void onMes(const ConnectPtr& con){
    string ret=con->recv();
    cout<<"msg>>"<<ret<<endl;
    con->send(ret);
}
void onClose(const ConnectPtr&con){
    cout<<"---a client has close---"<<endl;

}

int main()
{

    TcpServer server("127.0.0.1","1234");   
    server.setAllFunction(onNew,onMes,onClose);
    server.start();
    return 0;
}

server

client

实现了一个执行回声业务的reactor单线程服务器模型。

原文地址：https://blog.csdn.net/weixin_73809064/article/details/143972768

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

Windows电脑本地部署llamafile并接入Qwen大语言模型远程AI对话实战
Windows电脑本地部署llamafile并接入Qwen大语言模型远程AI对话实战
阅读更多2024-11-23
解决vm虚拟机中ctrl键粘滞问题（一直处于按下状态）
虚拟机中点击鼠标左键没有问题，单独按下键盘也没有问题，但是如果按下键盘的同时在按下鼠标左键就会卡住。而且100%稳定复现。发现问题原因：vm版本的问题，我的版本是17.5.0，升级成17.5.2 bu
阅读更多2024-11-23
第十章 JavaScript的应用
10.1.1 JavaScript简介1 简单性JavaScript 设计之初就考虑到了非专业程序员的使用，因此它的语法相对简单，容易上手。2 动态性JavaScript 是一种解释型语言，代码可以在
阅读更多2024-11-23
非root用户安装CUDA
4.下载完成后执行.run文件，需要添加两个参数，一个是--override，用于跳过某些检查或限制；另一个是--toolkitpath，它表示安装路径，必须是自己目录下面的路径，如果是根目录就会安装
阅读更多2024-11-23
Python操作neo4j库py2neo使用之py2neo 删除及事务相关操作（三）
【代码】## Python操作neo4j库py2neo使用之py2neo 删除及事务相关操作（三）
阅读更多2024-11-23
【通俗理解】Jensen不等式与变分分布q(z)在积分计算中的应用
Jensen不等式 #变分分布 #积分计算 #期望 #凸函数 #优化问题 #下界估计 #机器学习。
阅读更多2024-11-23
「Mac玩转仓颉内测版27」基础篇7 - 字符串类型详解
本篇将介绍 Cangjie 中的字符串类型，包括字符串的定义、字面量形式、插值表达、常用操作及应用场景，帮助开发者熟练掌握字符串的使用。
阅读更多2024-11-23
数据科学与SQL：组距分组分析 | 区间分布问题
绝对值分布分析也可以理解为组距分组分析。对于某个指标而言，一个记录对应的指标值的绝对值，肯定落在所有指标值的绝对值的最小值和最大值构成的区间内，根据一定的算法，在把这个区间划分为等距离的几个小区间，，
阅读更多2024-11-23
c++编译报C1004错误的原因及解决办法
可能是你的代码文件最后没有正确的结束符，或者是缺少了一些必要的头文件或语句。默认磁盘驱动器没有足够的空间用于临时文件，需要大约两倍于源文件的空间。编译器到达了源文件尾但未解析构造。计算结果为假的 #i
阅读更多2024-11-23
mysql 去重补全取出重复变量函数和存储过程
mysql 去重补全取出重复变量函数和存储过程
阅读更多2024-11-23