EventLoop框架C++服务端（Tcp/Ip）搭建---完善

以下使用VS2022进行演示。

在此之前已经实现了初步版本（https://blog.csdn.net/qq135595696/article/details/125475334?spm=1001.2014.3001.5501），本次对其进行完善，思路仅供参考。

#ifndef UNETCORE_H_
#define UNETCORE_H_

//表示当前IO内核使用的是select模型
#define U_SELECT_NETCORE
#define _WINSOCK_DEPRECATED_NO_WARNINGS
#include 

namespace U
{
	//EventLoop抽象类，事件循环机制
	class IEventLoop
	{
	public:
		virtual void Init() = 0;

		virtual void LoopOnce() = 0;

		virtual void UnInit() = 0;
	};

	//向客户端发送数据后的响应回调函数
	typedef std::function FTcpSocketWriteCB;

	//用于管理连接上来的客户端
	//TODO
	class ITcpSocket
	{
	public:
		virtual void Init(IEventLoop* loop, FTcpSocketWriteCB writeCb) = 0;
	};

	//响应客户端连接回调函数
	typedef std::function FTcpServerCB;

	//ITcpServer抽象类，服务端
	class ITcpServer
	{
	public:
		/// 

		/// 初始化TCP服务，与循环事件进行关联
		/// 
		/// 事件循环机制
		/// 当用户连接成功后的回调函数
		/// 
		virtual bool Init(IEventLoop* loop, FTcpServerCB cb) = 0;

		/// 

		/// 初始化socket
		/// 
		/// 服务端IP
		/// 监听端口号
		/// 错误返回false
		virtual bool Listen(const char* ip, int port) = 0;
	};

	//加载Windows网络库
	void InitNetCore();

	//卸载Windows网络库
	void UnNetCore();

	//实例化事件循环机制
	IEventLoop* CreateEventLoop();

	//实例化服务端类
	ITcpServer* CreateTcpServer();
}
#endif // !UNETCORE_H_

#include "UNetCore.h"

#ifdef U_SELECT_NETCORE
#include 
#include 
#include 
#pragma comment(lib,"ws2_32")

#ifndef EVENTLOOP_H_
#define EVENTLOOP_H_


namespace U
{
	//保存服务端单元的信息。即与服务端句柄关联，保存至事件循环机制中
	class sEvent
	{
	public:
		//该服务端是属于什么类型的（TCP、UDP、PIPE）
		enum class Type
		{
			E_TCPSERVER,
			//UDP，PIPE
			E_TCPCLIENT,
		};
		Type type;
		SOCKET sock;//保存服务端句柄
		//由于sEvent类必须要附带服务端类，那么如果不使用联合体，会造成过度的内存资源浪费，即这里使用联合体保存各种类型服务端类的指针。
		union
		{
			class TcpServer* tcpServer;
			//class UcpServer* ucpServer;
			//...
			class TcpSocket* tcpSocket;
		};
	};


	class EventLoop :public IEventLoop
	{
	private:
		std::unordered_map _events;
	public:
		void Init() override;

		void LoopOnce() override;

		void UnInit() override;

		//添加监听的循环事件，压入sEvent单元
		void AddEvent(sEvent* event);
	};
}

#endif // !EVENTLOOP_H_

#endif // U_SELECT_NETCORE

#include "EventLoop.h"
#include "TcpServer.h"

#ifdef U_SELECT_NETCORE

namespace U
{
	void InitNetCore()
	{
		WSADATA wsa;
		WSAStartup(MAKEWORd(2, 2), &wsa);

		//TODO 可以添加异常导出库，在程序崩溃生成崩溃信息
	}

	void UnNetCore()
	{
		WSACleanup();
	}

	IEventLoop* CreateEventLoop()
	{
		return new EventLoop;
	}
}

void U::EventLoop::Init()
{

}

//采用select IO模型
void U::EventLoop::LoopOnce()
{
	FD_SET reads;
	FD_ZERO(&reads);

	for (auto& event : _events)
		FD_SET(event.first, &reads);

	int nRet = select(0, &reads, nullptr, nullptr, nullptr);
	if (nRet <= 0)
		return;

	auto begin = _events.begin();
	auto end = _events.end();
	for (; begin != end;)
	{
		SOCKET sock = begin->first;
		sEvent* pEvent = begin->second;
		bool isEnd = false;
		if (FD_ISSET(sock, &reads))
		{
			switch (pEvent->type)
			{
			case U::sEvent::Type::E_TCPSERVER://处理TCP类的服务端
				pEvent->tcpServer->OnAccept();
				break;
			case U::sEvent::Type::E_TCPCLIENT:
			{
				//客户端断开连接
				if (!pEvent->tcpSocket->OnRecv())
				{
					std::cout << "客户端断开连接" << std::endl;
					pEvent->tcpSocket->CLoseSock();
					pEvent->tcpSocket->_tcpServer->CloseTcpSock(pEvent->tcpSocket);
					begin = _events.erase(begin);
					if (begin == end)
						isEnd = true;
					break;
				}
				//处理接收到的客户端消息
				int len = pEvent->tcpSocket->_tcpServer->OnNetMsg(pEvent->tcpSocket);
				pEvent->tcpSocket->UseBuf(len);

				FD_SET writes;
				FD_ZERO(&writes);
				FD_SET(pEvent->tcpSocket->GetSock(), &writes);
				int nWRet = select(0, nullptr, &writes, nullptr, nullptr);
				if (nWRet > 0)
				{
					if (FD_ISSET(pEvent->tcpSocket->GetSock(), &writes))
						pEvent->tcpSocket->Send();
				}
				break;
			}
			default:
				break;
			}
		}
		if (!isEnd)
			begin++;
	}
}

void U::EventLoop::UnInit()
{

}


void U::EventLoop::AddEvent(sEvent* event)
{
	std::cout << "添加　事件" << event->sock
		<< " type:" << (int)event->type
		<< " serverPtr:" << event->tcpServer
		<< std::endl;
	_events.insert(std::make_pair(event->sock, event));
}

#endif // U_SELECT_NETCORE

#include "UNetCore.h"

#ifdef U_SELECT_NETCORE
#include "EventLoop.h"
#ifndef TCPSERVER_H_
#define TCPSERVER_H_

namespace U
{
	class TcpSocket :public ITcpSocket
	{
	private:
		EventLoop* _loop;//事件循环机制
		SOCKET _clientSock;

		//接收缓冲区
		char _recvBuf[1024];
		int _recvLen;
		int _recvMaxLen;

		//发送缓冲区
		char _sendBuf[1024];
		int _sendLen;
		int _sendMaxLen;
	public:
		FTcpSocketWriteCB _writeCb;
		sEvent _event;
		class TcpServer* _tcpServer;
	public:
		void Init(IEventLoop* loop, FTcpSocketWriteCB writeCb) override;

		TcpSocket(SOCKET clientSock, class TcpServer* tcpServer);

		~TcpSocket();

		bool OnRecv();

		void UseBuf(int len);

		const char* GetBuf() { return _recvBuf; }

		int GetLen() { return _recvLen; }

		void Send();

		int OnSend(const char* buf, int len);

		SOCKET GetSock() { return _clientSock; }

		void CLoseSock()
		{
			if (INVALID_SOCKET == _clientSock)
				return;
			closesocket(_clientSock);
		}
	};


	class TcpServer :public ITcpServer
	{
	private:
		EventLoop* _loop;//事件循环机制
		FTcpServerCB _cb;//客户端连接回调事件
		SOCKET _sock;//服务端socket
		sEvent _event;//往事件循环机制注册消息的最小单元，即用于注册循环事件
		std::unordered_map _tcs;
	public:
		TcpServer();

		~TcpServer();

		bool Init(IEventLoop* loop, FTcpServerCB cb) override;

		bool Listen(const char* ip, int port) override;

	public:
		//处理客户端连接
		void OnAccept();

		int OnNetMsg(TcpSocket* client);

		void CloseTcpSock(TcpSocket* client);
	};
}

#endif // !TCPSERVER_H_
#endif // U_SELECT_NETCORE

#include "TcpServer.h"
#ifdef U_SELECT_NETCORE

namespace U
{
	ITcpServer* CreateTcpServer()
	{
		return new TcpServer;
	}
}

void U::TcpSocket::Init(IEventLoop* loop, FTcpSocketWriteCB writeCb)
{
	_loop = dynamic_cast(loop);
	_writeCb = writeCb;
}


U::TcpSocket::TcpSocket(SOCKET clientSock, class TcpServer* tcpServer)
{
	_event.tcpSocket = this;
	_event.type = U::sEvent::Type::E_TCPCLIENT;
	_event.sock = clientSock;

	_clientSock = clientSock;
	_tcpServer = tcpServer;

	_recvLen = 0;
	_recvMaxLen = 1024;

	_sendLen = 0;
	_sendMaxLen = 1024;
}

U::TcpSocket::~TcpSocket()
{
	if (INVALID_SOCKET != _clientSock)
		closesocket(_clientSock);
	_clientSock = INVALID_SOCKET;
}

bool U::TcpSocket::OnRecv()
{
	int nRecv = recv(_clientSock, _recvBuf + _recvLen, _recvMaxLen - _recvLen, 0);
	if (nRecv <= 0)
		return false;
	_recvLen += nRecv;
	return true;
}

void U::TcpSocket::UseBuf(int len)
{
	memmove(_recvBuf, _recvBuf + len, _recvLen - len);
	_recvLen -= len;
}

void U::TcpSocket::Send()
{
	if (0 == _sendLen)return;
	int nSendLen = send(_clientSock, _sendBuf, _sendLen, 0);
	_sendLen -= nSendLen;
}

int U::TcpSocket::OnSend(const char* buf, int len)
{
	if (_sendLen == _sendMaxLen)
		return 0;
	if (_sendLen + len > _sendMaxLen)
	{
		memmove(_sendBuf + _sendLen, buf, _sendMaxLen - _sendLen);
		_sendLen = _sendMaxLen;
		return _sendMaxLen;
	}
	memmove(_sendBuf + _sendLen, buf, len);
	_sendLen += len;
	return _sendLen;
}

U::TcpServer::TcpServer()
{
	std::cout << "初始化　TCPServer " << std::endl;

	_loop = nullptr;
	_cb = nullptr;
	_sock = INVALID_SOCKET;

	_event.tcpServer = this;
	_event.type = U::sEvent::Type::E_TCPSERVER;
	_event.sock = INVALID_SOCKET;
}

U::TcpServer::~TcpServer()
{
	_loop = nullptr;
	_cb = nullptr;
	if (INVALID_SOCKET != _sock)
		closesocket(_sock);
	_sock = INVALID_SOCKET;
}

bool U::TcpServer::Init(IEventLoop* loop, FTcpServerCB cb)
{
	_loop = dynamic_cast(loop);
	_cb = cb;

	return true;
}

bool U::TcpServer::Listen(const char* ip, int port)
{
	std::cout << "Listen IP:" << ip << " port:" << port << std::endl;

	//TODO 待完善
	if (_sock != INVALID_SOCKET)
	{
		closesocket(_sock);
		return false;
	}

	_sock = socket(AF_INET, SOCK_STREAM, 0);
	SOCKADDR_IN addr;
	addr.sin_family = AF_INET;
	addr.sin_port = htons(port);
	addr.sin_addr.s_addr = inet_addr(ip);
	if (SOCKET_ERROR == bind(_sock, (sockaddr*)&addr, sizeof(addr)))
		return false;

	if (SOCKET_ERROR == listen(_sock, 5))
		return false;

	_event.sock = _sock;
	_loop->AddEvent(&_event);
	return true;
}


void U::TcpServer::OnAccept()
{
	//TODO 未处理客户端socket（句柄）
	SOCKADDR_IN addr;
	int addrLen = sizeof(SOCKADDR_IN);
	SOCKET sock = accept(_sock, (sockaddr*)&addr, &addrLen);
	if (INVALID_SOCKET == sock)
		return;
	TcpSocket* p = new TcpSocket(sock, this);
	_tcs.insert(std::make_pair(sock, p));

	auto writeCb = []() {
		std::cout << "向客户端发送消息后的回调函数" << std::endl;
	};

	p->Init(_loop, writeCb);
	_loop->AddEvent(&p->_event);

	//连接成功后调用回调函数
	_cb(p);
}


int U::TcpServer::OnNetMsg(TcpSocket* client)
{
	std::cout << "接收客户端消息后的回调函数" << std::endl;
	std::cout << "接收到客户端数据长度:" << client->GetLen() << std::endl;
	client->OnSend(client->GetBuf(), client->GetLen());
	return client->GetLen();
}

void U::TcpServer::CloseTcpSock(TcpSocket* client)
{
	auto item = _tcs.find(client->GetSock());
	if (item != _tcs.end())
	{
		free(item->second);
		_tcs.erase(item);
	}
}

#endif // U_SELECT_NETCORE

#include "UNetCore.h"
#include 

int main()
{
	//加载Windows网络库
	U::InitNetCore();

	//调用EventLoop循环机制
	U::IEventLoop* loop = U::CreateEventLoop();
	//开启两个服务端
	U::ITcpServer* server = U::CreateTcpServer();
	U::ITcpServer* server1 = U::CreateTcpServer();

	//将服务端与事件循环机制loop关联
	server->Init(loop, [](U::ITcpSocket* sock) {
		std::cout << "客户端1连接" << std::endl;
		});
	server1->Init(loop, [](U::ITcpSocket* sock) {
		std::cout << "客户端2连接" << std::endl;
		});

	server->Listen("0.0.0.0", 7890);
	server1->Listen("0.0.0.0", 7891);

	//启动循环机制监听消息
	while (true)
		loop->LoopOnce();

	//卸载Windows网络库
	U::UnNetCore();
	return 0;
}