1.EasyTcpServer服务端的使用。
2.客户端 发送 登录消息
服务端接收登录消息 返回消息给 客户端
存在问题:网络抖动较大
客户端:
DataHeader.hpp
EasyTcpClient.hpp
main.cpp
服务端:继承EasyTcpServer,重写接口函数
DataHeader.hpp
同客户端
CELLTimestamp.hpp
EasyTcpServer.hpp
#ifndef _EasyTcpServer_hpp_ #define _EasyTcpServer_hpp_ #ifdef _WIN32 #define FD_SETSIZE 2506 #define WIN32_LEAN_AND_MEAN #define _WINSOCK_DEPRECATED_NO_WARNINGS //也可以放到工程属性,预处理中 #define _CRT_SECURE_NO_WARNINGS #include#include #pragma comment(lib, "ws2_32.lib") #else #include #include #include<string.h> #define SOCKET int #define INVALID_SOCKET (SOCKET)(~0) #define SOCKET_ERROR (-1) #endif #include #include #include #include #include #include "DataHeader.hpp" #include "CELLTimestamp.hpp" #define _CELLSERVERS_THREAD_COUNT 4 #define RECV_BUFF_SIZE 10240 //客户端数据类型 class ClientSocket { public: ClientSocket(SOCKET sockfd = INVALID_SOCKET) { _sockfd = sockfd; //memset(_szMsgBuf, 0, sizeof(_szMsgBuf)); } SOCKET sockfd() { return _sockfd; } char* msgBuf() { return _szMsgBuf; } int getlastPos() { return _lastPos; } void setlastPos(int pos) { _lastPos = pos; } //指定socket发送数据 int SendData(DataHeader* header) { if (header) { return send(_sockfd, (const char*)header, header->dataLength, 0); } return SOCKET_ERROR; } private: SOCKET _sockfd = INVALID_SOCKET; char _szMsgBuf[RECV_BUFF_SIZE * 5] = {}; int _lastPos = 0; }; //网络事件接口 class INetEvent { public: //客户端加入事件 virtual void OnNetJoin(ClientSocket* pClient) = 0; //客户端离开事件 virtual void OnNetLeave(ClientSocket* pClient) = 0; //客户端消息事件 virtual void OnNetMsg(ClientSocket* pClient, DataHeader* header) = 0; }; class CellServer { public: CellServer(SOCKET sock = INVALID_SOCKET) { _sock = sock; _pNetEvent = nullptr; } ~CellServer() { Close(); _sock = INVALID_SOCKET; } //关闭socket void Close() { #ifdef _WIN32 // 关闭套接字 for (size_t n = 0; n<_clients.size>) { closesocket(_clients[n]->sockfd()); delete _clients[n]; } #else for (size_t n = 0; n<_clients.size>) { close(_clients[n]->sockfd()); delete _clients[n]; } #endif _clients.clear(); } //是否工作中 bool isRun() { return _sock != INVALID_SOCKET; } void Start() { //std::thread t(&CellServer::onRun, this); _pThread = std::thread(std::mem_fn(&CellServer::onRun), this); } //接收数据 处理粘包 拆分包 int RecvData(ClientSocket* pClient) { // 接收客户端数据 int nLen = (int)recv(pClient->sockfd(), _szRecv, RECV_BUFF_SIZE, 0); //printf("nLen=%d\n", nLen); if (nLen 0) { //printf("客户端%d已退出。\n", pClient->sockfd()); return -1; } memcpy(pClient->msgBuf() + pClient->getlastPos(), _szRecv, nLen); //消息缓冲区的数据尾部位置后移 pClient->setlastPos(pClient->getlastPos() + nLen); //判断消息缓冲区的数据长度大于消息头DataHeader长度 while (pClient->getlastPos() >= sizeof(DataHeader)) { //这时就可以知道当前消息的长度 DataHeader* header = (DataHeader*)pClient->msgBuf(); //判断是否可以获取一个完整消息 if (pClient->getlastPos() >= header->dataLength) { //剩余的未处理消息缓冲区数据的长度 int nSize = pClient->getlastPos() - header->dataLength; //处理网络消息 onNetMsg(pClient, header); //将消息缓冲区剩余未处理数据前移 memcpy(pClient->msgBuf(), pClient->msgBuf() + header->dataLength, nSize); //将消息缓冲区的数据尾部位置前移 pClient->setlastPos(nSize); } else { //消息缓冲区剩余数据不够一条完整消息 break; } } return 0; } //响应网络消息 virtual void onNetMsg(ClientSocket* pClient, DataHeader* header) { _pNetEvent->OnNetMsg(pClient, header); switch (header->cmd) { case CMD_LOGIN: { Login *pLogin = (Login*)header; //printf("客户端%d,cmd logint,user:%s,password:%s\n", _cSock, pLogin->userName, pLogin->passWord); //---------------------------- //---------------------------- LoginResult longinRet = {}; pClient->SendData(&longinRet); } break; case CMD_LOGOUT: { LogOut *pLogOut = (LogOut*)header; printf("客户端%d,cmd logout,user:%s\n", pClient->sockfd(), pLogOut->userName); //---------------------------- LogOutResult logOutRet = {}; } break; default: { DataHeader header = {}; header.cmd = CMD_ERROR; } break; } } bool onRun() { while ( isRun() ) { if (_clientsBuff.size() > 0) { std::lock_guard<:mutex> lock(_mutex); for (auto pClient : _clientsBuff) { _clients.push_back(pClient); } _clientsBuff.clear(); } //如果没有需要处理的客户端,就跳过 if (_clients.empty()) { std::chrono::milliseconds t(1); std::this_thread::sleep_for(t); continue; } fd_set fdRead; //fd_set fdWrite; //fd_set fdExp; FD_ZERO(&fdRead); //FD_ZERO(&fdWrite); //FD_ZERO(&fdExp); SOCKET maxSock = _clients[0]->sockfd(); for (size_t n = 0; n<_clients.size>) { FD_SET(_clients[n]->sockfd(), &fdRead); if (maxSock < _clients[n]->sockfd()) { maxSock = _clients[n]->sockfd(); } } timeval t = { 0, 1}; //int ret = select(maxSock + 1, &fdRead, &fdWrite, &fdExp, &t); int ret = select(maxSock + 1, &fdRead, nullptr, nullptr, nullptr); if (ret < 0) { printf("select任务结束。\n"); Close(); return false; } for (int n = (int)_clients.size() - 1; n >= 0; n--) { if (FD_ISSET(_clients[n]->sockfd(), &fdRead)) { if (-1 == RecvData(_clients[n])) { auto iter = _clients.begin() + n; if (iter != _clients.end()) { if (_pNetEvent) { _pNetEvent->OnNetLeave(_clients[n]); } delete _clients[n]; _clients.erase(iter); } } } } } return true; } void addClient(ClientSocket* pClient) { _mutex.lock(); _clientsBuff.push_back(pClient); _mutex.unlock(); } size_t getClientCount() { return _clients.size() + _clientsBuff.size(); } void setEventObj(INetEvent* event) { _pNetEvent = event; } private: SOCKET _sock; //正式客户队列 std::vector _clients; //缓冲客户队列 std::vector *>_clientsBuff; //缓冲客户端队列的锁 std::mutex _mutex; std::thread _pThread; char _szRecv[RECV_BUFF_SIZE] = { }; private: //网络事件对象 INetEvent* _pNetEvent; }; class EasyTcpServer:public INetEvent { public: EasyTcpServer() { _sock = INVALID_SOCKET; _clientCount = 0; _recvCount = 0; } virtual ~EasyTcpServer() { Close(); } //初始化socket SOCKET InitSocket() { #ifdef _WIN32 WORD ver = MAKEWORD(2, 2); WSADATA dat; WSAStartup(ver, &dat); #endif // _WIN32 if (_sock != INVALID_SOCKET) { Close(); } _sock = socket(AF_INET, SOCK_STREAM, 0); if (INVALID_SOCKET == _sock) { printf("错误,建立Socket失败...\n"); } else { printf("建立Socket成功...\n"); } return _sock; } //绑定端口号 int Bind(const char* ip, unsigned short port) { sockaddr_in _sin = {}; _sin.sin_family = AF_INET; _sin.sin_port = htons(4567);//host to net unsigned short //使用127.0.0.1可以防止外网访问 //启用本机全部的ip地址可以使用,INADDR_ANY #ifdef _WIN32 if (ip) { _sin.sin_addr.S_un.S_addr = inet_addr(ip); } else { _sin.sin_addr.S_un.S_addr = INADDR_ANY; } #else if (ip) { _sin.sin_addr.s_addr = inet_addr(ip); } else { _sin.sin_addr.s_addr = INADDR_ANY; } #endif int ret = bind(_sock, (sockaddr*)&_sin, sizeof(sockaddr_in)); if (SOCKET_ERROR == ret) { printf("错误,绑定网络端口失败...\n"); } else { printf("绑定网络端口成功...\n"); } return ret; } //监听端口号 int Listen(int cnt) { int ret = listen(_sock, cnt); if (SOCKET_ERROR == ret) { printf("错误,监听网络端口失败...\n"); } else { printf("监听网络端口成功...\n"); } return ret; } //接收客户端连接 int Accept() { sockaddr_in clientAddr = {}; int nAddrLen = sizeof(clientAddr); SOCKET _cSock = INVALID_SOCKET; #ifdef _WIN32 _cSock = accept(_sock, (sockaddr*)&clientAddr, &nAddrLen); #else _cSock = accept(_sock, (sockaddr*)&clientAddr, (socklen_t*)&nAddrLen); #endif if (INVALID_SOCKET == _cSock) { printf("错误,接受到无效客户端socket...\n"); } else { //NewUserJoin msg; //msg.sock = _cSock; //Send2All(&msg); //_clients.push_back(new ClientSocket(_cSock)); //将新客户端分配给客户端数量最少的cellServer addClientToCellServer(new ClientSocket(_cSock)); //printf("new user join in:socket=%d,IP=%s,count=%d\n", _cSock, inet_ntoa(clientAddr.sin_addr), _clients.size()); } return _cSock; } void Start(int nCellServer) { for (int n=0; n *>) { auto ser = new CellServer(_sock); _cellServers.push_back(ser); //注册网络事件接收对象 ser->setEventObj(this); //启动消息处理线程 ser->Start(); } } //关闭socket void Close() { if (_sock != INVALID_SOCKET) { #ifdef _WIN32 closesocket(_sock); //清除Windows socket环境 WSACleanup(); #else close(_sock); #endif _sock = INVALID_SOCKET; } } void addClientToCellServer(ClientSocket* pClient) { OnNetJoin(pClient); //查找客户数量最少的CellServer消息处理对象 auto pMinServer = _cellServers[0]; for (auto pCellServer : _cellServers) { if (pCellServer->getClientCount() < pMinServer->getClientCount()) { pMinServer = pCellServer; } } pMinServer->addClient(pClient); } //处理网络消息 bool OnRun() { if (!isRun()) { return false; } time4msg(); fd_set fdRead; //fd_set fdWrite; //fd_set fdExp; FD_ZERO(&fdRead); //FD_ZERO(&fdWrite); //FD_ZERO(&fdExp); FD_SET(_sock, &fdRead); //FD_SET(_sock, &fdWrite); //FD_SET(_sock, &fdExp); SOCKET maxSock = _sock; timeval t = {0, 10}; //int ret = select(maxSock + 1, &fdRead, &fdWrite, &fdExp, &t); int ret = select(maxSock + 1, &fdRead, 0, 0, &t); if (ret < 0) { printf("select任务结束。\n"); Close(); return false; } if (FD_ISSET(_sock, &fdRead)) { FD_CLR(_sock, &fdRead);//从集合移除了_sock Accept(); } return true; } //是否工作中 bool isRun() { return _sock != INVALID_SOCKET; } //计算每秒钟收到了多少网络消息 void time4msg() { auto t1 = _tTime.getElapsedTimeInSec(); if (t1 > 1.0) { printf("time:%lf,sock:%d,clients:%d,recv count:%d\n", t1, _sock, _clientCount, (int)(_recvCount/t1)); _recvCount = 0; _tTime.update(); } } //void Send2All(DataHeader* header) //{ // if (isRun() && header) // { // for (int n = (int)_clients.size() - 1; n >= 0; n--) // { // SendData(_clients[n]->sockfd(), header); // } // } //} virtual void OnNetJoin(ClientSocket* pClient) { _clientCount++; } virtual void OnNetLeave(ClientSocket* pClient) { _clientCount--; } virtual void OnNetMsg(ClientSocket* pClient, DataHeader* header) { _recvCount++; } private: SOCKET _sock; //消息处理对象,内部会创建线程 std::vector ;>_cellServers; CELLTimestamp _tTime; protected: std::atomic<int> _clientCount; //收到消息计数 std::atomic<int> _recvCount; }; #endif *>
main.cpp
Original: https://www.cnblogs.com/zhangxuan/p/14448900.html
Author: 邶风
Title: 29.多线程,分组服务端,1万客户端连接,服务端使用
原创文章受到原创版权保护。转载请注明出处:https://www.johngo689.com/539838/
转载文章受原作者版权保护。转载请注明原作者出处!