Linux——应用层自定义协议与序列化

目录

一应用层

1再谈 "协议"

2序列化与反序列化 

3理解read,write,recv,send

4Udp vs Tcp

二网络版本计算器

三手写序列和反序列化

四进程间关系与守护进程

1进程组

1.1什么是进程组

1.2组长进程

2会话 

2.1什么是会话

2.2会话下的前后台进程

3作业控制

3.1概念 

 3.2作业号

4守护进程

---

一应用层

我们平时写的一个个代码，满足我们日常需求的网络程序, 其实都是在应用层

1再谈 "协议"

协议是一种 "约定". socket api 的接口, 在读写数据时, 都是按 "字符串" 的方式来发送接收的.

但如果我们要传输一些 "结构化的数据" 怎么办呢?

简单：双方约定好定义相同的结构化数据即可~

但是这会出现问题：如果在Linux平台下代码能正常通信，但是有可能换做其它平台下就通信不了

所以一般进行传输“结构化数据”时，我们用一个专业名词：序列化与反序列化

2序列化与反序列化 

定义：发送数据时将这个结构体按照一个规则转换成字符串, 接收到数据的时候再按照相同的规则把字符串转化回结构体 

 将结构体进行序列化：把信息由多到变成一段字符串进行网络传送给对方；对方收到的必定是一段字符串，这时就要通过反序列化：把信息一变多，方便上层读取信息

这个过程我们通过下面的网络版本计算器代码来演示~

3理解read,write,recv,send

在前面学习文件操作时，我们说过：read write函数在系统内要将数据写入文件不是直接进行写读的：在系统内部存在着文件缓冲区和语言级缓冲区；我们把数据进行write写入时先将数据拷贝到语言级缓冲区中，再把它们拷贝到文件缓冲区中，最后由OS定期将数据拷贝到磁盘中，完成文件操作；文章：Linux——基础IO可跳转进行观看

在这里也是类似的，只不过：这里的fd == 连接 ==两个缓冲区：发送和接收缓冲区 

结论

1.read,write,recv,send本质都是拷贝函数

2.发数据的本质：是从发送方的发送缓冲区把数据通过协议栈和网络拷贝到接收方的解收缓冲区中；由于发送与接收可以做到互不干涉，因此：

3.这也是tcp支持全双工的原因~

4.tcp叫做传输控制协议的原因(OS)

这难免会存在：数据怎么发，发多少，出错了怎么办...问题：这就要通过tcp中定义出来的各种协议来解决；而传输层也是属于OS管理的范畴，所以在用户层面上我们可以不用担心~

---

缓冲区每时每刻都存在着：用户把数据拷贝到发送缓冲区,通过网络拷贝到对方的接收缓冲区中，一增一减，这不就是：

5.生产消费模型

6.而为什么我们在read,wrtie是要进行阻塞？？

a.read阻塞是缓冲区没数据，write阻塞是对方接收缓冲区数据满了

b.但是最根本是：为了维持同步关系！！

4Udp vs Tcp

Udp传输方式：面向数据报；跟我们平时发快递类似：对方收到了几个快递就是发了几个数据报

Tcp传输方式：面向字节流(客户端发的，服务器不一定能全部收到)；这跟我们平时在接自来水类似：接水的方式你可以选择用盆接，用水桶接，用手接...接收数据的多少不确定；

这就会带出一个问题：客户端怎么保证我收到的是一个完整的请求？？

直接说答案：分割报文；这部分通过代码来体现~~

二网络版本计算器

创建Tcp都是老套路了：socket,bind,listen,accpent；这里决定使用模板方法来设计

//Socket.hpp
#pragma once
#include <iostream>
#include <memory>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>#include "InetAddr.hpp"
#include "Log.hpp"class Socket;//声明
using SockPtr=std::shared_ptr<Socket>;enum
{SOCK_ERROR = 1,BIND_ERROR,LISTEN_ERROR,
};const int gbacklog = 8;//模板方法模式
class Socket
{
public:virtual void CreateSocket() = 0;virtual void InitSocket(uint16_t port,int backlog=gbacklog) = 0;virtual SockPtr AcceptSocket(InetAddr *addr) = 0;                         // 对象/变量virtual bool ConnectSocket(uint16_t port, const std::string &ip) = 0; // clinet连接成功与失败virtual int Sockfd() = 0;virtual void Close() = 0;virtual ssize_t Recv(std::string *out) = 0;virtual ssize_t Send(const std::string &in) = 0;public:void Tcp_ServerSocket(uint16_t port){CreateSocket();InitSocket(port);}bool Tcp_ClientSocket(uint16_t port, const std::string &ip){CreateSocket();return ConnectSocket(port, ip);}};class TcpSocket : public Socket
{
public:TcpSocket(int sockfd): _sockfd(sockfd){}TcpSocket(){}~TcpSocket(){}virtual void CreateSocket() override{_sockfd = ::socket(AF_INET, SOCK_STREAM, 0);if (_sockfd < 0){LOG(FATAL,"socket fail\n");exit(SOCK_ERROR);}LOG(INFO,"socket sucess sockfd: %d\n",_sockfd);}virtual void InitSocket(uint16_t port,int backlog) override{struct sockaddr_in perr;memset(&perr, 0, sizeof(perr));perr.sin_family = AF_INET;perr.sin_port = htons(port);perr.sin_addr.s_addr = INADDR_ANY;if (::bind(_sockfd, (struct sockaddr *)&perr, sizeof(perr)) < 0){LOG(FATAL,"bind fail\n"); exit(BIND_ERROR);}LOG(INFO,"bind sucess\n");if (::listen(_sockfd, backlog) < 0){LOG(ERROR, "listen fail\n");exit(LISTEN_ERROR);}LOG(INFO,"listen sucess\n");}virtual SockPtr AcceptSocket(InetAddr *addr) override // 外层要获取客户端信息{struct sockaddr_in client;socklen_t len = sizeof(client);int sockfd = ::accept(_sockfd, (struct sockaddr *)&client, &len);if (sockfd < 0){LOG(ERROR,"accept fail\n");return nullptr;}*addr = client;LOG(INFO, "get a new link %s sockfd: %d\n",addr->User().c_str(),sockfd);return std::make_shared<TcpSocket>(sockfd);//c++14}virtual bool ConnectSocket(uint16_t port, const std::string &ip) override{struct sockaddr_in server;memset(&server, 0, sizeof(server));server.sin_family = AF_INET;server.sin_port = htons(port);inet_pton(AF_INET, ip.c_str(), &server.sin_addr);int n = connect(_sockfd, (struct sockaddr *)&server, sizeof(server));if (n < 0){return false;}return true;}virtual int Sockfd() override{return _sockfd;}virtual void Close() override{if (_sockfd > 0){::close(_sockfd);}}virtual ssize_t Recv(std::string *out) override{char buffer[4096];ssize_t n = ::recv(_sockfd, buffer, sizeof(buffer) - 1, 0);if (n > 0){buffer[n] = 0;*out += buffer; // 不断获取server的信息(这里不能保证server每次发的报文是完整的)}return n;}virtual ssize_t Send(const std::string &in) override{return ::send(_sockfd, in.c_str(), in.size(), 0);}private:int _sockfd; // 两个角色
};

(会话层)设计Tcpserver：服务器的启动与要执行的事务(这里不进行服务器的recv,send)(回调出去,由上层决定) 

//TcpServer.hpp
#pragma once
#include <pthread.h>
#include <functional>#include "Socket.hpp"static const int gport = 8888;using service_io_t = std::function<void(SockPtr, InetAddr)>;class TcpServer
{
public:TcpServer(service_io_t server, uint16_t port = gport): _server(server), _port(port), _listensockfd(std::make_shared<TcpSocket>()), _isrunning(false){_listensockfd->Tcp_ServerSocket(_port); // socket bind listen}void Start(){_isrunning = true;while (_isrunning){InetAddr client;SockPtr newsock = _listensockfd->AcceptSocket(&client);if (newsock == nullptr)continue; // 断开连接// 进行服务// version 2 -- 多线程 -- 不能关fd -- 共享pthread_t pid;PthreadDate *date = new PthreadDate(newsock, this, client);pthread_create(&pid, nullptr, Excute, date);}_isrunning = false;}struct PthreadDate{SockPtr _sockfd;TcpServer *_self;InetAddr _addr;PthreadDate(SockPtr sockfd, TcpServer *self, const InetAddr& addr): _sockfd(sockfd),_self(self),_addr(addr){}};static void *Excute(void *args){pthread_detach(pthread_self());PthreadDate *date = static_cast<PthreadDate *>(args);date->_self->_server(date->_sockfd, date->_addr); // 进行回调date->_sockfd->Close(); // 关闭 sockfddelete date;return nullptr;}private:service_io_t _server;uint16_t _port;SockPtr _listensockfd;bool _isrunning;
};

(表示层) 设计IoServer和protocol：这里负责处理服务器的收请求，处理，发送

(请求不一定是完整的,这要自己设计报文与处理报文的逻辑，如：怎么保证发的报文是完整的，解包与封包...)

//IoServer.hpp
#pragma once
#include "Socket.hpp"
#include "Protocol.hpp"using protocol_t = std::function<std::shared_ptr<rep>(std::shared_ptr<req>)>;class IoServer
{
public:IoServer(protocol_t process): _process(process){}~IoServer(){}void server(SockPtr sockfd, InetAddr addr){std::string jsonstr; // 持续进行读取与解析while (true){// 读取ssize_t n = sockfd->Recv(&jsonstr);if (n <= 0){LOG(ERROR,"recv error\n");break;}LOG(INFO,"client request: %s",jsonstr.c_str());// 报文解析 -- 不能保证读到完整的报文std::string message = ::Decode(jsonstr);if (message == "")continue;// 反序列化 -- 能保证读到完整报文std::shared_ptr<req> q=Factory::BuidRequest();q->Deserialize(message);// 处理事务auto p = _process(q);// 序列化处理std::string result;p->Serialize(&result);// 添加报头result = Encode(result);// 发送sockfd->Send(result);}LOG(INFO,"%s quit\n",addr.User().c_str());sockfd->Close();}private:protocol_t _process;
};//protocol.hpp
#pragma once
#include <jsoncpp/json/json.h>// "len\r\n{json}\r\n" -- 自己设计出完整报文 len json的长度
// "\r\n" 第一个:区分len 和json边界 第二个:观察现象方便
const std::string sym = "\r\n";// jsonstr变成完整报文
std::string Encode(const std::string &jsonstr)
{int len = jsonstr.size();return std::to_string(len) + sym + jsonstr + sym;
}// 把json提取出来
// "len\r"
// "len\r\n{json}\r"
// "len\r\n{json}\r\n"
// "len\r\n{json}\r\n""len\r\n{json}\r\n""len\r\n{j"
std::string Decode(std::string &nameplate)
{size_t pos = nameplate.find(sym);if (pos == std::string::npos){return "";}std::string lenstr = nameplate.substr(0, pos);int len = std::stoi(lenstr);// 计算出完整报文长度int TotalLen = lenstr.size() + sym.size() + len + sym.size();if (nameplate.size() < TotalLen){return "";}// 读报文std::string jsonstr = nameplate.substr(pos + sym.size(), len);// 删报文nameplate.erase(0, TotalLen);return jsonstr;
}class req
{
public:req(){}req(int x, int y, std::string &sym): _x(x), _y(y), _sym(sym){}// 结构化->字符串void Serialize(std::string *out){// 1.自己做 -> "_x _sym _y"// 2.使用现成库:jsoncppJson::Value root;root["x"] = _x;root["y"] = _y;root["sym"] = _sym;Json::FastWriter writer;*out = writer.write(root);}// 字符串->结构化bool Deserialize(std::string &in){Json::Value root;Json::Reader reader;bool res = reader.parse(in, root);if (!res)return false;_x = root["x"].asInt();_y = root["y"].asInt();_sym = root["sym"].asString();return true;}void SetValue(int x, int y, char sym){_x = x;_y = y;_sym = sym;}~req() {}int _x;int _y;std::string _sym; //-x +-*/ _y
};class rep
{
public:rep(): _result(0), _code(0){}void Serialize(std::string *out){Json::Value root;root["result"] = _result;root["code"] = _code;Json::FastWriter writer;*out = writer.write(root);}bool Deserialize(std::string &in){Json::Value root;Json::Reader reader;bool res = reader.parse(in, root);if (!res)return false;_result = root["result"].asInt();_code = root["code"].asInt();return true;}void PrintResult(){std::cout << "result: " << _result << ", code: " << _code << std::endl;}~rep() {}int _result;int _code; // 0-> sucess 1->div zero 2->mod zero 3->failstd::string _des;
};//工厂模式
class Factory
{
public:static std::shared_ptr<req> BuidRequest(){return std::make_shared<req>();}static std::shared_ptr<rep> BuidReponse(){return std::make_shared<rep>();}
};

(应用层)设计Netcal:将client的请求进行处理(进行计算) 

//Netcal.hpp
#pragma once
#include "Protocol.hpp"// req -> rep
class Cal
{
public:Cal() {}~Cal() {}std::shared_ptr<rep> Count(std::shared_ptr<req> q){auto p = Factory::BuidReponse();const char *a = q->_sym.c_str();switch (*a){case '+':p->_result = q->_x + q->_y;break;case '-':p->_result = q->_x - q->_y;break;case '*':p->_result = q->_x * q->_y;break;case '/':{if (q->_y == 0){p->_result = -1;p->_code = 1;p->_des = "div zero";}else{p->_result = q->_x / q->_y;}break;}case '%':if (q->_y == 0){p->_result = -1;p->_code = 2;p->_des = "mod zero";}else{p->_result = q->_x % q->_y;}break;default:p->_result = -1;p->_code = 3;p->_des = "illegal operation";break;}return p;}
};

将以上的所有逻辑进行整合成ServerMain,cc和ClientMain.cc 

//ServerMain.cc//#define _GLIBCXX_USE_CXX11_ABI 0#include "TcpServer.hpp"
#include"IoServer.hpp"
#include"Netcal.hpp"
int main(int args, char *argv[])
{if (args != 2){std::cerr << "./Server Localport" << std::endl;exit(0);}uint16_t port = std::stoi(argv[1]);Cal cal;IoServer ir(std::bind(&Cal::Count,&cal,std::placeholders::_1));std::unique_ptr<TcpServer> svr=std::make_unique<TcpServer>(std::bind(&IoServer::server, &ir,std::placeholders::_1, std::placeholders::_2),port);svr->Start();return 0;
}//ClientMain.cc
//#define _GLIBCXX_USE_CXX11_ABI 0#include "Socket.hpp"
#include "Protocol.hpp"int main(int argc, char *argv[])
{if (argc != 3){std::cerr << "Usage: " << argv[0] << " server-ip server-port" << std::endl;exit(0);}std::string serverip = argv[1];uint16_t serverport = std::stoi(argv[2]);SockPtr st=std::make_shared<TcpSocket>();if (!st->Tcp_ClientSocket(serverport, serverip)){std::cerr << "connect error" << std::endl;exit(1);}srand(time(nullptr) ^ getpid());const std::string opers = "+-*/%&^!";std::string packagestreamqueue;while (true){// 构建数据int x = rand() % 10;usleep(x * 1000);int y = rand() % 10;usleep(x * y * 100);char oper = opers[y % opers.size()];// 构建请求std::shared_ptr<req> q=Factory::BuidRequest();q->SetValue(x, y, oper);// 1. 序列化std::string reqstr;q->Serialize(&reqstr);// 2. 添加长度报头字段reqstr = Encode(reqstr);// 3. 发送数据st->Send(reqstr);while (true){// 4. 读取server发送过来的字符串ssize_t n = st->Recv(&packagestreamqueue);if (n <= 0){break;}// 我们能保证我们读到的是一个完整的报文吗？不能！// 5. 报文解析std::string package = Decode(packagestreamqueue);if (package.empty())continue;// 6. 反序列化std::shared_ptr<rep> p=Factory::BuidReponse();p->Deserialize(package);// 7. 打印结果p->PrintResult();break;}sleep(1);}return 0;
}

现象

三手写序列和反序列化

将json的工作换成我们来做：

#pragma once
#include <iostream>
#include <memory>
#include <jsoncpp/json/json.h>// #define SELF 1// "len\r\n{json}\r\n" -- 自己设计出完整报文 len json的长度
// "\r\n" 第一个:区分len 和json边界 第二个:观察现象方便
const std::string sym = "\r\n";
const std::string space = " ";// jsonstr变成完整报文
std::string Encode(const std::string &jsonstr)
{int len = jsonstr.size();return std::to_string(len) + sym + jsonstr + sym;
}// 把json提取出来
// "len\r"
// "len\r\n{json}\r"
// "len\r\n{json}\r\n"
// "len\r\n{json}\r\n""len\r\n{json}\r\n""len\r\n{j"
std::string Decode(std::string &nameplate)
{size_t pos = nameplate.find(sym);if (pos == std::string::npos){return "";}std::string lenstr = nameplate.substr(0, pos);int len = std::stoi(lenstr);// 计算出完整报文长度int TotalLen = lenstr.size() + sym.size() + len + sym.size();if (nameplate.size() < TotalLen){return "";}// 读报文std::string jsonstr = nameplate.substr(pos + sym.size(), len);// 删报文nameplate.erase(0, TotalLen);return jsonstr;
}class req
{
public:req(){}req(int x, int y, std::string &sym): _x(x), _y(y), _sym(sym){}// 结构化->字符串void Serialize(std::string *out){
#ifdef SELF// 1.自己做 -> "_x _sym _y"// 2.使用现成库:jsoncppJson::Value root;root["x"] = _x;root["y"] = _y;root["sym"] = _sym;Json::FastWriter writer;*out = writer.write(root);
#else//"len\r\n {_x _sym _y} \r\n"std::string x = std::to_string(_x);std::string y = std::to_string(_y);*out = x + space + _sym + space + y;#endif}// 字符串->结构化bool Deserialize(std::string &in){
#ifdef SELFJson::Value root;Json::Reader reader;bool res = reader.parse(in, root);if (!res)return false;_x = root["x"].asInt();_y = root["y"].asInt();_sym = root["sym"].asString();return true;
#else//"len\r\n {_x _sym _y} \r\n"auto left_space = in.find(space);if (left_space == std::string::npos)return false;auto right_space = in.rfind(space);if (right_space == std::string::npos)return false;if (left_space + space.size() + 1 != right_space)return false;std::string x = in.substr(0, left_space);if (x.empty())return false;std::string sym = in.substr(left_space + space.size(), right_space);if (sym.empty())return false;std::string y = in.substr(right_space + space.size());if (y.empty())return false;_x = std::stoi(x.c_str());_sym = sym;_y = std::stoi(y.c_str());return true;#endif}void SetValue(int x, int y, char sym){_x = x;_y = y;_sym = sym;}~req() {}int _x;int _y;std::string _sym; //-x +-*/ _y
};class rep
{
public:rep(): _result(0), _code(0), _des("sucess"){}void Serialize(std::string *out){
#ifdef SELFJson::Value root;root["result"] = _result;root["code"] = _code;root["des"] = _des;Json::FastWriter writer;*out = writer.write(root);
#elsestd::string result = std::to_string(_result);std::string code = std::to_string(_code);*out = result + space + code + space + _des;
#endif}bool Deserialize(std::string &in){
#ifdef SELFJson::Value root;Json::Reader reader;bool res = reader.parse(in, root);if (!res)return false;_result = root["result"].asInt();_code = root["code"].asInt();_des = root["des"].asString();return true;
#elseauto left_space = in.find(space);if (left_space == std::string::npos)return false;auto right_space = in.rfind(space);if (right_space == std::string::npos)return false;if (left_space + space.size() + 1 != right_space)return false;std::string result = in.substr(0, left_space);if (result.empty())return false;std::string code = in.substr(left_space + space.size(), right_space);if (code.empty())return false;std::string des = in.substr(right_space + space.size());if (des.empty())return false;_result = std::stoi(result.c_str());_code = std::stoi(code.c_str());_des = des;return true;
#endif}~rep() {}int _result;int _code; // 0-> sucess 1->div zero 2->mod zero 3->failstd::string _des;
};// 工厂模式
class Factory
{
public:static std::shared_ptr<req> BuidRequest(){return std::make_shared<req>();}static std::shared_ptr<rep> BuidReponse(){return std::make_shared<rep>();}
};

四进程间关系与守护进程

1进程组

1.1什么是进程组

进程组是一个或者多个进程的集合， 一个进程组可以包含多个进程

每一个进程组也有一个唯一的进程组 ID(PGID)， 并且这个 PGID 类似于进程 ID

1.2组长进程

每一个进程组都有一个组长进程：组长进程的 ID 等于其进程 ID。(程序启动的第一个进程)

查看进程有两种方式：

1查全部数据 

 2查指定数据

# -e 选项表示 every 的意思， 表示输出每一个进程信息
# -o 选项以逗号操作符（,） 作为定界符, 可以指定要输出的列 

主要某个进程组中有一个进程存在则该进程组就存在：这与其组长进程是否已经终止无关！

2会话 

2.1什么是会话

会话看成是一个或多个进程组的集合：一个会话可以包含多个进程组；会话有自己的SID

2.2会话下的前后台进程

有代码和指令来演示现象： 

#include <iostream>
#include <unistd.h>int main()
{pid_t n = fork();if (n == 0){while (true){std::cout << "I am a child process: pid: " << getpid() << std::endl;sleep(1);}}sleep(2);std::cout << "I am a father process pid: " << getpid() << std::endl;sleep(100);return 0;
}

我们发现前台与后台的会话id是一致的：说明前台与后台是在同一会话下进行的；

那么：前台与后台我们要怎么去理解？这个过程中bash去哪了？

同一个会话中可以同时存在多个进程组；会话中会存在着一个bash进程(启动多个终端验证)

但在任意时刻，只允许一个前台进程(组)，后台进程(组)可以同时存在多个

---

刚开始启动终端时，会先创建出终端文件(/dev/pts/)和bash共同组成会话，进而创建出bash进程：此时bash进程默认是前台进程；但我们执行程序代码时，bash会从前台转为后台，此时所以指令都失效了(bash在后台读不到终端数据)；程序结束时，bash默认又转回前台了 

完整关系如下 

3作业控制

3.1概念 

作业是针对用户来讲，用户完成某项任务而启动的进程：一个作业既可以只包含一个进程，也可以包含多个进程， 进程之间互相协作完成任务，通常是一个进程管道。

Shell 分前后台来控制的不是进程而是作业或者进程组。一个前台作业可以由多个进程组成： 一个后台作业也可以由多个进程组成， Shell 可以同时运⾏一个前台作业和任意多个后台作业， 这称为作业控制。

 3.2作业号

后台进程在执行完后会返回一个作业号以及一个进程号（PID） (指令最后加&就是后台进程)

指令使用： 

 

4守护进程

守护进程：进程所在的会话中进行关闭，自己不受影响

实现：进程自己创建会话与当前会话形成并列关系

//Daemon.cc
#pragma once
#include <iostream>
#include <signal.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>const std::string path = "/";
const std::string devpath = "/dev/null";void Daemon(bool ischdir, bool isclose)
{// 信号忽略signal(SIGCHLD, SIG_IGN);signal(SIGPIPE, SIG_IGN);// 子进程创建会话if (fork() > 0)exit(0);setsid();if (ischdir){chdir(path.c_str());}if (isclose){::close(0);::close(1);::close(2);}else{int fd = open(devpath.c_str(), O_RDWR);if (fd > 0){dup2(fd, 0);dup2(fd, 1);dup2(fd, 2);}::close(fd);}
}
//test.cc
#include "Daemon.cc"
int main()
{Daemon(false,false);   sleep(100);//模拟守护进程return 0;
}

守护进程可以用在前面我们自己写的服务器中：让服务器在后台运行不受会话影响！！ 

#define _GLIBCXX_USE_CXX11_ABI 0#include "TcpServer.hpp"
#include"IoServer.hpp"
#include"Netcal.hpp"#include "Daemon.cc"int main(int args, char *argv[])
{if (args != 2){std::cerr << "./Server Localport" << std::endl;exit(0);}SleftFile();//将信息打印在log.txt中Daemon(false,false);uint16_t port = std::stoi(argv[1]);Cal cal;IoServer ir(std::bind(&Cal::Count,&cal,std::placeholders::_1));std::unique_ptr<TcpServer> svr=std::make_unique<TcpServer>(std::bind(&IoServer::server, &ir,std::placeholders::_1, std::placeholders::_2),port);svr->Start();return 0;
}

以上便是全部内容，有错误欢迎在评论区指出，感谢观看！