一.網(wǎng)絡(luò)層與傳輸層協(xié)議

• 網(wǎng)絡(luò)層與傳輸層內(nèi)置于操作系統(tǒng)的內(nèi)核中,網(wǎng)絡(luò)層一般使用ip協(xié)議,傳輸層常用協(xié)議為Tcp協(xié)議和Udp協(xié)議,Tcp協(xié)議和Udp協(xié)議擁有各自的特點(diǎn)和應(yīng)用場(chǎng)景:
  

sockaddr結(jié)構(gòu)體繼承體系(Linux體系)

• sockaddr_in結(jié)構(gòu)體用于存儲(chǔ)網(wǎng)絡(luò)通信主機(jī)進(jìn)程的ip和端口號(hào)等信息
  

貫穿計(jì)算機(jī)系統(tǒng)的網(wǎng)絡(luò)通信架構(gòu)圖示:

二.實(shí)現(xiàn)并部署多線程并發(fā)Tcp服務(wù)器框架

小項(xiàng)目的完整文件的gittee鏈接

• Tcp服務(wù)器架構(gòu):
  

線程池模塊

#pragma once
#include <iostream>
#include <pthread.h>
#include "log.hpp"
#include <semaphore.h>
#include <vector>
#include <cstdio>template<class T>
class RingQueue{
private:pthread_mutex_t Clock_;pthread_mutex_t Plock_;sem_t Psem_;sem_t Csem_;std::vector<T> Queue_;int Pptr_;int Cptr_;int capacity_;
public:RingQueue(int capacity = 10) : Queue_(capacity),Pptr_(0),Cptr_(0),capacity_(capacity){sem_init(&Psem_,0,capacity);sem_init(&Csem_,0,0);pthread_mutex_init(&Clock_,nullptr);pthread_mutex_init(&Plock_,nullptr);}~RingQueue(){sem_destroy(&Psem_);sem_destroy(&Csem_);pthread_mutex_destroy(&Clock_);pthread_mutex_destroy(&Plock_);}T Pop(){sem_wait(&Csem_);pthread_mutex_lock(&Clock_);T tem = Queue_[Cptr_];Cptr_++;Cptr_ %= capacity_;pthread_mutex_unlock(&Clock_);sem_post(&Psem_);return tem;}void Push(T t){sem_wait(&Psem_);pthread_mutex_lock(&Plock_);Queue_[Pptr_] = t;Pptr_++;Pptr_%= capacity_;pthread_mutex_unlock(&Plock_);sem_post(&Csem_);}
};

#pragma once
#include "sem_cp.cpp"
#include <pthread.h>
#include <iostream>
#include <string>
#include <mutex>
#include "CalTask.cpp"template<class Task>
class Thread_Pool{struct Thread_Data{int Thread_num;pthread_t tid;};
private:RingQueue<Task> Queue_;  //線程安全的環(huán)形隊(duì)列std::vector<Thread_Data> thread_arr; //管理線程的容器static std::mutex lock_;            //單例鎖static Thread_Pool<Task> * ptr_;    //單例指針
private:Thread_Pool(int capacity_Of_queue = 20) : Queue_(capacity_Of_queue){}Thread_Pool(const Thread_Pool<Task>& Tp) = delete;Thread_Pool<Task>& operator=(const Thread_Pool<Task> & Tp) = delete;
public:~Thread_Pool(){}//獲取線程池單例-->注意C++的類模板靜態(tài)成員函數(shù)需要在類體外進(jìn)行定義static Thread_Pool<Task> * Getinstance();//創(chuàng)建多線程void Create_thread(int thread_num = 10){Thread_Data T_data;for(int i = 0 ; i < thread_num ; ++i){//注意線程池對(duì)象的this指針傳遞給線程pthread_create(&T_data.tid,nullptr,Routine,this);T_data.Thread_num = i + 1;thread_arr.push_back(T_data);}}//線程等待void Thread_join(){for(int i = 0 ;i < thread_arr.size() ; ++i){pthread_join(thread_arr[i].tid,nullptr);}}//向線程池中加入任務(wù)void Push(Task T){Queue_.Push(T);}void Push(Task && T){Queue_.Push(std::forward<Task>(T));}
private://線程函數(shù)-->該函數(shù)沒有在類外調(diào)用,所以無(wú)須在類體外定義static void* Routine(void * args){Thread_Pool<Task> * Pool = static_cast<Thread_Pool<Task> *>(args);while(true){std::cout << "Thread prepare to work\n" << std::endl;Task Thread_Task = Pool->Queue_.Pop();//要求Task類重載()-->用于執(zhí)行具體任務(wù)Thread_Task();}return nullptr;}
};//初始化靜態(tài)指針
template<class Task>
Thread_Pool<Task> * Thread_Pool<Task>::ptr_ = nullptr;
template<class Task>
std::mutex Thread_Pool<Task>::lock_;//注意C++的類模板靜態(tài)成員函數(shù)需要在類體外進(jìn)行定義
template<class Task>
Thread_Pool<Task> * Thread_Pool<Task>::Getinstance(){if(ptr_ == nullptr){lock_.lock();if(ptr_ == nullptr){ptr_ = new Thread_Pool<Task>;}lock_.unlock();}return ptr_;
}

序列化反序列化工具模塊

• 序列反序列化是保證通信過程中數(shù)據(jù)完整性的關(guān)鍵步驟,保證數(shù)據(jù)語(yǔ)義完整,結(jié)構(gòu)完整

#pragma once
#include <iostream>
#include <string>// 自定義序列化反序列化協(xié)議
const std::string blank_space_sep = " ";
const std::string protocol_sep = "\n";
//封裝報(bào)文
std::string Encode(std::string &content){//報(bào)文正文字節(jié)數(shù)std::string package = std::to_string(content.size());package += protocol_sep;package += content;    //用分隔符封裝正文package += protocol_sep;return package;
}//解析報(bào)文package-->"正文長(zhǎng)度"\n"正文"\n
bool Decode(std::string &package, std::string& content){size_t pos = package.find(protocol_sep);if(pos == std::string::npos) return false;//解析報(bào)文正文長(zhǎng)度size_t Len = std::atoi(package.substr(0,pos).c_str());//確定報(bào)文是否完整size_t total_Len = pos + Len + 2;if(package.size() != total_Len) return false;//獲取正文內(nèi)容content = package.substr(pos+1,Len);package.erase(0,total_Len);return true;
}//用戶層協(xié)議請(qǐng)求結(jié)構(gòu)體
class Request{
public:int x;int y;char op; 
public:Request(int data1 , int data2 , char op): x(data1),y(data2),op(op){}Request(){}
public://請(qǐng)求結(jié)構(gòu)體 序列化 成報(bào)文正文字符串 "x op y"bool Serialize(std::string& out){std::string content = std::to_string(x);content += blank_space_sep;content += op;content += blank_space_sep;content += std::to_string(y);out = content;return true;// 等價(jià)的jason代碼// Json::Value root;// root["x"] = x;// root["y"] = y;// root["op"] = op;// // Json::FastWriter w;// Json::StyledWriter w;// out = w.write(root);// return true;}//報(bào)文正文字符串 反序列化 成請(qǐng)求結(jié)構(gòu)體// "x op y"bool Deserialize(const std::string &in) {size_t left = in.find(blank_space_sep);if(left == std::string::npos)return false;x = std::stoi(in.substr(0,left).c_str());std::size_t right = in.rfind(blank_space_sep);if (right == std::string::npos)return false;y = std::atoi(in.substr(right + 1).c_str());if(left + 2 != right) return false;op = in[left+1];return true;// 等價(jià)的jason代碼// Json::Value root;// Json::Reader r;// r.parse(in, root);// x = root["x"].asInt();// y = root["y"].asInt();// op = root["op"].asInt();// return true;}void DebugPrint(){std::cout << "新請(qǐng)求構(gòu)建完成:  " << x << op << y << "=?" << std::endl;}
};//用戶層協(xié)議請(qǐng)求回應(yīng)結(jié)構(gòu)體
class Response{
public:int result;int code; 
public:Response(int res , int c): result(res),code(c){}Response(){}
public://請(qǐng)求回應(yīng)結(jié)構(gòu)體 序列化 成報(bào)文正文字符串 "result code"bool Serialize(std::string& out){std::string s = std::to_string(result);s += blank_space_sep;s += std::to_string(code);out = s;return true;// 等價(jià)的jason代碼// Json::Value root;// root["result"] = result;// root["code"] = code;// // Json::FastWriter w;// Json::StyledWriter w;// out = w.write(root);// return true;}//"result code"//報(bào)文正文字符串 反序列化 成請(qǐng)求回應(yīng)結(jié)構(gòu)體bool Deserialize(const std::string &in) {std::size_t pos = in.find(blank_space_sep);if (pos == std::string::npos)return false;if(pos == 0 || pos == in.size() - 1) return false;result = std::stoi(in.substr(0, pos).c_str());code = std::stoi(in.substr(pos+1).c_str());return true;// 等價(jià)的jason代碼// Json::Value root;// Json::Reader r;// r.parse(in, root);// result = root["result"].asInt();// code = root["code"].asInt();// return true;}void DebugPrint(){std::cout << "結(jié)果響應(yīng)完成, result: " << result << ", code: "<< code << std::endl;}
};

通信信道建立模塊

#pragma once
#include <iostream>
#include <string>
#include <sys/types.h>   
#include <sys/socket.h>
#include "log.hpp"
#include <memory.h>
#include <arpa/inet.h>
#include <netinet/in.h>namespace MySocket{//Tcp通訊構(gòu)建器class TcpServer{enum{UsageError = 1,SocketError,BindError,ListenError,};private:int socketfd_ = -1;std :: string ip_;uint16_t port_;int backlog_ = 10;public:TcpServer(const std::string& ip = "172.19.29.44", uint16_t port = 8081) : ip_(ip) , port_(port){}~TcpServer(){if(socketfd_ > 0) close(socketfd_);}public://確定通信協(xié)議,建立文件描述符void BuildSocket(){socketfd_ = socket(AF_INET,SOCK_STREAM,0);if(socketfd_ < 0){lg(Fatal,"socket error,%s\n",strerror(errno));exit(SocketError);}}//文件描述符與服務(wù)器ip : 端口號(hào)綁定void SocketBind(){struct sockaddr_in addr;memset(&addr,0,sizeof(addr));addr.sin_port = htons(port_);addr.sin_family = AF_INET;addr.sin_addr.s_addr = inet_addr(ip_.c_str());if(bind(socketfd_,(const sockaddr*)&addr,sizeof(addr)) < 0){lg(Fatal,"socket bind error,%s\n",strerror(errno));exit(BindError);}lg(Info,"socket bind success\n");}//啟動(dòng)服務(wù)監(jiān)聽,等待客戶端的連接void Socklisten(){if(socketfd_ <= 0){lg(Fatal,"socket error,%s\n",strerror(errno));exit(SocketError);}if(listen(socketfd_,backlog_) < 0){lg(Fatal, "listen error, %s: %d", strerror(errno), errno);exit(ListenError);}}//服務(wù)器接收客戶端的連接-->并創(chuàng)建用于通信的文件描述符-->一個(gè)客戶端連接對(duì)應(yīng)一個(gè)文件描述符int SockAccept(std::string& cilent_ip, uint16_t& cilent_port){struct sockaddr_in client_addr;  // 輸出型參數(shù),用于獲取用戶的ip : 端口號(hào)memset(&client_addr,0,sizeof(client_addr));socklen_t Len = sizeof(client_addr);int newfd = accept(socketfd_,(struct sockaddr*)&client_addr,&Len);if(newfd < 0){lg(Warning, "accept error, %s: %d", strerror(errno), errno);return -1;}//提取客戶端信息-->輸出參數(shù)char ipstr[64];cilent_ip = inet_ntop(AF_INET,&client_addr.sin_addr,ipstr,sizeof(ipstr));cilent_ip = ipstr;cilent_port = ntohs(client_addr.sin_port);return newfd;}public:int Get_Server_fd(){return socketfd_;}void Close_fd(){if(socketfd_ > 0){close(socketfd_);socketfd_ = -1;}}};
};

服務(wù)器主體模塊

#pragma once
#include "ThreadPool.cpp"
#include "TcpServer.cpp"
#include "CalTask.cpp"
#include "log.hpp"
#include <signal.h>//構(gòu)建計(jì)算器服務(wù)器
class CalServer{const int size = 2048;
private:Thread_Pool<CalTask> * Pool_ptr_;MySocket::TcpServer Socket_;int Socket_fd_ = -1;
public:CalServer(const std::string& de_ip = "172.19.29.44",uint16_t de_port = 8081): Socket_(de_ip,de_port){Pool_ptr_ = Thread_Pool<CalTask>::Getinstance();if(Pool_ptr_ == nullptr){lg(Fatal,"Pool_ptr_ is nullptr\n");return;}Pool_ptr_->Create_thread();}~CalServer(){}
public://建立Tcp連接條件bool Init(){Socket_.BuildSocket();Socket_fd_ = Socket_.Get_Server_fd();if(Socket_fd_ < 0){lg(Fatal,"BuildSocket failed\n");return true;}Socket_.SocketBind();Socket_.Socklisten();lg(Info, "init server .... done");return true;}//啟動(dòng)服務(wù)器void Start(){signal(SIGCHLD, SIG_IGN);signal(SIGPIPE, SIG_IGN);char ReadBuffer[size];while(true){//接受用戶請(qǐng)求std::string client_ip;uint16_t client_port;int client_fd = Socket_.SockAccept(client_ip,client_port);if(client_fd < 0){lg(Warning,"SockAccept error\n");continue;}lg(Info, "accept a new link, sockfd: %d, clientip: %s, clientport: %d", client_fd, client_ip.c_str(), client_port);int n = read(client_fd,ReadBuffer,sizeof(ReadBuffer));ReadBuffer[n] = 0;  std::string TaskStr(ReadBuffer);printf("receives mess from client : %s",ReadBuffer);if(n < 0){lg(Warning,"read error\n");break;}CalTask task(client_fd,client_ip,client_port,TaskStr);Pool_ptr_->Push(task);}}
};

任務(wù)回調(diào)模塊(根據(jù)具體應(yīng)用場(chǎng)景可重構(gòu))

#pragma once
#include <string>
#include "ThreadPool.cpp"
#include "Protocol.cpp"enum{Div_Zero = 1,Mod_Zero,Other_Oper
};class CalTask{
private:int socketfd_;                //網(wǎng)絡(luò)通信文件描述符std :: string ip_;            //客戶端ipuint16_t port_;               //客戶端端口號(hào)std::string package_;         //客戶請(qǐng)求字符串
public:CalTask(int socketfd,const std::string& ip , uint16_t & port,std::string & str): socketfd_(socketfd),ip_(ip),port_(port),package_(str){}CalTask(){}//類一定要有默認(rèn)構(gòu)造函數(shù)~CalTask(){}
public://執(zhí)行計(jì)算任務(wù)并將結(jié)果發(fā)送給用戶void operator() (){std::cout << "Task Running ... \n" << std::endl;std::string content;//將用戶發(fā)送的報(bào)文進(jìn)行解包獲取正文bool r = Decode(package_, content);if (!r)return;//將報(bào)文正文進(jìn)行反序列化Request req;r = req.Deserialize(content);if (!r)return ;req.DebugPrint();content = ""; //構(gòu)建計(jì)算結(jié)果                         Response resp = CalculatorHelper(req);resp.DebugPrint();//計(jì)算結(jié)果序列化成字符串resp.Serialize(content);//字符串正文封裝成報(bào)文發(fā)送給用戶std::string ResStr = Encode(content);write(socketfd_,ResStr.c_str(),ResStr.size());if(socketfd_ > 0)close(socketfd_);}private:Response CalculatorHelper(const Request &req){//構(gòu)建請(qǐng)求回應(yīng)結(jié)構(gòu)體Response resp(0, 0);switch (req.op){case '+':resp.result = req.x + req.y;break;case '-':resp.result = req.x - req.y;break;case '*':resp.result = req.x * req.y;break;case '/':{if (req.y == 0)resp.code = Div_Zero;elseresp.result = req.x / req.y;}break;case '%':{if (req.y == 0)resp.code = Mod_Zero;elseresp.result = req.x % req.y;}break;default:resp.code = Other_Oper;break;}return resp;}
};

Tips:DebugC++代碼過程中遇到的問題記錄

• 使用C++類模板時(shí),若在類模板中定義了靜態(tài)成員函數(shù),且該靜態(tài)成員函數(shù)在類外被調(diào)用,則該靜態(tài)成員函數(shù)必須定義在類外,不然鏈接器無(wú)法找到函數(shù)體.
• 注意類模板靜態(tài)成員的聲明格式需要加關(guān)鍵字temlpate<>
• 聲明類模板靜態(tài)成員時(shí)無(wú)需特化模版類型參數(shù)
• 跨主機(jī)并發(fā)通信測(cè)試: