伺服器和客戶端簡單的使用select網路模型(三)
阿新 • • 發佈:2020-08-08
在二的基礎上收發資料進行了改進(資料接收增加了二次快取),提供吞吐量。
Server部分:一共3個檔案(server.cpp,EasyTcpServer.hpp和MessageHeader.hpp)
server.cpp:
#include"EasyTcpServer.hpp" #include<thread> bool g_bRun = true; void CmdThread() { while (true) { char cmdBuf[256] = {}; scanf("%s", cmdBuf); if (0== strcmp(cmdBuf, "exit")) { g_bRun = false; printf("退出cmdThread執行緒\n"); break; } else printf("不支援的命令。 \n"); } } int main() { EsayTcpServer server; server.initSocket(); server.Bind(nullptr, 10000); server.Listen(5); std::thread t1(CmdThread); t1.detach();while (g_bRun) { server.OnRun(); } server.Close(); printf("已退出\n"); return 0; }
EasyTcpServer.hpp:
#ifndef _EASY_TCP_SERVER_HPP_ #define _EASY_TCP_SERVER_HPP_ #define _AFXDLL #ifdef _WIN32 #include"afx.h" #include<windows.h> #include "WinSock2.h" #else #include<unistd.h> #include<arpa/inet.h> #include<string.h> #define SOCKET int #define INVALID_SOCKET (SOCKET)(~0) #define SOCKET_ERROR (-1) #endif #include<iostream> #include<vector> #include<stdlib.h> #include"MessageHeader.hpp" #define RECV_BUFF_SIZE 10240 class ClientSocket { public: ClientSocket(SOCKET sockfd = INVALID_SOCKET) { _sockfd = sockfd; ZeroMemory(_szMsgBuf, sizeof(_szMsgBuf)); _lastPos = 0; } ~ClientSocket() { } SOCKET sockfd() { return _sockfd; } char * msgBuf() { return _szMsgBuf; } int getLastPos() { return _lastPos; } void setLastPos(int pos) { _lastPos = pos; } private: SOCKET _sockfd; char _szMsgBuf[RECV_BUFF_SIZE * 10]; //訊息緩衝區資料尾部位置 int _lastPos; }; class EsayTcpServer { public: SOCKET _sock; //訊息緩衝區 char _szRecv[RECV_BUFF_SIZE]; std::vector<ClientSocket*> _clients; public: EsayTcpServer() { _sock = INVALID_SOCKET; ZeroMemory(_szRecv, sizeof(_szRecv)); } virtual ~EsayTcpServer() { Close(); } //初始化Socket SOCKET initSocket() { // 啟動Windows socket 2.x環境 #ifdef _WIN32 WORD ver = MAKEWORD(2, 2); WSADATA dat; WSAStartup(ver, &dat); #endif //---------------- // 建立一個socket if (INVALID_SOCKET != _sock) { printf("<sock=%d>關閉之前的連線...\n", _sock); 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(port); //host to net unsigned short #ifdef _WIN32 if (ip != NULL) { sin.sin_addr.S_un.S_addr = inet_addr(ip); } else { sin.sin_addr.S_un.S_addr = INADDR_ANY; //本機上的外網和區域網ip任意一個拿來用 } #else if (ip != NULL) { sin.sin_addr.s_addr = inet_addr(ip); //本機上的外網和區域網ip任意一個拿來用 } else { sin.sin_addr.s_addr = INADDR_ANY; //本機上的外網和區域網ip任意一個拿來用 } #endif //2繫結 if (SOCKET_ERROR == bind(_sock, (sockaddr *)&sin, sizeof(sockaddr_in))) { //TRACE(_T("ERROR:繫結socket失敗\n")); return 0; } } //監聽埠 int Listen(int n) { int ret = listen(_sock, n); if (SOCKET_ERROR == ret) { TRACE(_T("ERROR:socket監聽失敗")); } else { printf("監聽開始\n"); } return ret; } //接受客戶端連線 SOCKET Accept() { //4接收客戶端連線 SOCKET cSock = INVALID_SOCKET; sockaddr_in clientAddr = {}; int nAddrLen = sizeof(sockaddr_in); cSock = accept(_sock, (sockaddr*)&clientAddr, &nAddrLen); if (INVALID_SOCKET == cSock) { printf("接收客戶端無效\n"); } else { //廣播所有客戶端,新加入一個新的客戶端 NewUserJoin newUser; SendDataToAll(&newUser); _clients.push_back(new ClientSocket(cSock)); printf("新客戶端加入socket = %d,ip:%s\n", cSock, inet_ntoa(clientAddr.sin_addr)); } return cSock; } //關閉socket void Close() { if (_sock != INVALID_SOCKET) { #ifdef _WIN32 for (int n = _clients.size() - 1; n >= 0; n--) { closesocket(_clients[n]->sockfd()); if (_clients[n] != NULL) { delete _clients[n]; } } WSACleanup(); #else for (int n = _clients.size() - 1; n >= 0; n--) { close(_clients[n]->sockfd()); if (_clients[n] != NULL) { delete _clients[n]; } } #endif _clients.clear(); _sock = INVALID_SOCKET; } } int nCount = 0; //處理網路訊息 bool OnRun() { fd_set fdRead; fd_set fdWrite; fd_set fdExp; FD_ZERO(&fdRead); FD_ZERO(&fdWrite); FD_ZERO(&fdExp); //將要監聽的socket新增到陣列 FD_SET(_sock, &fdRead); FD_SET(_sock, &fdWrite); FD_SET(_sock, &fdExp); SOCKET maxSock = _sock; //每次進來都把所有要監視的socket新增到相應的陣列中 for (std::vector<ClientSocket*>::iterator it = _clients.begin(); it != _clients.end(); it++) { FD_SET((*it)->sockfd(), &fdRead); if (maxSock < (*it)->sockfd()) { maxSock = (*it)->sockfd(); } } //開始查詢,會清空3個數組所有的socket,然後如果哪個socket有事件傳送,則會把此socket放入到陣列返回 //當有客戶端連線或者客戶端傳送訊息都會進來 timeval time = { 1, 0 }; /*int select( IN int nfds, //windows下無意義,linux有意義IN OUT fd_set* readfds, //檢查可讀性 IN OUT fd_set* writefds, //檢查可寫性 IN OUT fd_set* exceptfds, //例外資料 IN const struct timeval* timeout); //函式的返回時間*/ int ret = select(maxSock + 1, &fdRead, &fdWrite, &fdExp, &time); if (ret < 0) { printf("select任務結束\n"); Close(); return false; } if (FD_ISSET(_sock, &fdRead)) { FD_CLR(_sock, &fdRead); //4接收客戶端連線 Accept(); } for (size_t n = 0; n < fdRead.fd_count; n++) { if (FD_ISSET(_clients[n]->sockfd(),&fdRead)) if (-1 == RecvData(_clients[n])) { auto iter = _clients.begin() + n; if (iter != _clients.end()) { delete _clients[n]; _clients.erase(iter); } } } } //是否在工作中 bool isRun() { return _sock != INVALID_SOCKET; } // 接受資料,處理粘包,拆分包 int RecvData(ClientSocket* pClient) { //接受客戶端請求資料 int nLen = recv(pClient->sockfd(), _szRecv, RECV_BUFF_SIZE, 0); if (nLen <= 0) { printf("客戶端<socket=%d>已退出,任務結束\n", pClient->sockfd()); return -1; } memcpy(pClient->msgBuf() + pClient->getLastPos(), _szRecv, nLen); pClient->setLastPos(pClient->getLastPos() + nLen); //判斷訊息緩衝區的資料長度大於訊息頭 while (pClient->getLastPos() >= sizeof(DataHeader)) { DataHeader* header = (DataHeader*)pClient->msgBuf(); //判斷訊息緩衝區的資料長度大於訊息體 if (pClient->getLastPos() >= header->dataLength) { //剩餘未處理訊息緩衝區資料的長度 int nSize = pClient->getLastPos() - header->dataLength; OnNetMsg(pClient->sockfd(), header); //將未處理的資料前移 memcpy(pClient->msgBuf(), pClient->msgBuf() + header->dataLength, nSize); pClient->setLastPos(nSize); } else { //剩餘資料小於一個訊息體的大小 break; } } return 0; } // 響應網路訊息 virtual void OnNetMsg(SOCKET cSock, DataHeader* header) { switch (header->cmd) { case CMD_LOGIN: { Login* login = (Login*)header; //printf("收到命令 使用者名稱:%s密碼%s\n", login->userName, login->PassWord); LoginResult ret = {}; ret.result = 0; SendData(cSock, &ret); break; } case CMD_LOGOUT: { Logout* loginOut = (Logout*)header; //printf("收到命令: 資料長度:%d 使用者名稱:%s密碼%s\n", loginOut->dataLength, loginOut->userName); LogOutResult retOut = {}; retOut.result = 0; SendData(cSock, &retOut); break; } case CMD_ERROR: { printf("<socket = %d>收到服務端訊息:CMD_ERROR,資料長度:%d\n", _sock, header->dataLength); break; } default: { printf("<socket = %d>收到未定義的訊息,資料長度:%d\n", _sock, header->dataLength); break; } } } // 傳送資料給指定的socket int SendData(SOCKET _cSock, DataHeader* header) { if (isRun() && header) send(_cSock, (const char*)header, header->dataLength, 0); return SOCKET_ERROR; } //群發 void SendDataToAll(DataHeader* header) { if (isRun() && header) { //廣播所有客戶端,新加入一個新的客戶端 for (std::vector<ClientSocket*>::iterator it = _clients.begin(); it != _clients.end(); it++) { SendData((*it)->sockfd(), header); } } } private: }; #endif
MessageHeader.hpp:
#ifndef _MessageHeader_hpp_ #define _MessageHeader_hpp_ enum CMD { CMD_LOGIN, CMD_LOGIN_RESULT, CMD_LOGOUT, CMD_LOGOUT_RESULT, CMD_NEW_USER_JOIN, CMD_ERROR, }; // 訊息頭 struct DataHeader { DataHeader() { dataLength = sizeof(DataHeader); cmd = CMD_ERROR; } short dataLength; short cmd; }; struct Login :public DataHeader { Login() { dataLength = sizeof(Login); cmd = CMD_LOGIN; } char userName[32]; char PassWord[32]; char data[932]; }; struct LoginResult : public DataHeader { LoginResult() { dataLength = sizeof(LoginResult); cmd = CMD_LOGIN_RESULT; result = 0; } char data[992]; int result; }; struct Logout :public DataHeader { Logout() { dataLength = sizeof(Logout); cmd = CMD_LOGOUT; } char userName[32]; }; struct LogOutResult :public DataHeader { LogOutResult() { dataLength = sizeof(LogOutResult); cmd = CMD_LOGOUT_RESULT; result = 0; } int result; }; struct NewUserJoin :public DataHeader { NewUserJoin() { dataLength = sizeof(NewUserJoin); cmd = CMD_NEW_USER_JOIN; sock = 0; } int sock; }; #endif
Client部分:一共3個檔案(Client.cpp,EasyTcpClient.hpp和MessageHeader.hpp)
Client.cpp:
#define _CRT_SECURE_NO_WARNINGS #include <thread> #include "EasyTcpClient.hpp" #pragma comment(lib, "ws2_32.lib") void cmdThread(EasyTcpClient *client) { while (true) { char cmdBuf[256] = {}; scanf("%s", cmdBuf); if (0 == strcmp(cmdBuf, "exit")) { client->Close(); printf("退出cmdThread執行緒\n"); break; } else if (0 == strcmp(cmdBuf, "login")) { Login login; strcpy(login.userName, "lyd"); strcpy(login.PassWord, "lydmm"); client->SendData(&login); } else if (0 == strcmp(cmdBuf, "logout")) { Login logout; strcpy(logout.userName, "lyd"); client->SendData(&logout); } else printf("不支援的命令。 \n"); } } int main() { EasyTcpClient client; client.initSocket(); client.Connect("127.0.0.1", 10000); //std::thread t1(cmdThread, &client); //t1.detach(); // 執行緒分離 while (client.isRun()) { client.OnRun(); // Sleep(1000); Login login; strcpy(login.userName, "lyd"); strcpy(login.PassWord, "lydmm"); client.SendData(&login); } client.Close(); printf("已退出。 \n"); getchar(); return 0; }
EasyTcpClient.hpp:
#pragma once #ifdef _WIN32 #include <WinSock2.h> #include <windows.h> #include "MessageHeader.hpp" #include <stdio.h> #else #include<unistd.h> #include<arpa/inet.h> #include<string.h> #define SOCKET int #define INVALID_SOCKET (SOCKET)(~0) #define SOCKET_ERROR (-1) #include <stdio.h> #endif #define RECV_BUFF_SIZE 10240 class EasyTcpClient { SOCKET _sock; //接受緩衝區 char _szRecv[RECV_BUFF_SIZE]; //第二緩衝區,訊息緩衝區 char _szMsgBuf[RECV_BUFF_SIZE * 10]; //訊息緩衝區的大小 int _lastPos = 0; public: EasyTcpClient() { _sock = INVALID_SOCKET; } virtual ~EasyTcpClient() { Close(); } // 初始化socket void initSocket() { // 啟動Windows socket 2.x環境 #ifdef _WIN32 WORD ver = MAKEWORD(2, 2); WSADATA dat; WSAStartup(ver, &dat); #endif //---------------- // 建立一個socket if (INVALID_SOCKET != _sock) { printf("<sock=%d>關閉之前的連線...\n", _sock); Close(); } _sock = socket(AF_INET, SOCK_STREAM, 0); if (INVALID_SOCKET == _sock) printf("錯誤,建立Socket失敗...\n"); else printf("建立Socket成功...\n"); } // 連線伺服器 int Connect(const char* ip, unsigned short port) { if (INVALID_SOCKET == _sock) { initSocket(); } // 連線伺服器 connect sockaddr_in _sin = {}; _sin.sin_family = AF_INET; _sin.sin_port = htons(port); #ifdef _WIN32 _sin.sin_addr.S_un.S_addr = inet_addr(ip); #else _sin.sin_addr.s_addr = inet_addr(ip); #endif int ret = connect(_sock, (sockaddr*)&_sin, sizeof(sockaddr_in)); if (SOCKET_ERROR == ret) printf("連線伺服器錯誤...\n"); else printf("連線伺服器成功...\n"); return ret; } void Close() { if (_sock != INVALID_SOCKET) { #ifdef _WIN32 closesocket(_sock); WSACleanup(); #else close(_sock); #endif _sock = INVALID_SOCKET; } } // 傳送資料 // 接受資料 // 處理網路訊息 bool OnRun() { if (isRun()) { fd_set fdReads; FD_ZERO(&fdReads); FD_SET(_sock, &fdReads); timeval t = { 1, 0 }; int ret = select(_sock + 1, &fdReads, 0, 0, &t); if (ret < 0) { printf("<socket=%d>select任務結束\n", _sock); return false; } if (FD_ISSET(_sock, &fdReads)) { FD_CLR(_sock, &fdReads); if (-1 == RecvData(_sock)) { printf("<socket=%d>select任務結束2\n", _sock); return false; } } return true; } return false; } bool isRun() { return _sock != INVALID_SOCKET; } // 接受資料,處理粘包,拆分包 int RecvData(SOCKET _cSock) { // 緩衝區 //ZeroMemory(_szRecv, sizeof(_szRecv)); int nLen = recv(_cSock, (char*)&_szRecv, RECV_BUFF_SIZE, 0); memcpy(_szMsgBuf + _lastPos, _szRecv, nLen); _lastPos += nLen; //判斷訊息緩衝區的資料長度大於訊息頭 while (_lastPos >= sizeof(DataHeader)) { DataHeader* header = (DataHeader*)_szMsgBuf; //判斷訊息緩衝區的資料長度大於訊息體 if (_lastPos >= header->dataLength) { //剩餘未處理訊息緩衝區資料的長度 int nSize = _lastPos - header->dataLength; OnNetMsg(header); //將未處理的資料前移 memcpy(_szMsgBuf, _szMsgBuf + header->dataLength, nSize); _lastPos = nSize; } else { //剩餘資料小於一個訊息體的大小 break; } } return 0; } // 響應網路 void OnNetMsg(DataHeader* header) { switch (header->cmd) { case CMD_LOGIN_RESULT: { LoginResult* login = (LoginResult*)header; //printf("收到服務端訊息:CMD_LOGIN_RESULT,資料長度:%d\n", header->dataLength); break; } case CMD_LOGOUT_RESULT: { LogOutResult* logout = (LogOutResult*)header; //printf("收到服務端訊息:CMD_LOGOUT_RESULT,資料長度:%d\n", header->dataLength); break; } case CMD_NEW_USER_JOIN: { NewUserJoin* userJoin = (NewUserJoin*)header; //printf("收到服務端訊息:CMD_NEW_USER_JOIN,資料長度:%d\n", header->dataLength); break; } case CMD_ERROR: { printf("<socket = %d>收到服務端訊息:CMD_ERROR,資料長度:%d\n", _sock, header->dataLength); break; } default: { printf("<socket = %d>收到未定義的訊息,資料長度:%d\n", _sock, header->dataLength); break; } } } // 傳送資料 int SendData(DataHeader* header) { if (isRun() && header) send(_sock, (const char*)header, header->dataLength, 0); return SOCKET_ERROR; } };
MessageHeader.hpp:
#ifndef _MessageHeader_hpp_ #define _MessageHeader_hpp_ enum CMD { CMD_LOGIN, CMD_LOGIN_RESULT, CMD_LOGOUT, CMD_LOGOUT_RESULT, CMD_NEW_USER_JOIN, CMD_ERROR, }; // 訊息頭 struct DataHeader { DataHeader() { dataLength = sizeof(DataHeader); cmd = CMD_ERROR; } short dataLength; short cmd; }; struct Login :public DataHeader { Login() { dataLength = sizeof(Login); cmd = CMD_LOGIN; } char userName[32]; char PassWord[32]; char data[932]; }; struct LoginResult : public DataHeader { LoginResult() { dataLength = sizeof(LoginResult); cmd = CMD_LOGIN_RESULT; result = 0; } char data[992]; int result; }; struct Logout :public DataHeader { Logout() { dataLength = sizeof(Logout); cmd = CMD_LOGOUT; } char userName[32]; }; struct LogOutResult :public DataHeader { LogOutResult() { dataLength = sizeof(LogOutResult); cmd = CMD_LOGOUT_RESULT; result = 0; } int result; }; struct NewUserJoin :public DataHeader { NewUserJoin() { dataLength = sizeof(NewUserJoin); cmd = CMD_NEW_USER_JOIN; sock = 0; } int sock; }; #endif