1. 程式人生 > 實用技巧 >伺服器和客戶端簡單的使用select網路模型(三)

伺服器和客戶端簡單的使用select網路模型(三)

在二的基礎上收發資料進行了改進(資料接收增加了二次快取),提供吞吐量。

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