|
Reactor事件驱动模式 一，什么是Reactor模式？ 1.它的中文名叫反应器，是事件驱动中的一种（还有另外一种叫Proactor：主动器） 2.在我们写一些比较小的[服务器-客户端]的时候，经常简单地使用BSD Socket(accept，connect，recv，send等等)， 但是这样的话，程序的[Socket处理逻辑]和[业务逻辑]就混合在一起，导致2者的耦合性非常高，Reactor模式正是为了 解决这种耦合性而诞生的 3.Reactor(反应器)是相对于调用者而言的，因为调用之后，当有事件，Reactor就会通知我们，我们就去响应事件。 二，Reactor模式的实现（C++）： (1).实现类简介： 1.多路复用组件接口：IReactor 2.具体多路复用组件：CSelectReactor（在Windows是用select实现，在Linux可以用epoll等等） 3.事件组件接口：IEvent（事件是和Socket关联在一起的） 4.具体事件组件：CAcceptEvent，CReadEvent (2).实现思路： 1.定义一个具体的多路复用组件，该组件负责： A.多路复用事件 B.注册一个事件到多路复用事件列表中 C.当有事件发生，判断事件的类型（分别有：Read，Write），然后调用分别调用事件的Read事件或者Write事件 2.定义事件，事件都是继承于IEvent接口类，并且提供2个接口： A.ReadEvent接口，当有Read事件的时候，该接口会被调用 B.WriteEvent接口，当有Write事件的时候，该接口被调用 (3).实现： A.多路复用组件接口(IReactor)： /////////////类声明/////////////// typedef map<SOCKET,IEvent*> SOCKET_MAP_DEF;//以Socket为映射的IEvent*列表 typedef map<SOCKET,IEvent*>::iterator SOCKET_MAP_ITER_DEF; class IReactor { public: virtual int Init() = 0;//初始化 virtual int Dispatch() = 0;//多路复用Dispatch int RegisterEvent(IEvent* pEvent);//注册一个Event IReactor(); protected: SOCKET_MAP_DEF cEventMap; }; /////////////类实现/////////////// IReactor::IReactor() { WORD wVersionRequested; WSADATA wsaData; int err; wVersionRequested = MAKEWORD( 2,2 ); err = WSAStartup( wVersionRequested,&wsaData ); if ( err != 0 ) { return; } if ( LOBYTE( wsaData.wVersion ) != 2 || HIBYTE( wsaData.wVersion ) != 2 ) { WSACleanup( ); return; } } int IReactor::RegisterEvent(IEvent* pEvent) { cEventMap.insert(make_pair(pEvent->GetSocket(),pEvent));//注册事件 return TRUE; } B.具体多路复用组件(CSelectReactor)： /////////////类声明/////////////// class CSelectReactor : public IReactor//继承于IReactor { public: int Init(); int Dispatch(); private: typedef struct { FD_SET stReadFDSet; FD_SET stWriteFDSet; FD_SET stExpFDSet; }SELECT_REACTOR_DATA;//select的3中FD，分别用于select函数 protected: SELECT_REACTOR_DATA stFDSet; }; /////////////类实现/////////////// int CSelectReactor::Init() { return 0; } int CSelectReactor::Dispatch() { while(1) { FD_ZERO(&stFDSet.stReadFDSet); FD_ZERO(&stFDSet.stWriteFDSet); FD_ZERO(&stFDSet.stExpFDSet); if (cEventMap.size() == 0) { break; } //1.根据Event_Type把事件增加到相应的FD中 for (SOCKET_MAP_ITER_DEF i = cEventMap.begin(); i != cEventMap.end(); i++) { switch (i->second->GetEventType()) { case enumEvent_Type_Read: { FD_SET(i->second->GetSocket(),&stFDSet.stReadFDSet); }break; case enumEvent_Type_Write: { FD_SET(i->second->GetSocket(),&stFDSet.stWriteFDSet); }break; } } //2.开始Select int nRet = select(0,&stFDSet.stReadFDSet,&stFDSet.stWriteFDSet,&stFDSet.stExpFDSet,NULL); if (nRet == SOCKET_ERROR)//出错 { break; } else if (nRet == 0)//超时 { } else { for (SOCKET_MAP_ITER_DEF i = cEventMap.begin(); i != cEventMap.end();) { BOOL bIsEvent = FALSE; if (FD_ISSET(i->second->GetSocket(),&stFDSet.stReadFDSet))//有Read事件 { i->second->ReadEvent(); bIsEvent = TRUE; } if (FD_ISSET(i->second->GetSocket(),&stFDSet.stWriteFDSet))//有Write事件 { i->second->WriteEvent(); bIsEvent = TRUE; } if(bIsEvent && i->second->IsEventPersist() == FALSE) { delete i->second;//!!!这里暂时这样做 i = cEventMap.erase(i); } else { i++; } } } } return 0; } C.事件组件接口(IEvent)： /////////////类声明/////////////// class IEvent { public: IEvent() { bIsPersist = FALSE; } VOID SetEvent(SOCKET hSocket,EVENT_TYPE enumEventType);//设置一个Event的事件 SOCKET GetSocket(); EVENT_TYPE GetEventType();
VOID SetEventPersist(BOOL bIsPersist) { this->bIsPersist = bIsPersist; }//设置Event是否[持久性] BOOL IsEventPersist() { return bIsPersist == TRUE; } virtual int ReadEvent() = 0;//Read事件接口 virtual int WriteEvent() = 0;//Write事件接口 protected: SOCKET hSocket; EVENT_TYPE enumEventType; BOOL bIsPersist; }; /////////////类实现/////////////// VOID IEvent::SetEvent(SOCKET hSocket,EVENT_TYPE enumEventType) { this->hSocket = hSocket; this->enumEventType = enumEventType; } SOCKET IEvent::GetSocket() { return hSocket; } EVENT_TYPE IEvent::GetEventType() { return enumEventType; } D.具体事件组件(CAcceptEvent，CReadEvent)： class CReadEvent : public IEvent { public: int ReadEvent()//Read事件 { char szBuf[1024 * 2] = {0}; int nRet = recv(GetSocket(),szBuf,1024 * 2,0);//从Socket读取数据 if (nRet == -1)//对方关闭Socket { SetEventPersist(FALSE); closesocket(GetSocket()); return 0; } cout<<"ReadBuffer:"<<szBuf<<endl; return 0; } int WriteEvent() { return 0; } }; class CAcceptEvent : public IEvent { public: int ReadEvent() { CReadEvent* pReadEvent = new CReadEvent();//客户端的Event SOCKADDR_IN stClientAddr = {0}; int nLen = sizeof(SOCKADDR_IN); SOCKET hClient = accept(GetSocket(),(SOCKADDR*)&stClientAddr,&nLen);//Accept客户端 pReadEvent->SetEvent(hClient,enumEvent_Type_Read); pReadEvent->SetEventPersist(TRUE); pSelectReactor->RegisterEvent(pReadEvent);//注册到Reactor中 return 0; } int WriteEvent() { cout<<"WriteEvent"<<endl; return 0; } CAcceptEvent(CSelectReactor* pSelectReactor) { this->pSelectReactor = pSelectReactor; } protected: CSelectReactor* pSelectReactor; }; E.Main函数： int main() { CSelectReactor cSelect; CAcceptEvent* pEvent = new CAcceptEvent(&cSelect); SOCKADDR_IN stSockAddr = {0}; stSockAddr.sin_addr.s_addr = INADDR_ANY; stSockAddr.sin_family = AF_INET; stSockAddr.sin_port = htons(9999); SOCKET hSocket = socket(AF_INET,SOCK_STREAM,0); bind(hSocket,(SOCKADDR*)&stSockAddr,sizeof(SOCKADDR)); listen(hSocket,5); pEvent->SetEvent(hSocket,enumEvent_Type_Read);//增加一个Listen的Event pEvent->SetEventPersist(TRUE); cSelect.RegisterEvent(pEvent);//注册到Reacotr中 cSelect.Dispatch();//Select多路复用 return 0; } |
