Linux序列与反序列化、Json的使用、对TCP面向字节流的理解与应用

序列与反序列化应用场景套接字相关接口在读写数据时都是按字符串的方式来发送接收的如果我们要传输一些结构化的数据该怎么办呢?这时就需要采用序列化和反序列化。序列与反序列化的定义一端发送数据时将数据结构体按照定义好的规则转换成字符串,另一端收到数据时再按照相同的规则把字符串转回结构体这就是序列化和反序列化。应用实例现在要实现网络版计算器根据序列化和反序列化的定义客户端会把需求拼接成一形如11的字符串服务端收到该字符串后根据分隔符拆分字符串反序列化为需求结构体从而实际得到两个操作数和操作符进行计算再将计算结果序列化成形如“10 0”的字符串结果和状态码再发回给客户端客户端再对收到的字符串反序列化得到实际结果和状态码完成一次闭环。实现代码如下#pragma once #include string #include string.h // 分隔符 const std::string content_sep ; const std::string messagesep \n; class request_protocol { public: request_protocol() { // 反序列化调用需要用到的默认构造 } request_protocol(int x, int y, char op) : x_(x), y_(y), op_(op) { // 序列化的构造 } bool serialization(std::string out) { // x op y 序列化 std::string res; res std::to_string(x_); res content_sep; res op_; res content_sep; res std::to_string(y_); out res; return true; } bool deserialization(std::string question) { // 1 1 反序列化 size_t head question.find(content_sep); if (head std::string::npos) { return false; } x_ std::stoi(question.substr(0, head)); size_t tail question.rfind(content_sep); if (tail std::string::npos || tail head) { return false; } y_ std::stoi(question.substr(tail)); // 计算总数合理性,两个操作数的长度不定只能判断空格数和操作符的总数 if (head 2 ! tail) { return false; } op_ question[head 1]; return true; } void Print() { std::cout x: x_ y: y_ op: op_ std::endl; } public: int x_; int y_; char op_; }; class response_protocol { public: response_protocol() { // 反序列化调用需要的默认构造 } response_protocol(int result, int code 0) : result_(result), code_(code) { // 序列化的构造 } bool serialization(std::string out) { // result code 序列化 std::string res; res std::to_string(result_); res content_sep; res std::to_string(code_); out res; return true; } bool deserialization(std::string question) { // 100 0 反序列化 size_t head question.find(content_sep); if (head std::string::npos) { return false; } result_ std::stoi(question.substr(0, head)); size_t tail question.rfind(content_sep); if (tail std::string::npos || tail ! head) { return false; } code_ std::stoi(question.substr(tail)); return true; } void Print() { std::cout result: result_ code: code_ std::endl; } public: int result_; int code_ 0; };不难看出主要是字符串的处理比较繁琐同时通过判断分隔符的存在与否、是否合理来保障字符串是符合预定格式的。Json实际上已有成熟的序列化和反序列化方法供我们使用如Json库上述实现仅做演示用下面为Json版本的请求类和响应类#include json/json.h // 确保包含 jsoncpp 头文件 #include iostream #include string class request_protocol { public: request_protocol() { // 反序列化的默认构造 } request_protocol(int x, int y, char op) : x_(x), y_(y), op_(op) { // 序列化的构造 } bool serialization(std::string out) { // JSON 序列化 Json::Value root; root[x] x_; root[y] y_; root[op] std::string(1, op_); Json::StyledWriter w; out w.write(root); // 返回的是 string 类型的字符串 return true; } bool deserialization(std::string question) { // JSON 反序列化 Json::Value root; Json::Reader r; // 从报文中读取相应的字段 if (!r.parse(question, root)) { return false; // 解析失败 } x_ root[x].asInt(); y_ root[y].asInt(); // 注意这里假设 op 存在且非空实际生产环境建议加判断 std::string op_str root[op].asString(); if (!op_str.empty()) { op_ op_str[0]; } else { return false; } return true; } void Print() { std::cout x: x_ y: y_ op: op_ std::endl; } public: int x_; int y_; char op_; }; class response_protocol { public: response_protocol() { // 反序列化的默认构造 } response_protocol(int result, int code 0) : result_(result), code_(code) { // 序列化的构造 } bool serialization(std::string out) { // JSON 序列化 Json::Value root; root[result] result_; root[code] code_; Json::StyledWriter w; out w.write(root); // 返回的是 string 类型的字符串 return true; } bool deserialization(std::string question) { // JSON 反序列化 Json::Value root; Json::Reader r; if (!r.parse(question, root)) { return false; // 解析失败 } result_ root[result].asInt(); code_ root[code].asInt(); return true; } void Print() { std::cout result: result_ code: code_ std::endl; } public: int result_; int code_ 0; };与TCP的联系如何保证读取到完整报文?TCP具有可靠传输的性质诸如只发送了部分报文这类情况需得到妥善处理。从数据传输的过程看来TCP通过调用write和read实现网络数据收发而write和read本质是拷贝函数只负责把数据拷贝到缓冲区至于对方是否接收完全上层是无法确定的。既然数据传输过程无法控制应用层就需要自己进行协议定制也就是定义好一种服务端和客户端都约定好的规则对即将发送和收到的数据进行检查和处理。前面的序列化与反序列化的实现代码只能保证数据是符合特定的格式如计算式具有两个分隔符而数据量具体大小却无法保证能否在此过程中加入相关保障呢用下面两个函数即可实现// 发送前给报文添加总数 bool addsize(std::string message) { std::string temp; size_t size message.size(); temp std::to_string(size); temp messagesep; temp message; temp messagesep; message temp; //??安全吗 return true; } // 检查收到的报文是否完整,还要自检查,另外把缓冲区中对应的字符删掉 bool checkmessage(std::string message, std::string content) { size_t head message.find(messagesep); if (head std::string::npos) { return false; } size_t size std::stoi(message.substr(0, head)); size_t totalsize size head 2; std::string pure_message message.substr(head 1, size); // 只能信任报文中的数字是绝对正确的对比通过read接收到的实际字符,而且不能保证是完全对的 if (message.size() totalsize) { return false; } content pure_message; // 把原缓冲区被使用的报文移除 message.erase(0, totalsize); return true; }第一个函数的意思是计算序列化字符串大小size作为报头添加到字符串头部使用\n作为size与原报文的分隔符以及两个报文之间的分隔符第二个函数则用于根据\n的位置将接收到的加上size后的字符串拆分成size和报文如果报文的大小并不等于size说明报文是不完整的。(“报文不完整”这一事实是”read没得到预期报文“这一事实决定的作为接收方只能相信发送方发来的的报文是完整的因此size是绝对正确的)引入这两个函数后发送方和接收方还需另外遵守这样的规则发送前调用addsize为序列化字符串添加大小接收后调用checkmessage对报文完整性进行检查并得到原始报文。checkmessage函数中erase的作用是什么我们知道TCP是面向字节流的发送方可能通过多次发送才能把一个报文完整发给接收方所以接收方的缓冲区里可能存着半个、一个、一个半或两个报文每次接收方把缓冲区传给checkmessage后若检测到一条完整的报文则将其拆分并删掉缓冲区中原来的报文从而保证下一次接收消息时不会读到先前的报文残留。网络版计算器的实现结合前面学习的TCP和序列化反序列化相关知识下面实现网络版计算器。实现思路用户利用随机数生成计算式发送给服务端服务端处理后将结果发回给客户端关键在于即便客户端一次发送了多个式子服务端也能一一处理相应地客户端也要具备读取服务端一次发送多个结果的能力。设计上服务端只负责收发消息协议处理和具体业务逻辑在计算器类中提供方法。代码示例封装套接字供客户端和服务端使用#pragma once #include sys/socket.h #include netinet/in.h #include arpa/inet.h #include sys/types.h #include string #include log.hpp const int backlog 10; class Socket { public: Socket() { listenfd_ socket(AF_INET, SOCK_STREAM, 0); if(listenfd_-1){ LOG(FATAL, create socket fail,errno:%d,%s, errno, strerror(errno)); exit(0); } else{ LOG(INFO, create socket sucess); } } Socket(std::string port) { port_ static_castuint16_t(std::stoi(port)); listenfd_ socket(AF_INET, SOCK_STREAM, 0); if(listenfd_-1){ LOG(FATAL, create socket fail,errno:%d,%s, errno, strerror(errno)); exit(0); } else{ LOG(INFO, create socket sucess); } } void Bind() { struct sockaddr_in local; bzero(local, sizeof(local)); local.sin_family AF_INET; local.sin_port htons(port_); local.sin_addr.s_addr INADDR_ANY; socklen_t len sizeof(local); if (bind(listenfd_, (sockaddr *)local, len)-1) { LOG(FATAL, bind fail,errno:%d,%s, errno, strerror(errno)); exit(0); } else{ LOG(INFO, bind sucess); } } void Listen() { if (listen(listenfd_, backlog)-1) { LOG(FATAL, listen fail,errno:%d,%s, errno, strerror(errno)); exit(0); } else{ LOG(INFO, listen sucess); } } int Accept(std::string *remote_ip, std::string *remote_port) { struct sockaddr_in remote; bzero(remote, sizeof(remote)); socklen_t len sizeof(remote); int sockfd accept(listenfd_, (sockaddr *)remote, len); if (sockfd -1) { LOG(FATAL, accept fail,errno:%d,%s, errno, strerror(errno)); exit(0); } else{ LOG(INFO, accept sucess); } char buf[INET_ADDRSTRLEN]; const char *ip_str inet_ntop(AF_INET, remote.sin_addr, buf, sizeof(buf)); *remote_ip ip_str; *remote_port ntohs(remote.sin_port); return sockfd; } void Connect(const std::string remote_ip,const std::string remote_port) { struct sockaddr_in remote; bzero(remote, sizeof(remote)); socklen_t len sizeof(remote); remote.sin_family AF_INET; remote.sin_addr.s_addr inet_addr(remote_ip.c_str()); remote.sin_port htons(static_castuint16_t(std::stoi(remote_port.c_str()))); if(connect(listenfd_, (sockaddr *)remote, len)-1){ LOG(FATAL, connect fail,errno:%d,%s, errno, strerror(errno)); exit(0); } else{ LOG(INFO, connect sucess); } } int getid(){ return listenfd_; } void Close(){ close(listenfd_); } ~Socket() { close(listenfd_); } private: int listenfd_; uint16_t port_; };服务端主函数server.cc#includeTcp_Server.hpp #include Calculator_AND_Protocol.hpp #includeiostream #includeunistd.h int main(int argc, char *argv[]){ // 启动服务端时需要指定一个端口 if (argc ! 2) { std::cout please enter server_port std::endl; exit(0); } Calculator cal; Tcp_Server *tsvp new Tcp_Server(argv[1], std::bind(Calculator::Recv_Send, cal, std::placeholders::_1)); tsvp-Start(); int n daemon(0,0); if(n0){ std::cerrdaemon failstd::endl; } tsvp-Run(); }服务器类封装Tcp_Server.hpp#pragma once #include Socket.hpp #include Calprotocol.hpp #include functional #include sys/wait.h using func_t std::functionstd::string(std::string package); class Tcp_Server { public: Tcp_Server(std::string port, func_t callback) : local_sock_(port), callback_(callback) { } void Start() { local_sock_.Bind(); local_sock_.Listen(); } void Run() { signal(SIGCHLD, SIG_IGN); signal(SIGPIPE, SIG_IGN); while (true) { std::string remote_ip; std::string remote_port; // 主线程收到连接后创建子进程然后就关掉fd int fd local_sock_.Accept(remote_ip, remote_port); if (fork() 0) { // 子进程负责读取消息 local_sock_.Close(); // 关掉不需要的 char buffer[1024]; std::string recmessage; while (true) { int n read(fd, buffer, sizeof(buffer) - 1); if (n 0) { buffer[n] \0; } else if (n 0) { LOG(INFO, client close); break; } else { LOG(ERROR, read error: %s, strerror(errno)); break; } // 读取后调用回调函数获取序列化的结果发送给客户端 recmessage buffer; while (true) // 这个循环实现了一次处理多个报文的能力 { std::string sendmessage callback_(recmessage); //如果callback_返回空字符串说明报文不完整 if (sendmessage ) { break; } write(fd, sendmessage.c_str(), sendmessage.size()); } } exit(0); // 子进程自己退出不需要主进程等待 } close(fd);//父进程关闭已连接的fd避免文件描述符泄露。 } } private: Socket local_sock_; func_t callback_; };可以看到为了保证服务端具备一次处理多个报文的能力额外添加了一层循环以检测缓冲区是否还有残余的报文。计算器类Calculator_AND_Protocol.hpp#pragma once #includeCalprotocol.hpp enum{ ModZero1, DivZero, Unknown }; class Calculator{ public: Calculator(){} Calculator(std::string request):request_(request){} void calculate(request_protocolin,response_protocolout){ switch (in.op_) { case : out.result_ in.x_ in.y_; break; case -: out.result_ in.x_ - in.y_; break; case *: out.result_ in.x_ * in.y_; break; case /: { if (in.y_ 0) out.code_ DivZero; else out.result_ in.x_ / in.y_; } break; case %: { if (in.y_ 0) out.code_ ModZero; else out.result_ in.x_ % in.y_; } break; default: out.code_ Unknown; break; } } std::string Recv_Send(std::string buffer){//拿到缓冲区的报文 //用户会传进来一个序列化的任务这里负责反序列化计算结果返回序列化结果 std::string content; if(checkmessage(buffer,content)false){ return ; } std::coutget a question: contentstd::endl; request_protocol in; if(in.deserialization(content)false){ return ; } response_protocol out; calculate(in,out);//计算 std::string outcome; if(out.serialization(outcome)false){ return ; } addsize(outcome); return outcome; } private: std::string request_; };客户端主函数client.cc#include Socket.hpp #include Calprotocol.hpp #include time.h const char op_service[4] {, -, *, /}; int main(int argc, char *argv[]) { srand((unsigned int)time(NULL)); // 客户端在启动时要指定服务端的ip和端口号所以命令应包含3个参数 if (argc ! 3) { std::cout please enter server_ip and server_port std::endl; exit(0); } Socket local_sock; local_sock.Connect(argv[1], argv[2]); int cnt 0; std::string recmessage;//定义到外面不然重新创建 包会丢失 while (cnt 10) { int x rand() % 100; usleep(5000); int y rand() % 100; char op op_service[rand() % sizeof(op_service)]; // 构建请求类对象序列化后发给服务端 request_protocol question(x, y, op); std::cout 第 cnt 次任务: std::endl; question.Print(); std::string check; question.serialization(check); addsize(check); if (cnt 9) { // 测试多次处理 write(local_sock.getid(), check.c_str(), check.size()); write(local_sock.getid(), check.c_str(), check.size()); write(local_sock.getid(), check.c_str(), check.size()); write(local_sock.getid(), check.c_str(), check.size()); } write(local_sock.getid(), check.c_str(), check.size()); sleep(1); char buf[1024]; int n read(local_sock.getid(), buf, sizeof(buf)-1); if (n 0) { buf[n] \0; } else if(n0){ LOG(INFO,server close); exit(0); } else { LOG(ERROR, read error: %s, strerror(errno)); exit(0); } recmessage buf; while (true) { // 一次处理所有收到的响应 response_protocol rec; std::string content; if (checkmessage(recmessage, content) false) { break; } rec.deserialization(content); rec.Print(); } } return 0; }协议相关方法Calprotocol.hpp#pragma once #include string #include string.h #include json/json.h // 分隔符 const std::string content_sep ; const std::string messagesep \n; // 发送前给报文添加总数 bool addsize(std::string message) { std::string temp; size_t size message.size(); temp std::to_string(size); temp messagesep; temp message; temp messagesep; message temp; //??安全吗 return true; } // 检查收到的报文是否完整,还要自检查,另外把缓冲区中对应的字符删掉 bool checkmessage(std::string message, std::string content) { size_t head message.find(messagesep); if (head std::string::npos) { return false; } size_t size std::stoi(message.substr(0, head)); size_t totalsize size head 2; std::string pure_message message.substr(head 1, size); // 只能信任报文中的数字是绝对正确的对比通过read接收到的实际字符,而且不能保证是完全对的 if (message.size() totalsize) { return false; } content pure_message; // 把原缓冲区被使用的报文移除 message.erase(0, totalsize); return true; } class request_protocol { public: request_protocol() { // 反序列化的默认构造 } request_protocol(int x, int y, char op) : x_(x), y_(y), op_(op) { // 序列化的构造 } bool serialization(std::string out) { // x op y序列化 #ifdef MySelf std::string res; res std::to_string(x_); res content_sep; res op_; res content_sep; res std::to_string(y_); out res; return true; #else Json::Value root; root[x] x_; root[y] y_; root[op] std::string(1, op_); Json::StyledWriter w; out w.write(root); // 返回的是string类型的字符串 return true; #endif } bool deserialization(std::string question) { //1 1 #ifdef MySelf size_t head question.find(content_sep); if (head std::string::npos) { // 1 1 return false; } x_ std::stoi(question.substr(0, head)); size_t tail question.rfind(content_sep); if (tail std::string::npos || tail head) { //1 1 return false; } y_ std::stoi(question.substr(tail)); // 计算总数合理性,两个操作数的长度不定只能判断空格数和操作符的总数 if (head 2 ! tail) { return false; } op_ question[head 1]; return true; #else Json::Value root; Json::Reader r; // 这个时候协议的形式已经是前面的json定下来的所以可以从报文中读取相应的分隔符 r.parse(question, root); x_ root[x].asInt(); y_ root[y].asInt(); op_ root[op].asString()[0]; return true; #endif } void Print() { std::cout x: x_ y: y_ op: op_ std::endl; } public: int x_; int y_; char op_; }; class response_protocol { public: response_protocol() { // 反序列化的默认构造 } response_protocol(int result, int code 0) : result_(result), code_(code) { // 序列化的构造 } bool serialization(std::string out) { // x op y序列化 #ifdef MySelf std::string res; res std::to_string(result_); res content_sep; res std::to_string(code_); out res; return true; #else Json::Value root; root[result] result_; root[code] code_; Json::StyledWriter w; out w.write(root); // 返回的是string类型的字符串 return true; #endif } bool deserialization(std::string question) { //1 1 #ifdef MySelf size_t head question.find(content_sep); if (head std::string::npos) { // 1 1 return false; } result_ std::stoi(question.substr(0, head)); size_t tail question.rfind(content_sep); if (tail std::string::npos || tail ! head) { //1 1 return false; } code_ std::stoi(question.substr(tail)); return true; #else Json::Value root; Json::Reader r; // 这个时候协议的形式已经是前面的json定下来的所以可以从报文中读取相应的分隔符 r.parse(question, root); result_ root[result].asInt(); code_ root[code].asInt(); return true; #endif } void Print() { std::cout result: result_ code: code_ std::endl; } int result_; int code_ 0; };makefile.PHONY:all all:tcpserver tcpclient Flag#-DMySelf1 Lib-ljsoncpp tcpserver:server.cc Tcp_Server.hpp Calculator_AND_Protocol.hpp g $ -o $ -stdc11 $(Lib) $(Flag) tcpclient:client.cc Calprotocol.hpp Socket.hpp g $ -o $ -stdc11 -g $(Lib) $(Flag) .PHONY:clean clean: rm -rf tcpserver tcpclient Log编译时中允许可以定义宏因此在makefile中可以定义Flag-DMySelf1编译时自动定义宏MySelf1当启用-DMySelf1时协议类使用自定义分隔符空格序列化不启用时使用 JSON序列化。通过注释可以灵活切换两种实现完整代码详见网络版计算器