006.反应器接口

缺乏、安全感 2023-06-30 01:48 2阅读 0赞

/*******************************************************************
    ** 文件名:	IReactor.h
    ** 描  述:
    ** 应  用:
    ********************************************************************/
    #pragma once
    
    /// 事件处理程序
    struct EventHandler;
    
    /**
    @name : 反应器接口 (Reactor)
    @brief: 反应器就是等待多个事件然后调用相应的处理程序的这么一个装置
    Windows下用WaitForMultipleObjects实现
    */
    struct IReactor
    {
        
    	/**
    	@name       : 填加要监听的事件
    	@brief      : 这些函数都是多线程安全的
    	@param event: 事件句柄
    	@return		  : true  ... 注册成功
    	false ... 已经达到最大事件个数
    	false ... 该事件不存在
    	*/
    	virtual bool AddEvent(HANDLE event) = 0;
    
    	/**
    	@name       : 移除要监听的事件
    	@brief      : 这些函数都是多线程安全的
    	@param event: 事件句柄
    	*/
    	virtual void RemoveEvent(HANDLE event) = 0;
    
    	/**
    	@name         : 注册一个事件 (一个Handler可以处理多个事件,一个事件可以由多个处理器处理)
    	@brief        : 这些函数都是多线程安全的
    	@param event  : 事件句柄
    	@param handler: 事件处理程序
    	@return		  : true  ... 注册成功
    	false ... 已经达到最大事件个数
    	false ... 该事件不存在
    	*/
    	virtual bool RegisterEventHandler(HANDLE event, EventHandler * handler) = 0;
    
    	/**
    	@name         : 取消一个事件处理器
    	@brief        : 这些函数都是多线程安全的
    	@note         : 该处理器不再处理这个事件，但是这个事件依旧监听，只有调用RemoveEvent才不再监听事件
    	@param event  : 事件句柄
    	@param handler: 事件处理程序
    	@return		  :
    	*/
    	virtual void UnRegisterEventHandler(HANDLE event, EventHandler * handler) = 0;
    
    	/**
    	@purpose      : 等待事件触发并分派
    	@param	wait  : 等待的毫秒数
    	@return       : 等待的过程中是否出错
    	*/
    	virtual bool HandleEvents(DWORD wait) = 0;
    
    	/**
    	@purpose      : 循环等待/分派事件
    	*/
    	virtual bool HandleEventsLoop() = 0;
    
    	/**
    	@name         : 通知反应器线程退出
    	@brief        : 这些函数都是多线程安全的
    	*/
    	virtual void NotifyExit() = 0;
    
    	virtual void Release() = 0;
    };
    
    /**
    @name : 事件处理器
    @brief:
    */
    struct EventHandler
    {
        
    	virtual void OnEvent(HANDLE event) = 0;
    
    	// 通知注册/注销成功，注意，是在反应器线程里面回调这个接口
    	virtual void OnRegisterOK(HANDLE event) {
        }
    	virtual void OnUnRegisterOK(HANDLE event) {
        }
    };

/*******************************************************************
    ** 文件名:	Reactor.h
    ** 描  述:
    ** 应  用:
    ********************************************************************/
    #pragma once
    #include "Common.h"
    #include "IReactor.h"
    #include "singleton.h"
    #include "SwappingList.h"
    #include "Lock.h"
    
    
    // 轮询时间间隔
    #	define  POLLING_INTERVAL     1
    
    /**
    @name : 反应器 (Reactor)
    @brief: 反应器就是等待多个事件然后调用相应的处理程序的这么一个装置
    Windows下用WaitForMultipleObjects实现
    @warning: 注意:
    1.添加/删除事件需要注意多线程加锁
    2.支持多于64个事件的监听,多于64个事件时使用轮询方式检查事件
    */
    class Reactor : public IReactor, public EventHandler
    	, public SingletonEx<Reactor>
    {
        
    public:
    	/**
    	@purpose    : 填加要监听的事件
    	@param event: 事件句柄
    	@return		  : true  ... 注册成功
    	false ... 已经达到最大事件个数
    	false ... 该事件不存在
    	*/
    	virtual bool AddEvent(HANDLE event);
    
    	/**
    	@purpose    : 移除要监听的事件
    	@param event: 事件句柄
    	*/
    	virtual void RemoveEvent(HANDLE event);
    
    	/**
    	@purpose      : 注册一个事件 (一个Handler可以处理多个事件,一个事件可以由多个处理器处理)
    	@param event  : 事件句柄
    	@param handler: 事件处理程序
    	@return		  : true  ... 注册成功
    	false ... 已经达到最大事件个数
    	false ... 该事件不存在
    	*/
    	virtual bool RegisterEventHandler(HANDLE event, EventHandler * handler);
    
    	/**
    	@purpose      : 取消一个事件处理器
    	@note         : 该处理器不再处理这个事件，但是这个事件依旧监听，只有调用RemoveEvent才不再监听事件
    	@param event  : 事件句柄
    	@param handler: 事件处理程序
    	@return		  :
    	*/
    	virtual void UnRegisterEventHandler(HANDLE event, EventHandler * handler);
    
    	/**
    	@purpose      : 等待事件触发并分派
    	@param	wait  : 等待的毫秒数
    	@return       : 等待的过程中是否出错
    	*/
    	virtual bool HandleEvents(DWORD wait);
    
    	/**
    	@purpose      : 循环等待/分派事件
    	*/
    	virtual bool HandleEventsLoop();
    
    	/**
    	@name         : 通知反应器线程退出
    	@brief        : 这些函数都是多线程安全的
    	*/
    	virtual void NotifyExit();
    
    	/**
    	@name         : 处理自身的操作事件,用来做多线程同步
    	@brief        :
    	*/
    	virtual void OnEvent(HANDLE event);
    
    	virtual void Release() {
         delete this; }
    
    	Reactor();
    
    	virtual ~Reactor();
    
    protected:
    	// 真正的处理函数,这里已经到了反应器自己的线程,没有多线程同步的问题
    	bool AddEventAux(HANDLE event);
    	void RemoveEventAux(HANDLE event);
    	bool RegisterEventHandlerAux(HANDLE event, EventHandler * handler);
    	void UnRegisterEventHandlerAux(HANDLE event, EventHandler * handler);
    	void NotifyExitAux();
    
    protected:
    	typedef std::list<EventHandler *>     HANDLER_LIST;
    	typedef std::map<HANDLE, HANDLER_LIST> HANDLERS_MAP;
    
    	HANDLE    *  m_EventHandles;                        // 待监听的事件数组
    	DWORD        m_dwEventsNum;                         // 事件个数
    	DWORD        m_dwArrayCapability;                   // 数组容量
    	HANDLERS_MAP m_HandlersMap;                         // 事件处理器列表
    
    	HANDLE       m_hSelfEvent;                          // 反应器自己的维护事件
    	bool		 exit_flag;								// 退出标记
    	struct       REACTOR_TASK
    	{
        
    		enum
    		{
        
    			REACTOR_ADD = 0,
    			REACTOR_DELETE,
    			REACTOR_REGISTER,
    			REACTOR_UNREGISTER,
    			REACTOR_EXIT,
    		};
    
    		int      type;          // 操作类型
    		HANDLE   event;
    		EventHandler * handler;
    	};
    
    	SwappingList<REACTOR_TASK>   m_TaskList; // 任务列表
    	Mutex						 m_Mu;
    };

/*******************************************************************
    ** 文件名:	Reactor.cpp
    ** 描  述:
    ** 应  用:
    ********************************************************************/
    
    #include "stdafx.h"
    #include "Reactor.h"
    #include "Trace.h"
    
    //
    Reactor::Reactor()
    {
        
    	m_dwEventsNum = 0;
    	exit_flag = FALSE;
    	m_dwArrayCapability = MAXIMUM_WAIT_OBJECTS;
    	m_EventHandles = new HANDLE[m_dwArrayCapability];
    	memset(m_EventHandles, 0, m_dwArrayCapability*sizeof(HANDLE));
    
    	m_hSelfEvent = ::CreateEvent(NULL, FALSE, FALSE, 0);
    	AddEventAux(m_hSelfEvent);
    	RegisterEventHandlerAux(m_hSelfEvent, this);
    }
    
    //
    Reactor::~Reactor()
    {
        
    	if (m_EventHandles)
    	{
        
    		delete[]m_EventHandles;
    		m_EventHandles = 0;
    	}
    
    	if (m_hSelfEvent != INVALID_HANDLE_VALUE)
    	{
        
    		::CloseHandle(m_hSelfEvent);
    		m_hSelfEvent = INVALID_HANDLE_VALUE;
    	}
    }
    
    //
    bool Reactor::AddEventAux(HANDLE event)
    {
        
    	// 扩容
    	if (m_dwEventsNum >= m_dwArrayCapability)
    	{
        
    		m_dwArrayCapability += MAXIMUM_WAIT_OBJECTS;
    		HANDLE * newArray = new HANDLE[m_dwArrayCapability];
    		memcpy(newArray, m_EventHandles, m_dwEventsNum*sizeof(HANDLE));
    		delete[]m_EventHandles;
    		m_EventHandles = newArray;
    	}
    
    	for (DWORD i = 0; i < m_dwEventsNum; ++i)
    	{
        
    		if (m_EventHandles[i] == event)
    		{
        
    			return false;
    		}
    	}
    
    	m_EventHandles[m_dwEventsNum++] = event;
    	return true;
    }
    
    //
    void Reactor::RemoveEventAux(HANDLE event)
    {
        
    	for (DWORD i = 0; i < m_dwEventsNum; ++i)
    	{
        
    		if (m_EventHandles[i] == event)
    		{
        
    			m_EventHandles[i] = 0;
    		}
    	}
    
    	m_HandlersMap.erase(event);
    }
    
    //
    bool Reactor::RegisterEventHandlerAux(HANDLE event, EventHandler * handler)
    {
        
    	HANDLER_LIST & handlerList = m_HandlersMap[event];
    	HANDLER_LIST::iterator it = std::find(handlerList.begin(), handlerList.end(), handler);
    	if (it != handlerList.end())
    		return false;
    
    	handlerList.push_back(handler);
    	handler->OnRegisterOK(event);
    	return true;
    }
    
    //
    void Reactor::UnRegisterEventHandlerAux(HANDLE event, EventHandler * handler)
    {
        
    	HANDLER_LIST & handlerList = m_HandlersMap[event];
    	HANDLER_LIST::iterator it = std::find(handlerList.begin(), handlerList.end(), handler);
    	if (it != handlerList.end())
    	{
        
    		handlerList.erase(it);
    		handler->OnUnRegisterOK(event);
    	}
    }
    
    //
    void Reactor::NotifyExitAux()
    {
        
    	//::ExitThread(0);
    	exit_flag = true;
    }
    
    //
    bool Reactor::HandleEvents(DWORD wait)
    {
        
    	// 如果事件个数大于64就应该是polling模式,这时候不应该是INFINITE的等待时间
    	if (m_dwEventsNum > MAXIMUM_WAIT_OBJECTS)
    	{
        
    		if (wait == INFINITE)
    		{
        
    			wait = POLLING_INTERVAL;
    		}
    	}
    
    	// 统一清理已删除的事件
    	for (DWORD i = 0; i < m_dwEventsNum; ++i)
    	{
        
    		if (m_EventHandles[i] == 0)
    		{
        
    			if (i < m_dwEventsNum - 1)
    			{
        
    				memmove(&m_EventHandles[i], &m_EventHandles[i + 1], (m_dwEventsNum - i - 1)*sizeof(HANDLE));
    			}
    
    			--m_dwEventsNum;
    		}
    	}
    
    	// 没有事件就休眠
    	if (m_dwEventsNum == 0)
    	{
        
    		::Sleep(1);
    		return true;
    	}
    
    	DWORD offset = 0;
    
    	// 开始轮询
    	do
    	{
        
    		ulong wait_num = min(m_dwEventsNum - offset, MAXIMUM_WAIT_OBJECTS);
    		DWORD result = ::WaitForMultipleObjects(wait_num, m_EventHandles + offset, FALSE, wait);
    		if (result == WAIT_TIMEOUT)
    		{
        
    			offset += wait_num;
    			continue;
    		}
    
    		if (result == WAIT_FAILED)
    		{
        
    			// 应用层删除事件,因为Reactor的线程可能还在等待,
    			// 所以由于事件句柄已释放,这里可能会产生一个错误,待删除后就会正常了
    
    			OnEvent(m_hSelfEvent);
    			continue;
    		}
    
    		DWORD index = result - WAIT_OBJECT_0;
    		if (index >= 0 && index < wait_num)
    		{
        
    			// 派发给事件处理器
    			HANDLER_LIST & handlerList = m_HandlersMap[m_EventHandles[index + offset]];
    			HANDLER_LIST::iterator it = handlerList.begin();
    			for (; it != handlerList.end(); ++it)
    			{
        
    				EventHandler * handler = *it;
    				handler->OnEvent(m_EventHandles[index + offset]);
    
    				// 只要触发了一轮就一定要设置为POLLING模式，否则后面如果死循环监听就会出问题
    				wait = POLLING_INTERVAL;
    			}
    
    			// 如果是第一个事件表示有事件新增或减少,需要重新监听一下
    			if (index + offset == 0)
    			{
        
    				return !exit_flag;
    			}
    		}
    		else
    		{
        
    			TraceLn(0, "Reactor::HandleEvents error" << GetLastError());  // 奇怪的返回值
    			return true;
    		}
    		//继续检查下一组
    		offset += wait_num;
    
    	} while (!exit_flag && offset < m_dwEventsNum);
    
    	return !exit_flag;
    }
    
    //
    bool Reactor::HandleEventsLoop()
    {
        
    	// 大于64个事件自动转为轮询模式
    	while (HandleEvents(m_dwEventsNum > MAXIMUM_WAIT_OBJECTS ? POLLING_INTERVAL : INFINITE))
    	{
        
    	}
    
    	return true;
    }
    
    
    //
    bool Reactor::AddEvent(HANDLE event)
    {
        
    	REACTOR_TASK task;
    	task.event = event;
    	task.type = REACTOR_TASK::REACTOR_ADD;
    	task.handler = 0;
    	m_TaskList.Add(task);
    	SetEvent(m_hSelfEvent);
    	return true;
    }
    
    //
    void Reactor::RemoveEvent(HANDLE event)
    {
        
    	REACTOR_TASK task;
    	task.event = event;
    	task.type = REACTOR_TASK::REACTOR_DELETE;
    	task.handler = 0;
    
    	m_TaskList.Add(task);
    	SetEvent(m_hSelfEvent);
    	return;
    }
    
    //
    bool Reactor::RegisterEventHandler(HANDLE event, EventHandler * handler)
    {
        
    	REACTOR_TASK task;
    	task.event = event;
    	task.type = REACTOR_TASK::REACTOR_REGISTER;
    	task.handler = handler;
    
    	m_TaskList.Add(task);
    	SetEvent(m_hSelfEvent);
    	return true;
    }
    
    //
    void Reactor::UnRegisterEventHandler(HANDLE event, EventHandler * handler)
    {
        
    	REACTOR_TASK task;
    	task.event = event;
    	task.type = REACTOR_TASK::REACTOR_UNREGISTER;
    	task.handler = handler;
    
    	m_TaskList.Add(task);
    	SetEvent(m_hSelfEvent);
    	return;
    }
    
    //
    void Reactor::NotifyExit()
    {
        
    	REACTOR_TASK task;
    	task.event = 0;
    	task.type = REACTOR_TASK::REACTOR_EXIT;
    	task.handler = 0;
    
    	m_TaskList.Add(task);
    	SetEvent(m_hSelfEvent);
    	return;
    }
    
    //
    void Reactor::OnEvent(HANDLE event)
    {
        
    	if (event == INVALID_HANDLE_VALUE)
    		return;
    
    	REACTOR_TASK task;
    	while (m_TaskList.Get(task))
    	{
        
    		switch (task.type)
    		{
        
    		case REACTOR_TASK::REACTOR_ADD:
    			AddEventAux(task.event);
    			break;
    		case REACTOR_TASK::REACTOR_DELETE:
    			RemoveEventAux(task.event);
    			break;
    		case REACTOR_TASK::REACTOR_REGISTER:
    			RegisterEventHandlerAux(task.event, task.handler);
    			break;
    		case REACTOR_TASK::REACTOR_UNREGISTER:
    			UnRegisterEventHandlerAux(task.event, task.handler);
    			break;
    		case REACTOR_TASK::REACTOR_EXIT:
    			NotifyExitAux();
    			break;
    		default:
    			assert(false);
    			break;
    		}
    	}
    }

/*******************************************************************
    ** 文件名:	Api.h
    ** 描  述:
    ** 应  用:
    ********************************************************************/
    #pragma once
    #include "Common.h"
    #include "IReactor.h"
    
    /**
    @name         : 创建一个反应器
    @brief        : 通过GetAsynIoReactor获得的反应器是在IO线程监听的,如果
    			  : 你需要监听事件又不想在IO线程可以自己创建一个IReactor去监听
    */
    T20_API IReactor * CreateReactor();
    
    /**
    @name         : 取得用于异步IO的反应器指针
    @brief        : 其他有些地方可以用得到
    */
    T20_API IReactor * GetReactor();

/*******************************************************************
    ** 文件名:	Api.cpp
    ** 描  述:
    ** 应  用:
    ********************************************************************/
    #pragma once
    #include "stdafx.h"
    #include "Api.h"
    #include "Reactor.h"
    /**
    @name         : 创建一个反应器
    @brief        : 通过GetAsynIoReactor获得的反应器是在IO线程监听的,如果
    : 你需要监听事件又不想在IO线程可以自己创建一个IReactor去监听
    */
    T20_API IReactor * CreateReactor()
    {
        
    	return new Reactor;
    }
    
    /**
    @name         : 取得用于异步IO的反应器指针
    @brief        : 其他有些地方可以用得到
    */
    T20_API IReactor * GetReactor()
    {
        
    	return Reactor::getInstancePtr();
    }

测试

#include <iostream>
    #include "WorkerThreadPool.h"
    #include "Api.h"
    using namespace std;
    
    class RunnableReactor : public IRunnable
    {
        
    public:
    	void run()
    	{
        
    		GetReactor()->HandleEventsLoop();
    	};
    
    	void release(){
        };
    };
    
    class Runnable : public IRunnable, public EventHandler
    {
        
    public:
    	void run()
    	{
        
    		HANDLE hEvent = ::CreateEvent(NULL, FALSE, FALSE, NULL);
    		GetReactor()->AddEvent(hEvent);
    		GetReactor()->RegisterEventHandler(hEvent, this);
    		SetEvent(hEvent);
    	};
    	
    	void OnEvent(HANDLE event)
    	{
        
    		cout << "Runnable OnEvent 线程id: " << GetCurrentThreadId() << " event:" << event << endl;
    	}
    
    	void release(){
        };
    };
    
    void ThreadFun()
    {
        
    	for (size_t i = 0; i < 10; i++)
    	{
        
    		WorkerThreadPool::getInstancePtr()->add(new Runnable);
    	}
    	
    	Sleep(1000);
    	cout << "..." << endl;
    }
    
    void EndFun()
    {
        
    	HANDLE hEvent = ::CreateEvent(NULL, FALSE, FALSE, L"EndF");
    	::WaitForSingleObject(hEvent, INFINITE);
    	::CloseHandle(hEvent);
    }
    
    int _tmain(int argc, _TCHAR* argv[])
    {
        
    	new WorkerThreadPool;
    	CreateReactor();
    	WorkerThreadPool::getInstancePtr()->add(new RunnableReactor);
    
    	ThreadFun();
    
    	EndFun();
    	return 0;
    }

测试结果：  
![在这里插入图片描述][20200113202619490.png]

[20200113202619490.png]: https://img-blog.csdnimg.cn/20200113202619490.png