前言

之前我們學(xué)習(xí)過socket之tcp通信，知道了使用tcp建立連接的一系列操作，并通過write與read函數(shù)能讓客戶端與服務(wù)端進(jìn)行通信，但是tcp是面向字節(jié)流的，有可能我們write時只寫入了部分?jǐn)?shù)據(jù)，此時另一端就來read了，可能會導(dǎo)致讀取的數(shù)據(jù)不完整的問題。這就引入到tcp通信的一個重要操作了——序列化與反序列化！

一、使用結(jié)構(gòu)體進(jìn)行傳輸

在我們之前學(xué)習(xí)的代碼中，一般想傳遞很多數(shù)據(jù)，都會使用到結(jié)構(gòu)體。

比如pthread_create()需要傳遞函數(shù)地址，你如果需要傳遞很多參數(shù)，就得創(chuàng)建一個結(jié)構(gòu)體，將他們組織起來，傳遞這個結(jié)構(gòu)體指針給到pthread_create()，因為他第四個參數(shù)只能接受void*指針。不是可變參數(shù)包。

?如下，制作一個網(wǎng)絡(luò)計算器，我們客戶端創(chuàng)建結(jié)構(gòu)體數(shù)據(jù)并發(fā)送

服務(wù)端對數(shù)據(jù)進(jìn)行接受

這樣服務(wù)器就可以接受到數(shù)據(jù)并做相應(yīng)的處理了，這里我們顯示出來表示接收到了數(shù)據(jù)，處理起來也就很簡單了。

結(jié)構(gòu)體代碼鏈接

二、自定義序列化與反序列化

tcp中的數(shù)據(jù)使用結(jié)構(gòu)體傳輸在網(wǎng)絡(luò)壓力很大，或者傳輸內(nèi)容很多時，就會發(fā)生問題。同樣在不同的編譯環(huán)境下，結(jié)構(gòu)體內(nèi)存對齊也不一致，并且有可能客戶端是使用其他代碼進(jìn)行編寫，C/C++的結(jié)構(gòu)體也不會適合。

• 如果我們將傳輸?shù)臄?shù)據(jù)統(tǒng)一處理成字符串，讓字符串在tcp中進(jìn)行傳輸，同時設(shè)置一些固定的字段，代表當(dāng)前數(shù)據(jù)包的結(jié)束，是不是能解決上述問題呢？
• 這就是數(shù)據(jù)序列化后進(jìn)行傳輸，接收方將數(shù)據(jù)進(jìn)行反序列化后提取再做處理。序列化的本質(zhì)就是對字符串作處理。

一樣的計算器，一樣的需要傳遞兩個整形數(shù)字與一個字符。

• 如果我們將該結(jié)構(gòu)體轉(zhuǎn)為字符串，如?"x oper y\n" ，這樣我們就可以通過讀取空格來表示取到的整形，字符，整形，如果一直沒讀到"\n"證明當(dāng)前數(shù)據(jù)包還沒讀完，讀到"\n"證明取到了一個完整的數(shù)據(jù)包，這樣就可以完美解決問題了。
• 但是我們這個寫法沒有普適性，因為如果后面如果結(jié)構(gòu)體內(nèi)容比較復(fù)雜，帶了一個"\n"，就會導(dǎo)致序列化代碼需要重新設(shè)計。如果在前面添加一個報頭len，代表后面內(nèi)容的長度，這樣無論后面代碼是什么，都可以統(tǒng)一看待處理。
• 于是，我們得到如下的序列化公式。

"len\nx op y"

這里len后面的"\n"代表我們讀取到了len數(shù)據(jù)，后面就是真正要傳輸并處理的數(shù)據(jù)了。

了解了這些，那么我們進(jìn)行網(wǎng)絡(luò)計算器的序列化與反序列化也就不難了，代碼如下

Protocol.hpp? 自定義協(xié)議

#pragma once#include <iostream>
#include <memory>
using namespace std;namespace Protocol
{const string ProtSep = " ";const string LineBreakSep = "\n";// message 現(xiàn)在我要給他頭部添加len\n 尾部添加\n// 如"x op y"-> "len\nx op y\n"   "result code"-> "len\nresult code\n"string Encode(string &message){string len = to_string(message.size());string package = len + LineBreakSep + message + LineBreakSep;return package;}// 從一個完整的package進(jìn)行解析，"len\nx op y\n"->"x op y"bool Decode(string &package, string *message){int left = package.find(LineBreakSep);if (left == string::npos)return false;int len = stoi(package.substr(0, left));int total = left + len + 2 * LineBreakSep.size();if (package.size() < total)return false;// 到這里至少有一個完整的報文 截取出來一個報文寫入message，然后刪除該報文*message = package.substr(left + LineBreakSep.size(), len);package.erase(0, total);return true;}class Request{public:Request() : _data_x(0), _data_y(0), _oper(0){}Request(int x, int y, char op): _data_x(x), _data_y(y), _oper(op){}void Debug(){cout << "_data_x: " << _data_x << endl;cout << "_data_y: " << _data_y << endl;cout << "_oper: " << _oper << endl;}void Inc(){_data_x++;_data_y++;}// 序列化  變成這種字符串"x op y"void Serialize(string *out){*out = to_string(_data_x) + ProtSep + _oper + ProtSep + to_string(_data_y);}// 反序列化  "x op y"字符串分別取出bool Deserialize(string &in){size_t left = in.find(ProtSep);if (left == string::npos)return false;size_t right = in.rfind(ProtSep);if (right == string::npos)return false;_data_x = stoi(in.substr(0, left));_data_y = stoi(in.substr(right + ProtSep.size()));string op = in.substr(left + ProtSep.size(), right - left - ProtSep.size());if (op.size() != 1)return false;_oper = op[0];return true;}int Getx(){return _data_x;}int Gety(){return _data_y;}char Getop(){return _oper;}string To_string(){return to_string(_data_x)+" "+_oper+ " "+to_string(_data_y)+" = ";}private:// x oper y 比如 10 + 20int _data_x;int _data_y;char _oper; // 符號};class Response{public:Response() : _result(0), _code(0){}Response(int result, int code): _result(result), _code(code){}void Serialize(string *out){//_result _code*out = to_string(_result) + ProtSep + to_string(_code);}bool Deserialize(string &in){size_t pos = in.find(ProtSep);if (pos == string::npos)return false;_result = stoi(in.substr(0, pos));_code = stoi(in.substr(pos + ProtSep.size()));return true;}int Getresult(){return _result;}int Getcode(){return _code;}void Setresult(int result){_result = result;}void Setcode(int code){_code = code;}private:int _result; // 運算結(jié)果int _code;   // 運算狀態(tài)};// 工廠模式class Factory{public:shared_ptr<Request> BuildRequest(){return make_shared<Request>();}shared_ptr<Request> BuildRequest(int x, int y, char op){return make_shared<Request>(x, y, op);}shared_ptr<Response> BuildResponse(){return make_shared<Response>();}shared_ptr<Response> BuildResponse(int result, int code){return make_shared<Response>(result, code);}};
}

?Socket.hpp 套接字函數(shù)封裝

#pragma once#include <iostream>
#include <string>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <cstring>
#include <unistd.h>
using namespace std;
namespace Net_Work
{static const int default_backlog = 5;static const int default_sockfd = -1;using namespace std;enum{SocketError = 1,BindError,ListenError,ConnectError,};// 封裝套接字接口基類class Socket{public:// 封裝了socket相關(guān)方法virtual ~Socket() {}virtual void CreateSocket() = 0;virtual void BindSocket(uint16_t port) = 0;virtual void ListenSocket(int backlog) = 0;virtual bool ConnectSocket(string &serverip, uint16_t serverport) = 0;virtual Socket *AcceptSocket(string *peerip, uint16_t *peerport) = 0;virtual int GetSockFd() = 0;virtual void SetSockFd(int sockfd) = 0;virtual void CloseSocket() = 0;virtual bool Recv(string *buff, int size) = 0;virtual void Send(string &send_string) = 0;// 方法的集中在一起使用public:void BuildListenSocket(uint16_t port, int backlog = default_backlog){CreateSocket();BindSocket(port);ListenSocket(backlog);}bool BuildConnectSocket(string &serverip, uint16_t serverport){CreateSocket();return ConnectSocket(serverip, serverport);}void BuildNormalSocket(int sockfd){SetSockFd(sockfd);}};class TcpSocket : public Socket{public:TcpSocket(int sockfd = default_sockfd): _sockfd(sockfd){}~TcpSocket() {}void CreateSocket() override{_sockfd = socket(AF_INET, SOCK_STREAM, 0);if (_sockfd < 0)exit(SocketError);}void BindSocket(uint16_t port) override{int opt = 1;setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt));struct sockaddr_in local;memset(&local, 0, sizeof(local));local.sin_family = AF_INET;local.sin_port = htons(port);local.sin_addr.s_addr = INADDR_ANY;int n = bind(_sockfd, (struct sockaddr *)&local, sizeof(local));if (n < 0)exit(BindError);}void ListenSocket(int backlog) override{int n = listen(_sockfd, backlog);if (n < 0)exit(ListenError);}bool ConnectSocket(string &serverip, uint16_t serverport) override{struct sockaddr_in addr;memset(&addr, 0, sizeof(addr));addr.sin_family = AF_INET;addr.sin_port = htons(serverport);// addr.sin_addr.s_addr = inet_addr(serverip.c_str());inet_pton(AF_INET, serverip.c_str(), &addr.sin_addr);int n = connect(_sockfd, (sockaddr *)&addr, sizeof(addr));if (n == 0)return true;return false;}Socket *AcceptSocket(string *peerip, uint16_t *peerport) override{struct sockaddr_in addr;socklen_t len = sizeof(addr);int newsockfd = accept(_sockfd, (sockaddr *)&addr, &len);if (newsockfd < 0)return nullptr;// *peerip = inet_ntoa(addr.sin_addr);// INET_ADDRSTRLEN 是一個定義在頭文件中的宏，表示 IPv4 地址的最大長度char ip_str[INET_ADDRSTRLEN];inet_ntop(AF_INET, &addr.sin_addr, ip_str, INET_ADDRSTRLEN);*peerip = ip_str;*peerport = ntohs(addr.sin_port);Socket *s = new TcpSocket(newsockfd);return s;}int GetSockFd() override{return _sockfd;}void SetSockFd(int sockfd) override{_sockfd = sockfd;}void CloseSocket() override{if (_sockfd > default_sockfd)close(_sockfd);}bool Recv(string *buff, int size) override{char inbuffer[size];ssize_t n = recv(_sockfd, inbuffer, size - 1, 0);if (n > 0){inbuffer[n] = 0;*buff += inbuffer;return true;}elsereturn false;}void Send(string &send_string) override{send(_sockfd, send_string.c_str(),send_string.size(),0);}private:int _sockfd;string _ip;uint16_t _port;};
}

Calculate.hpp? 計算業(yè)務(wù)封裝

#pragma once#include <iostream>
#include <memory>
#include "Protocol.hpp"enum ErrCode
{Sucess = 0,DivZeroErr,ModZeroErr,UnKnowOper,
};//計算業(yè)務(wù)
class Calculate
{
public:Calculate(){}shared_ptr<Protocol::Response> Cal(shared_ptr<Protocol::Request> req){shared_ptr<Protocol::Response> resp = factory.BuildResponse();resp->Setcode(Sucess);switch (req->Getop()){case '+':resp->Setresult(req->Getx() + req->Gety());break;case '-':resp->Setresult(req->Getx() - req->Gety());break;case '*':resp->Setresult(req->Getx() * req->Gety());break;case '/':{if (req->Gety() == 0)resp->Setcode(DivZeroErr);elseresp->Setresult(req->Getx() / req->Gety());}break;case '%':{if (req->Gety() == 0)resp->Setcode(ModZeroErr);elseresp->Setresult(req->Getx() % req->Gety());}break;default:resp->Setcode(UnKnowOper);break;}return resp;}private:Protocol::Factory factory;
};

?TcpServer.hpp? tcp服務(wù)端封裝

#pragma once#include "Protocol.hpp"
#include "Socket.hpp"
#include <thread>
#include <functional>using func_t = function<string(string &inbufferstream, bool *error_code)>;class TcpServer
{
public:TcpServer(uint16_t port, func_t hander_request): _port(port), _listen_socket(new Net_Work::TcpSocket()), _handler_request(hander_request){_listen_socket->BuildListenSocket(port);}void ThreadRun(Net_Work::Socket *s){// cout<< " in HandlerRequest !"<<endl;string inbufferstream;while (true){bool resultcode = true;// 1.讀取報文if (!s->Recv(&inbufferstream, 1024))break;// 2.報文處理string send_string = _handler_request(inbufferstream, &resultcode);if (resultcode){// 發(fā)送數(shù)據(jù)if (!send_string.empty()){s->Send(send_string);}}else{break;}}s->CloseSocket();delete s;}void Loop(){while (true){string peerip;uint16_t peerport;Net_Work::Socket *newsockfd = _listen_socket->AcceptSocket(&peerip, &peerport);if (newsockfd == nullptr)continue;cout << "獲取一個新鏈接, sockfd: " << newsockfd->GetSockFd() << " ,client info: " << peerip << ":" << peerport << endl;thread t1(&TcpServer::ThreadRun, this, newsockfd);t1.detach();}}~TcpServer(){delete _listen_socket;}private:int _port;Net_Work::Socket *_listen_socket;func_t _handler_request;
};

Main.cc? ?服務(wù)端實現(xiàn)

#include "TcpServer.hpp"
#include "Calculate.hpp"
#include <unistd.h>
#include <memory>string HandlerRequest(string &inbufferstream, bool *error_code)
{Calculate calculate;// 1.創(chuàng)建請求對象unique_ptr<Protocol::Factory> factory(new Protocol::Factory());shared_ptr<Protocol::Request> req = factory->BuildRequest();string total_resp_string;// 2.判斷字節(jié)流是否讀取到了一個完整的報文  把報文讀完string message;while (Protocol::Decode(inbufferstream, &message)){// 3.這里已經(jīng)讀到完整的報文，需要進(jìn)行反序列化  反序列化失敗證明是非法請求if (!req->Deserialize(message)){// 反序列化失敗  不能忍受，因為客戶端的請求是不合法的*error_code = false;return string();}// 4.業(yè)務(wù)處理shared_ptr<Protocol::Response> resp = calculate.Cal(req);// 5.序列化string send_string;resp->Serialize(&send_string);// 6.添加報頭send_string = Protocol::Encode(send_string);total_resp_string+=send_string;}return total_resp_string;
}int main(int argc, char *argv[])
{if (argc != 2){cout << "Usage : " << argv[0] << " port" << endl;return 0;}uint16_t localport = stoi(argv[1]);unique_ptr<TcpServer> tsvr(new TcpServer(localport, HandlerRequest));tsvr->Loop();return 0;
}

TcpClient.cc 服務(wù)端實現(xiàn)?

#include <iostream>
#include <string>
#include <unistd.h>
#include <ctime>
#include <cstdlib>
#include "Socket.hpp"
#include "Protocol.hpp"using namespace std;
int main(int argc, char *argv[])
{if (argc != 3){cout << "Usage : " << argv[0] << " serverip serverport" << endl;return 0;}string serverip = argv[1];uint16_t serverport = stoi(argv[2]);// 創(chuàng)建套接字對象Net_Work::Socket *s = new Net_Work::TcpSocket();if (!s->BuildConnectSocket(serverip, serverport)){cerr << "connect [" << serverip << ":" << serverport << "] failed" << endl;return Net_Work::ConnectError;}cout << "connect [" << serverip << ":" << serverport << "] success" << endl;// 創(chuàng)建工廠unique_ptr<Protocol::Factory> factory(new Protocol::Factory());srand(time(nullptr) ^ getpid());const string opers = "+-*/%^&=";while (true){// 1.構(gòu)建一個請求int x = rand() % 100;usleep(rand() % 5555);int y = rand() % 100;int oper = opers[rand() % opers.size()];shared_ptr<Protocol::Request> req = factory->BuildRequest(x, y, oper);// 2.序列化該請求string requeststr;req->Serialize(&requeststr);// 3.添加報頭信息requeststr = Protocol::Encode(requeststr);// 4.發(fā)送請求s->Send(requeststr);string message;while (true){// 5.讀取響應(yīng)string responsestr;s->Recv(&responsestr, 1024);// 6.解析報文  收到的"len\n result code"  現(xiàn)在開始解析if(!Protocol::Decode(responsestr,&message))continue;// 7.拆分?jǐn)?shù)據(jù) 目前是"result code"  現(xiàn)在開始寫入結(jié)構(gòu)體shared_ptr<Protocol::Response> resp = factory->BuildResponse();resp->Deserialize(message);// 8.resp已經(jīng)得到的數(shù)據(jù)cout<<req->To_string()<<resp->Getresult()<<"\tcode: "<<resp->Getcode()<<endl;break;}sleep(1);}s->CloseSocket();return 0;
}

?Makefile

.PHONY:all
all:tcpserver tcpclienttcpserver:Main.ccg++ -o $@ $^ -std=c++11 -lpthread
tcpclient:TcpClient.ccg++ -o $@ $^ -std=c++11 -lpthread.PHONY:clean
clean:rm -f tcpclient tcpserver

代碼鏈接

運行結(jié)果吐下