跳到主要内容应用层自定义协议与序列化设计 | 极客日志C++算法
应用层自定义协议与序列化设计
应用层自定义协议的设计，通过序列化和反序列化处理结构化数据传输。以网络版计算器为例，展示了基于 TCP 和 Jsoncpp 库实现请求（Request）与响应（Response）的编码解码过程。内容涵盖 Socket 封装、流式数据处理、报文完整性校验及客户端与服务端代码实现，旨在帮助理解网络通信中的协议约定与数据转换机制。
ByteFlow3 浏览 应用层
我们程序员写的一个个解决实际问题、满足日常需求的网络程序，都是在应用层。
再识协议
协议是一种"约定"。Socket API 的接口在读写数据时，都是按"字符串"的方式来发送接收的。如果我们要传输一些"结构化的数据"怎么办呢？
其实，协议就是双方约定好的结构化的数据
结构化数据的传输
通信双方在进行网络通信时：
- 如果需要传输的数据是一个字符串，那么直接将这一个字符串发送到网络当中，此时对端也能从网络当中获取到这个字符串。
- 但如果需要传输的是一些结构化的数据，此时就不能将这些数据一个个发送到网络当中。
这是显而易见的，在传递结构化数据时如果直接发送会面临许多问题，诸如字节序问题、内存对齐问题和数据类型大小不一致等的问题。
那么该怎么传递结构化的数据呢？
制定一个协议 (序列化和反序列化) 来实现
下面以设计一个基于网络 TCP 的计算器为例，我们需要实现一个服务器版的加法器。我们需要客户端把要计算的两个数发过去，然后由服务器进行计算，最后再把结果返回给客户端。
约定方案一
- 客户端发送一个形如 "1+1" 的字符串;
- 这个字符串中有两个操作数，都是整形;
- 两个数字之间会有一个字符是运算符，运算符是 + 或 - 等等;
- 数字和运算符之间没有空格
定制结构体 + 序列化和反序列化
约定方案二
- 定义结构体来表示我们需要交互的信息;
- 发送数据时将这个结构体按照一个规则转换成字符串，接收到数据的时候再按照相同的规则把字符串转化回结构体;
- 这个过程叫做"序列化"和"反序列化"
客户端可以定制一个结构体，将需要交互的信息定义到这个结构体当中。客户端发送数据时先对数据进行序列化，服务端接收到数据后再对其进行反序列化，此时服务端就能得到客户端发送过来的结构体，进而从该结构体当中提取出对应的信息。
序列化和反序列化
序列化和反序列化：
- 序列化是将对象的状态信息转换为可以存储或传输的形式（字节序列）的过程。
- 反序列化是把字节序列恢复为对象的过程。
OSI 七层模型中表示层的作用就是，实现设备固有数据格式和网络标准数据格式的转换。其中设备固有的数据格式指的是数据在应用层上的格式，而网络标准数据格式则指的是序列化之后可以进行网络传输的数据格式。
无论我们采用方案一，还是方案二，还是其他的方案，只要保证，一端发送时构造的数据，在另一端能够正确的进行解析，就是 ok 的。这种约定，就是应用层协议
重新理解 read、write、recv、send 和 tcp 为什么支持全双工
在任何一台主机上，TCP 连接既有发送缓冲区，又有接受缓冲区，所以在内核中，可以在发消息的同时，也可以收消息，即全双工。
网络版计算器
下面实现一个网络版的计算器，主要目的是感受一下什么是协议。
服务层代码
将 TcpSocket 封装成一个类。
微信扫一扫，关注极客日志
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
相关免费在线工具
- 加密/解密文本
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
- Base64 字符串编码/解码
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
- Base64 文件转换器
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
- Markdown转HTML
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online
- HTML转Markdown
将 HTML 片段转为 GitHub Flavored Markdown，支持标题、列表、链接、代码块与表格等；浏览器内处理，可链接预填。 在线工具，HTML转Markdown在线工具，online
- JSON 压缩
通过删除不必要的空白来缩小和压缩JSON。 在线工具，JSON 压缩在线工具，online
namespace SocketModule {
using namespace LogModule;
const static int gbacklog = 16;
class Socket {
public:
virtual ~Socket() {}
virtual void SocketOrDie() = 0;
virtual void BindOrDie(uint16_t port) = 0;
virtual void ListenOrDie(int backlog) = 0;
virtual std::shared_ptr<Socket> Accept(InetAddr *client) = 0;
virtual void Close() = 0;
virtual int Recv(std::string *out) = 0;
virtual int Send(const std::string &message) = 0;
virtual int Connect(const std::string &server_ip, uint16_t port) = 0;
public:
void BuildTcpSocketMethod(uint16_t port, int backlog = gbacklog) {
SocketOrDie();
BindOrDie(port);
ListenOrDie(backlog);
}
void BuildTcpClientSocketMethod() {
SocketOrDie();
}
};
const static int defaultfd = -1;
class TcpSocket : public Socket {
public:
TcpSocket() : _sockfd(defaultfd) { }
TcpSocket(int fd) : _sockfd(fd) { }
~TcpSocket() { }
void SocketOrDie() override {
_sockfd = ::socket(AF_INET, SOCK_STREAM, 0);
if (_sockfd < 0) {
LOG(LogLevel::FATAL) << "socket error";
exit(SOCKET_ERR);
}
LOG(LogLevel::INFO) << "socket success";
}
void BindOrDie(uint16_t port) override {
InetAddr local(port);
int n = ::bind(_sockfd, local.NetAddrPtr(), local.NetAddrLen());
if (n < 0) {
LOG(LogLevel::ERROR) << "bind error";
exit(BIND_ERR);
}
LOG(LogLevel::INFO) << "bind success";
}
void ListenOrDie(int backlog) override {
int n = ::listen(_sockfd, backlog);
if (n < 0) {
LOG(LogLevel::ERROR) << "listen error";
exit(LIST_ERR);
}
LOG(LogLevel::INFO) << "listen success";
}
std::shared_ptr<Socket> Accept(InetAddr *client) override {
struct sockaddr_in peer;
socklen_t len = sizeof(peer);
int fd = ::accept(_sockfd, CONV(peer), &len);
if (fd < 0) {
LOG(LogLevel::WARNING) << "accept warning";
return nullptr;
}
client->SetAddr(peer);
return std::make_shared<TcpSocket>(fd);
}
int Recv(std::string *out) override {
char buffer[1024];
ssize_t n = ::recv(_sockfd, buffer, sizeof(buffer) - 1, 0);
if (n > 0) {
buffer[n] = 0;
*out += buffer;
}
return n;
}
int Send(const std::string &message) override {
return send(_sockfd, message.c_str(), message.size(), 0);
}
void Close() override {
if (_sockfd >= 0) ::close(_sockfd);
}
int Connect(const std::string &server_ip, uint16_t port) override {
InetAddr server(server_ip, port);
return ::connect(_sockfd, server.NetAddrPtr(), server.NetAddrLen());
}
private:
int _sockfd;
};
}
定制协议
要实现一个网络版的计算器，就必须保证通信双方能够遵守某种协议约定，因此我们需要设计一套简单的约定。数据可以分为请求数据和响应数据，因此我们分别需要对请求数据和响应数据进行约定。我们没有自己实现一个序列化和反序列化的类，下面代码都是通过 Jsoncpp 库提供的方法来实现序列化与反序列化详情见附录。
Request 类 (请求数据)
class Request {
public:
Request() { }
Request(int x, int y, char oper) : _x(x), _y(y), _oper(oper) { }
std::string Serialize() {
Json::Value root;
root["x"] = _x;
root["y"] = _y;
root["oper"] = _oper;
Json::FastWriter writer;
std::string s = writer.write(root);
return s;
}
bool Deserialize(std::string &in) {
Json::Value root;
Json::Reader reader;
bool ok = reader.parse(in, root);
if (ok) {
_x = root["x"].asInt();
_y = root["y"].asInt();
_oper = root["oper"].asInt();
}
return ok;
}
~Request() {}
int X() { return _x; }
int Y() { return _y; }
char Oper() { return _oper; }
private:
int _x;
int _y;
char _oper;
};
Response 类 (响应数据)
class Response {
public:
Response() {}
Response(int result, int code) : _result(result), _code(code) { }
std::string Serialize() {
Json::Value root;
root["result"] = _result;
root["code"] = _code;
Json::FastWriter writer;
return writer.write(root);
}
bool Deserialize(std::string &in) {
Json::Value root;
Json::Reader reader;
bool ok = reader.parse(in, root);
if (ok) {
_result = root["result"].asInt();
_code = root["code"].asInt();
}
return ok;
}
~Response() {}
void SetResult(int res) { _result = res; }
void SetCode(int code) { _code = code; }
private:
int _result;
int _code;
};
Protocol 类 (协议)
const std::string sep = "\r\n";
using func_t = std::function<Response(Request &req)>;
class Protocol {
public:
Protocol() { }
Protocol(func_t func) : _func(func) { }
std::string Encode(const std::string &jsonstr) {
std::string len = std::to_string(jsonstr.size());
return len + sep + jsonstr + sep;
}
bool Decode(std::string &buffer, std::string *package) {
ssize_t pos = buffer.find(sep);
if (pos == std::string::npos) return false;
std::string package_len_str = buffer.substr(0, pos);
int package_len_int = std::stoi(package_len_str);
int target_len = package_len_str.size() + package_len_int + 2 * sep.size();
if (buffer.size() < target_len) return false;
*package = buffer.substr(pos + sep.size(), package_len_int);
buffer.erase(0, target_len);
return true;
}
void GetRequest(std::shared_ptr<Socket> &sock, InetAddr &client) {
std::string buffer_queue;
while (true) {
int n = sock->Recv(&buffer_queue);
if (n > 0) {
std::cout << "-----------request_buffer--------------" << std::endl;
std::cout << buffer_queue << std::endl;
std::cout << "------------------------------------" << std::endl;
std::string json_package;
while (Decode(buffer_queue, &json_package)) {
LOG(LogLevel::DEBUG) << client.StringAddr() << " 请求：" << json_package;
Request req;
bool ok = req.Deserialize(json_package);
if (!ok) continue;
Response resp = _func(req);
std::string json_str = resp.Serialize();
std::string send_str = Encode(json_str);
sock->Send(send_str);
}
} else if (n == 0) {
LOG(LogLevel::INFO) << "client:" << client.StringAddr() << "Quit!";
break;
} else {
LOG(LogLevel::WARNING) << "client:" << client.StringAddr() << ", recv error";
break;
}
}
}
bool GetResponse(std::shared_ptr<Socket> &client, std::string &resp_buff, Response *resp) {
while (true) {
int n = client->Recv(&resp_buff);
if (n > 0) {
std::string json_package;
while (Decode(resp_buff, &json_package)) {
resp->Deserialize(json_package);
}
return true;
} else if (n == 0) {
std::cout << "server quit " << std::endl;
return false;
} else {
std::cout << "recv error" << std::endl;
return false;
}
}
}
std::string BuildRequestString(int x, int y, char oper) {
Request req(x, y, oper);
std::string json_req = req.Serialize();
return Encode(json_req);
}
~Protocol() { }
private:
func_t _func;
};
- 客户端写入服务器时是依据协议的 (Encode)，这样我们才能读到想要的数据
- 收到数据后我们进行解析报文 (Decode) 来正确读到序列化后 (即一串字符串) 的数据。
然后通过反序列化读到 x、y、oper 的值并依此初始化 Request
- 我们在向应用层发送数据时也要依据协议，发送的报文要序列化 + 报头 (Encode)
关于流式数据的处理
- TCP 协议会将数据拆分为多个 IP 包传输，接收方的
recv()可能只读到部分数据。
- 操作系统内核的 Socket 缓冲区大小有限，可能分批交付数据。
- 网络延迟或拥塞可能导致数据分片到达。
bool Decode(std::string &buffer, std::string *package) {
ssize_t pos = buffer.find(sep);
if (pos == std::string::npos) return false;
std::string package_len_str = buffer.substr(0, pos);
int package_len_int = std::stoi(package_len_str);
int target_len = package_len_str.size() + package_len_int + 2 * sep.size();
if (buffer.size() < target_len) return false;
*package = buffer.substr(pos + sep.size(), package_len_int);
buffer.erase(0, target_len);
return true;
}
首先读取的数据要有分隔符 sep(\r\n)，如果没有找到则返回错误
每条报文的报头是报文的长度 package_len_str，一条完整的报文包括报头、两个分隔符、报文具体内容 (target_len)。如果传过来的 buffer 小于这个长度显然消息不完整返回错误。
NetCal 计算机类
class Cal {
public:
Response Execute(Request &req) {
Response resp(0, 0);
switch (req.Oper()) {
case '+': resp.SetResult(req.X() + req.Y()); break;
case '-': resp.SetResult(req.X() - req.Y()); break;
case '*': resp.SetResult(req.X() * req.Y()); break;
case '/': {
if (req.Y() == 0) {
resp.SetCode(1);
} else {
resp.SetResult(req.X() / req.Y());
}
} break;
case '%': {
if (req.Y() == 0) {
resp.SetCode(2);
} else {
resp.SetResult(req.X() % req.Y());
}
} break;
default: resp.SetCode(3);
}
return resp;
}
};
Client.cc(客户端)
using namespace SocketModule;
void Usage(std::string proc) {
std::cerr << "Usage: " << proc << " server_ip server_port" << std::endl;
}
void GetDataFromStdin(int *x, int *y, char *oper) {
std::cout << "Please Enter x: ";
std::cin >> *x;
std::cout << "Please Enter y: ";
std::cin >> *y;
std::cout << "Please Enter oper: ";
std::cin >> *oper;
}
int main(int argc, char *argv[]) {
if (argc != 3) {
Usage(argv[0]);
exit(USAGE_ERR);
}
std::string server_ip = argv[1];
uint16_t server_port = std::stoi(argv[2]);
std::shared_ptr<Socket> client = std::make_shared<TcpSocket>();
client->BuildTcpClientSocketMethod();
if (client->Connect(server_ip, server_port) != 0) {
std::cerr << "connect error" << std::endl;
exit(CONNECT_ERR);
}
std::unique_ptr<Protocol> protocol = std::make_unique<Protocol>();
std::string resp_buffer;
while (true) {
int x, y;
char oper;
GetDataFromStdin(&x, &y, &oper);
std::string req_str = protocol->BuildRequestString(x, y, oper);
client->Send(req_str);
Response resp;
bool res = protocol->GetResponse(client, resp_buffer, &resp);
if(res == false) break;
resp.ShowResult();
}
client->Close();
return 0;
}
main.cc
void Usage(std::string proc) {
std::cerr << "Usage: " << proc << " port" << std::endl;
}
int main(int argc, char *argv[]) {
if (argc != 2) {
Usage(argv[0]);
exit(USAGE_ERR);
}
std::unique_ptr<Cal> cal = std::make_unique<Cal>();
std::unique_ptr<Protocol> protocol = std::make_unique<Protocol>([&cal](Request &req) {
return cal->Execute(req);
});
std::unique_ptr<TcpServer> tsvr = std::make_unique<TcpServer>(std::stoi(argv[1]), [&protocol](std::shared_ptr<Socket> &sock, InetAddr &client) {
protocol->GetRequest(sock, client);
});
tsvr->Start();
return 0;
}
效果演示
附录
Jsoncpp
Jsoncpp 是一个用于处理 JSON 数据的 C++ 库。它提供了将 JSON 数据序列化为字符串以及从字符串反序列化为 C++ 数据结构的功能。Jsoncpp 是开源的，广泛用于各种需要处理 JSON 数据的 C++ 项目中。
- 简单易用：Jsoncpp 提供了直观的 API，使得处理 JSON 数据变得简单。
- 高性能：Jsoncpp 的性能经过优化，能够高效地处理大量 JSON 数据。
- 全面支持：支持 JSON 标准中的所有数据类型，包括对象、数组、字符串、数字、布尔值和 null。
- 错误处理：在解析 JSON 数据时，Jsoncpp 提供了详细的错误信息和位置，方便开发者调试。
当使用 Jsoncpp 库进行 JSON 的序列化和反序列化时，确实存在不同的做法和工具类可供选择。以下是对 Jsoncpp 中序列化和反序列化操作的详细介绍：
序列化
序列化指的是将数据结构或对象转换为一种格式，以便在网络上传输或存储到文件中。Jsoncpp 提供了多种方式进行序列化：
- 使用 Json::Value 的 toStyledString 方法：
#include <iostream>
#include <string>
#include <jsoncpp/json/json.h>
int main() {
Json::Value root;
root["name"] = "joe";
root["sex"] = "男";
std::string s = root.toStyledString();
std::cout << s << std::endl;
return 0;
}
$./ test.exe
{ "name" : "joe", "sex" : "男" }
#include <iostream>
#include <string>
#include <sstream>
#include <memory>
#include <jsoncpp/json/json.h>
int main() {
Json::Value root;
root["name"] = "joe";
root["sex"] = "男";
Json::StreamWriterBuilder wbuilder;
std::unique_ptr<Json::StreamWriter> writer(wbuilder.newStreamWriter());
std::stringstream ss;
writer->write(root, &ss);
std::cout << ss.str() << std::endl;
return 0;
}
$./ test.exe
{ "name" : "joe", "sex" : "男" }
#include <iostream>
#include <string>
#include <sstream>
#include <memory>
#include <jsoncpp/json/json.h>
int main() {
Json::Value root;
root["name"] = "joe";
root["sex"] = "男";
Json::FastWriter writer;
std::string s = writer.write(root);
std::cout << s << std::endl;
return 0;
}
$./ test.exe
{"name" : "joe", "sex" : "男"}
反序列化
反序列化指的是将序列化后的数据重新转换为原来的数据结构或对象。Jsoncpp 提供了以下方法进行反序列化：
#include <iostream>
#include <string>
#include <jsoncpp/json/json.h>
int main() {
std::string json_string = "{\"name\":\"张三\", \"age\":30, \"city\":\"北京\"}";
Json::Reader reader;
Json::Value root;
bool parsingSuccessful = reader.parse(json_string, root);
if (!parsingSuccessful) {
std::cout << "Failed to parse JSON: " << reader.getFormattedErrorMessages() << std::endl;
return 1;
}
std::string name = root["name"].asString();
int age = root["age"].asInt();
std::string city = root["city"].asString();
std::cout << "Name: " << name << std::endl;
std::cout << "Age: " << age << std::endl;
std::cout << "City: " << city << std::endl;
return 0;
}
$./ test.exe
Name : 张三
Age : 30
City : 北京
总结
- toStyledString、StreamWriter 和 FastWriter 提供了不同的序列化选项，你可以根据具体需求选择使用。
- Json::Reader 和 parseFromStream 函数是 Jsoncpp 中主要的反序列化工具，它们提供了强大的错误处理机制。
- 在进行序列化和反序列化时，请确保处理所有可能的错误情况，并验证输入和输出的有效性。
