Linux 网络 序列化与反序列化~
概念
序列化（Serialization）是将对象的状态信息转换为可以存储或传输的形式的过程。以下是关于序列化与反序列化的介绍：
- 序列化：将对象的状态信息转换为可以存储或传输的格式，通常是字节序列或文本格式。
- 反序列化：将序列化后的数据还原为原始对象或数据结构的过程。
出现原因
序列化的出现主要是为了满足在不同系统、不同语言之间进行数据传输和存储的需求，以下是具体原因：
- 跨平台和跨语言通信：不同的操作系统和编程语言对数据的表示和存储方式各不相同。例如，Java中的对象在内存中的布局和C++中的对象就有很大差异。通过序列化，可以将数据转换为一种通用的格式，如JSON或XML，这样不同平台和语言编写的程序就能够相互理解和处理这些数据，实现跨平台和跨语言的通信。
- 网络传输：在网络通信中，数据是以字节流的形式传输的。为了能够在网络上传输复杂的数据结构和对象，需要将它们序列化为字节流，然后在接收端进行反序列化，还原为原始的数据结构和对象。
- 数据持久化：将对象存储到磁盘或数据库中时，需要先将其序列化为字节流或特定的存储格式，以便能够在需要时进行反序列化恢复。
常见的序列化格式
- JSON：一种轻量级的数据交换格式，易于阅读和编写，广泛应用于Web开发和API设计中。
- XML：一种标记语言，具有良好的扩展性和可读性，常用于数据存储和配置文件。
- Protocol Buffers：一种高效的二进制序列化格式，具有较小的存储空间和较快的解析速度，常用于分布式系统和大数据应用中。
JSON
其中，JSON（JavaScript Object Notation）是一种轻量级的数据交换格式，比较常用，以下是关于它的介绍：
特点
- 轻量级：JSON比XML更小、更快，更易解析，占用带宽小，适合在网络上传输。
- 易于阅读和编写：JSON采用类似于C语言家族的习惯，易于人阅读和编写，同时也易于机器解析和生成。
- 独立于语言：JSON使用JavaScript语法来描述数据对象，但它独立于语言和平台，支持多种编程语言，如C、C++、Java、Python、PHP等。
数据结构
- 对象：对象是一个无序的“名称/值”对集合，以 { 左括号开始， } 右括号结束，每个“名称”后跟一个 : 冒号，“名称/值”对之间使用, 逗号分隔。
- 数组：数组是值的有序集合，以 [ 左中括号开始， ] 右中括号结束，值之间使用, 逗号分隔。
- 值：值可以是双引号括起来的字符串、数值、true、false、null、对象或者数组，这些结构可以嵌套。
例如，有这样的数据
hello2025.1.18Mike
经过JSON序列化后，变成以下格式
{
"message": "hello",
"time": "2025.1.18",
"name": "Mike"
}
Jsoncpp
Jsoncpp
是一个用于处理
JSON
数据的
C++
库。它提供了将
JSON
数据序列化为字符串以及从字符串反序列化为 C++
数据结构的功能。
Jsoncpp
是开源的，广泛用于各种需要处理 JSON
数据的
C++
项目中。
安装
在Linux中，在不同的环境下可以使用对应的指令安装
C++
ubuntu：sudo apt-get install libjsoncpp-dev
Centos: sudo yum install jsoncpp-devel安装成功后就会包含在 /usr/include/jsoncpp 中，所以我们在使用该库时需要包含头文件<jsoncpp/json/json.h>.
序列化
序列化指的是将数据结构或对象转换为一种格式，以便在网络上传输或存储到文件中。Jsoncpp
提供了多种方式进行序列化：
使用 Json::Value 的 toStyledString 方法
优点：将 Json::Value 对象直接转换为格式化的 JSON 字符串。
示例：
#include <iostream>
#include <string>
#include <jsoncpp/json/json.h>
using namespace std;
int main()
{
Json::Value root;
root["name"] = "joe";
root["sex"] = "男";
string s = root.toStyledString();
cout << s << endl;
return 0;
}
执行成功后的结果如下：
{
"name" : "joe",
"sex" : "男"
}使用 Json::StreamWriter
优点：提供了更多的定制选项，如缩进、换行符等。
示例：
#include <iostream>
#include <string>
#include <sstream>
#include <memory>
#include <jsoncpp/json/json.h>
using namespace std;
int main()
{
Json::Value root;
root["name"] = "joe";
root["sex"] = "男";
Json::StreamWriterBuilder wbuilder; // StreamWriter 的工厂
unique_ptr<Json::StreamWriter> writer(wbuilder.newStreamWriter());
stringstream ss;
writer->write(root, &ss);
cout << ss.str() << endl;
return 0;
}
执行成功后的结果如下：
{
"name" : "joe",
"sex" : "男"
}使用 Json::FastWriter
优点：比 StyledWriter
更快，因为它不添加额外的空格和换行符。
示例：
#include <iostream>
#include <string>
#include <sstream>
#include <memory>
#include <jsoncpp/json/json.h>
using namespace std;
int main()
{
Json::Value root;
root["name"] = "joe";
root["sex"] = "男";
Json::FastWriter writer;
string s = writer.write(root);
cout << s << endl;
return 0;
}
执行成功后的结果如下：
{"name":"joe","sex":"男"}反序列化
反序列化指的是将序列化后的数据重新转换为原来的数据结构或对象。
Jsoncpp
提供了以下方法进行反序列化：
使用 Json::Reader
优点：提供详细的错误信息和位置，方便调试。
示例：
#include <iostream>
#include <string>
#include <jsoncpp/json/json.h>
using namespace std;
int main() {
// JSON 字符串
string json_string = "{\"name\":\"张三\",\"age\":30, \"city\":\"北京\"}";
// 解析 JSON 字符串
Json::Reader reader;
Json::Value root;
// 从字符串中读取 JSON 数据
bool parsingSuccessful = reader.parse(json_string,root);
if (!parsingSuccessful) {
// 解析失败，输出错误信息
cout << "Failed to parse JSON: " << reader.getFormattedErrorMessages() << endl;
return 1;
}
// 访问 JSON 数据
string name = root["name"].asString();
int age = root["age"].asInt();
string city = root["city"].asString();
// 输出结果
cout << "Name: " << name << endl;
cout << "Age: " << age << endl;
cout << "City: " << city << endl;
return 0;
}执行成功后的结果如下：
Name: 张三
Age: 30
City: 北京案例 网络版计算器
例如
,
我们需要实现一个服务器版的加法器，
我们需要客户端把要计算的两个加数发过去,
然后由服务器进行计算
,
最后再把结果返回给客户端。
makefile
all: server client
server:TcpServermain.cc
g++ -o $@ $^ -std=c++17 -lpthread -ljsoncpp
client:TcpClient.cc
g++ -o $@ $^ -std=c++17 -ljsoncpp
.PHONY:clean
clean:
rm -f server clientMutex.hpp
#pragma once
#include <iostream>
#include <pthread.h>
using namespace std;
class Mutex
{
public:
Mutex(const Mutex&)=delete;
const Mutex& operator=(const Mutex&)=delete;
Mutex()
{
pthread_mutex_init(&_lock,nullptr);
}
~Mutex()
{
pthread_mutex_destroy(&_lock);
}
void Lock()
{
pthread_mutex_lock(&_lock);
}
pthread_mutex_t * LockPtr()
{
return &_lock;
}
void Unlock()
{
pthread_mutex_unlock(&_lock);
}
private:
pthread_mutex_t _lock;
};
class LockGuard
{
public:
LockGuard(Mutex& m)
:_mutex(m)
{
_mutex.Lock();
}
~LockGuard()
{
_mutex.Unlock();
}
private:
Mutex& _mutex;
};Cond.hpp
#pragma once
#include"Mutex.hpp"
class Cond
{
public:
Cond()
{
pthread_cond_init(&_cond,nullptr);
}
~Cond()
{
pthread_cond_destroy(&_cond);
}
void Wait(Mutex& mutex)
{
pthread_cond_wait(&_cond,mutex.LockPtr());
}
void Notify()
{
pthread_cond_signal(&_cond);
}
void NotifyAll()
{
pthread_cond_broadcast(&_cond);
}
private:
pthread_cond_t _cond;
};Thread.hpp
#pragma once
#include <pthread.h>
#include <iostream>
#include <functional>
#include <string>
#include <unistd.h>
using namespace std;
using func_t = function<void(string)>;
static int number = 1;
enum STATUS
{
NEW,
RUNNING,
STOP
};
class Thread
{
private:
static void *Routine(void *arg)
{
Thread *t = static_cast<Thread *>(arg);
t->_func(t->_name);
return nullptr;
}
public:
Thread(func_t func)
: _func(func), _status(NEW), _joinable(true)
{
_name = "Thread-" + to_string(number++);
_pid = getpid();
}
bool Start()
{
if (_status != RUNNING)
{
_status = RUNNING;
int n = pthread_create(&_tid, nullptr, Routine, this);
if (n != 0)
{
return false;
}
return true;
}
return false;
}
bool Stop()
{
if (_status == RUNNING)
{
_status = STOP;
int n = pthread_cancel(_tid);
if (n != 0)
{
return false;
}
return true;
}
return false;
}
bool Join()
{
if (_joinable)
{
_status = STOP;
int n = pthread_join(_tid, nullptr);
if (n != 0)
{
return false;
}
return true;
}
return false;
}
void Detach()
{
_joinable = false;
pthread_detach(_tid);
}
string Name()
{
return _name;
}
private:
string _name;
pthread_t _tid;
pid_t _pid;
STATUS _status;
bool _joinable;
func_t _func;
};ThreadPool.hpp
#pragma once
#include <iostream>
#include <string>
#include <queue>
#include <vector>
#include <memory>
#include "Mutex.hpp"
#include "Cond.hpp"
#include "Thread.hpp"
using thread_t = shared_ptr<Thread>;
const static int defaultnum = 5;
template <class T>
class ThreadPool
{
private:
bool IsEmpty() { return _taskq.empty(); }
void HandlerTask(string name)
{
cout << "线程: " << name << ", 进入HandlerTask的逻辑" << endl;
while (true)
{
// 1. 拿任务
T t;
{
LockGuard lockguard(_lock);
while (IsEmpty() && _isrunning)
{
_wait_num++;
_cond.Wait(_lock);
_wait_num--;
}
// 2. 任务队列为空 && 线程池退出了
if (IsEmpty() && !_isrunning)
break;
t = _taskq.front();
_taskq.pop();
}
// 2. 处理任务
t(); // 规定，未来所有的任务处理，全部都是必须提供t()方法！
}
cout << "线程: " << name << " 退出";
}
ThreadPool(const ThreadPool<T> &) = delete;
ThreadPool<T> &operator=(const ThreadPool<T> &) = delete;
ThreadPool(int num = defaultnum) : _num(num), _wait_num(0), _isrunning(false)
{
for (int i = 0; i < _num; i++)
{
_threads.push_back(make_shared<Thread>(bind(&ThreadPool::HandlerTask, this, std::placeholders::_1)));
cout << "构建线程" << _threads.back()->Name() << "对象 ... 成功" << endl;
}
}
public:
static ThreadPool<T> *getInstance()
{
if (instance == NULL)
{
LockGuard lockguard(mutex);
if (instance == NULL)
{
cout << "单例首次被执行，需要加载对象..." << endl;
instance = new ThreadPool<T>();
instance->Start();
}
}
return instance;
}
void Equeue(T in)
{
LockGuard lockguard(_lock);
if (!_isrunning)
return;
_taskq.push(in);
if (_wait_num > 0)
_cond.Notify();
}
void Start()
{
if (_isrunning)
return;
_isrunning = true;
for (auto &thread_ptr : _threads)
{
cout << "启动线程" << thread_ptr->Name() << " ... 成功";
thread_ptr->Start();
}
}
void Wait()
{
for (auto &thread_ptr : _threads)
{
thread_ptr->Join();
cout << "回收线程" << thread_ptr->Name() << " ... 成功";
}
}
void Stop()
{
LockGuard lockguard(_lock);
if (_isrunning)
{
_isrunning = false; // 不工作
// 1. 让线程自己退出(要唤醒) && // 2. 历史的任务被处理完了
if (_wait_num > 0)
_cond.NotifyAll();
}
}
private:
vector<thread_t> _threads;
int _num;
int _wait_num;
std::queue<T> _taskq; // 临界资源
Mutex _lock;
Cond _cond;
bool _isrunning;
static ThreadPool<T> *instance;
static Mutex mutex; // 只用来保护单例
};
template <class T>
ThreadPool<T> *ThreadPool<T>::instance = NULL;
template <class T>
Mutex ThreadPool<T>::mutex; // 只用来保护单例InetAddr.hpp
#pragma once
#include <string>
#include <iostream>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <cstring>
using namespace std;
class InetAddr
{
public:
InetAddr();
InetAddr(int port, string ip = "")
: _port(port), _ip(ip)
{
bzero(&_sockaddr, sizeof(_sockaddr));
_sockaddr.sin_family = AF_INET;
_sockaddr.sin_port = htons(_port);
if (_ip.empty())
_sockaddr.sin_addr.s_addr = INADDR_ANY;
else
_sockaddr.sin_addr.s_addr = inet_addr(_ip.c_str());
}
InetAddr(const struct sockaddr_in &sockaddr)
{
_port = ntohs(sockaddr.sin_port);
char buf[64];
_ip = inet_ntop(AF_INET, &sockaddr.sin_addr, buf, sizeof(buf));
}
bool operator==(const InetAddr &other)
{
return _ip == other._ip;
}
InetAddr operator=(const InetAddr &other)
{
_ip = other._ip;
_port = other._port;
_sockaddr = other._sockaddr;
return *this;
}
struct sockaddr *getSockaddr()
{
return (struct sockaddr *)&_sockaddr;
}
int getSockaddrLen()
{
return sizeof(_sockaddr);
}
const string &getIp()
{
return _ip;
}
int getPort()
{
return _port;
}
private:
string _ip;
int _port;
struct sockaddr_in _sockaddr;
};Common.hpp
enum
{
SOCKET_ERROR=1,
BIND_ERROR,
LISTEN_ERROR,
ACCEPT_ERROR,
CONNECT_ERROR
};Protocol.hpp
#pragma once
#include <iostream>
#include <string>
#include <jsoncpp/json/json.h>
using namespace std;
const string Sep = "\r\n";
// 给信息添加报头
//{json} -> length\r\n{json}\r\n
bool Encode(string &message)
{
if (message.size() == 0)
return false;
string package = to_string(message.size()) + Sep + message + Sep;
message = package;
return true;
}
// 解析协议，提取信息
bool Decode(string &package, string *message)
{
auto pos = package.find(Sep);
if (pos == string::npos)
return false;
string message_length_str = package.substr(0, pos);
int message_length = stoi(message_length_str);
int full_length = message_length_str.size() + 2 * Sep.size() + message_length;
if (package.size() < full_length)
return false;
*message = package.substr(pos + Sep.size(), message_length);
package.erase(0,full_length);
return true;
}
class Request
{
public:
Request()
{
}
Request(int x, int y, char op)
: _x(x), _y(y), _op(op)
{
}
// 使用jsoncpp序列化
void Serialize(string &out_str)
{
Json::Value root;
root["x"] = _x;
root["y"] = _y;
root["op"] = _op;
out_str = root.toStyledString();
}
// 反序列化
bool Deserialize(string &in_str)
{
Json::Value root;
Json::Reader reader;
bool parsingSuccessful = reader.parse(in_str, root);
if (!parsingSuccessful)
{
cout << "Failed to parse JSON: " << reader.getFormattedErrorMessages();
return false;
}
_x = root["x"].asInt();
_y = root["y"].asInt();
_op = root["op"].asInt();
return true;
}
void Print()
{
cout<<"x: "<<_x << endl;
cout<<"y: "<<_y << endl;
cout<<"op: "<<_op << endl;
}
int X() const
{
return _x;
}
int Y() const
{
return _y;
}
char Op() const
{
return _op;
}
private:
int _x, _y;
char _op;
};
class Response
{
public:
Response()
{}
Response(int result ,int code)
:_result(result),_code(code)
{}
void Serialize(string& out_str)
{
Json::Value root;
root["result"]=_result;
root["code"]=_code;
out_str=root.toStyledString();
}
bool Deserialize(string& in_str)
{
Json::Value root;
Json::Reader reader;
bool parsingsuccessful=reader.parse(in_str,root);
if(!parsingsuccessful)
{
cout << "Failed to parse JSON: " << reader.getFormattedErrorMessages() << endl;
return false;
}
_result = root["result"].asInt();
_code = root["code"].asInt();
return true;
}
void SetResult(int res)
{
_result=res;
}
void SetCode(int c)
{
_code=c;
}
int Result()
{
return _result;
}
int Code()
{
return _code;
}
private:
int _result = 0;
int _code = 0;
};TcpServer.hpp
#pragma once
#include <iostream>
#include <pthread.h>
#include <functional>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <cstring>
#include <memory>
#include "Common.hpp"
#include "InetAddr.hpp"
#include "ThreadPool.hpp"
using namespace std;
#define BACKLOG 8
using handler_t = function<string(string &)>;
static const uint16_t gport = 8080;
class TcpServer
{
using task_t = function<void()>;
struct ThreadData
{
int sockfd;
TcpServer *self;
};
public:
TcpServer(handler_t handler, uint16_t port = gport)
: _handler(handler), _port(port), _isrunning(false)
{
}
void InitServer()
{
// 创建socket
_listensockfd = socket(AF_INET, SOCK_STREAM, 0);
if (_listensockfd < 0)
{
cout << "socket error" << endl;
exit(SOCKET_ERROR);
}
cout << "socket create success,sockfd is: " << _listensockfd << endl;
// 填写IP端口
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;
// bind
int ret = bind(_listensockfd, (struct sockaddr *)&local, sizeof(local));
if (ret < 0)
{
cout << "bind error" << endl;
exit(BIND_ERROR);
}
cout << "bind success" << endl;
// 将socket设置为监听状态
ret = listen(_listensockfd, BACKLOG);
if (ret < 0)
{
cout << "listen error" << endl;
exit(LISTEN_ERROR);
}
cout << "listen success" << endl;
}
void HandleRequest(int sockfd) // TCP为全双工通信
{
char buffer[4096];
string package;
while (true)
{
// int n = read(sockfd, buffer, sizeof(buffer) - 1);
int n = recv(sockfd, buffer, sizeof(buffer) - 1, 0);
if (n > 0)
{
buffer[n] = 0;
cout << buffer << endl;
package += buffer;
string cmd_result = _handler(package);
// write(sockfd, cmd_result.c_str(), cmd_result.size());
if (cmd_result.empty())
continue;
cout << cmd_result<<endl;
send(sockfd, cmd_result.c_str(), cmd_result.size(), 0);
}
else if (n == 0)
{
// 如果读取的值为0，说明client退出
cout << "client quit" << endl;
break;
}
else
{
// 读取失败
break;
}
}
close(sockfd);
}
void Start()
{
_isrunning = true;
while (_isrunning)
{
// 获取新连接
struct sockaddr_in peer;
socklen_t peerlen = sizeof(peer);
int sockfd = accept(_listensockfd, (struct sockaddr *)&peer, &peerlen);
if (sockfd < 0)
{
cout << "accept error" << endl;
exit(ACCEPT_ERROR);
}
cout << "accept success,sockfd is: " << sockfd << endl;
InetAddr addr(peer);
cout << "client info: " << addr.getIp() << ":" << addr.getPort() << endl;
// 将任务交给线程池
ThreadPool<task_t>::getInstance()->Equeue(
[&]()
{
this->HandleRequest(sockfd);
});
}
}
void Stop()
{
_isrunning = false;
}
private:
int _listensockfd; // 监听socket
uint16_t _port;
bool _isrunning;
// 处理上层任务的入口
handler_t _handler;
};TcpServermain.cc
#include "TcpServer.hpp"
#include "CommandExec.hpp"
#include "Daemon.hpp"
#include "Calculator.hpp"
#include <functional>
#include <unistd.h>
#include <memory>
// 解析package
using cal_func = function<Response(const Request &)>;
class Parse
{
public:
Parse(cal_func func)
: _func(func)
{
}
// 提取报文中一次计算的完整信息
string Entry(string &package)
{
// 判断报文完整性
string message;
string resstr;
while (Decode(package, &message))
{
cout << message;
if (message.empty())
break;
// 反序列化
Request req;
if (!req.Deserialize(message))
break;
cout << "Request: ";
req.Print();
// 计算
Response res = _func(req);
// 序列化
string tmp;
res.Serialize(tmp);
cout << "序列化: " << tmp << endl;
// 添加报头
Encode(tmp);
cout << "Encode: " << tmp;
// 拼接应答
resstr += tmp;
}
return resstr;
}
private:
cal_func _func;
};
int main()
{
// 变成守护进程
Daemon(false, false);
Calculator mycal;
Parse mypar([&](const Request &req)
{ return mycal.Execute(req); });
unique_ptr<TcpServer> server = make_unique<TcpServer>([&](string& package)
{ return mypar.Entry(package); });
server->InitServer();
server->Start();
return 0;
}TcpClient.cc
#include <iostream>
#include <string>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <sys/types.h>
#include <arpa/inet.h>
#include <cstring>
using namespace std;
#include "Common.hpp"
#include "Protocol.hpp"
//./client server_ip server_port
int main(int argc, char *argv[])
{
if (argc != 3)
{
cout << "Usage:./client server_ip server_port" << endl;
return 0;
}
string server_ip = argv[1];
int server_port = stoi(argv[2]);
int sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd < 0)
{
cout << "socket create error" << endl;
exit(SOCKET_ERROR);
}
// 填写网络信息
struct sockaddr_in server_addr;
bzero(&server_addr, sizeof(server_addr));
server_addr.sin_family = AF_INET;
server_addr.sin_port = htons(server_port);
server_addr.sin_addr.s_addr = inet_addr(server_ip.c_str());
// client 无需显示bind，connect连接时自动bind
int n = connect(sockfd, (struct sockaddr *)&server_addr, sizeof(server_addr));
if (n < 0)
{
cout << "connect error" << endl;
exit(CONNECT_ERROR);
}
string message;
while (true)
{
int x, y;
char op;
cout << "input x: ";
cin >> x;
cout << "input y: ";
cin >> y;
cout << "input op: ";
cin >> op;
Request req(x, y, op);
req.Serialize(message); // 序列化
Encode(message); // 添加协议
n = send(sockfd, message.c_str(), message.size(), 0);
if (n > 0)
{
char buffer[1024];
int m = recv(sockfd, buffer, sizeof(buffer) - 1, 0);
if (m > 0)
{
buffer[m] = 0;
string package = buffer;
string content;
Decode(package, &content); // 去报头提取内容
Response res; // 反序列化
res.Deserialize(content);
cout << res.Result() << "[" << res.Code() << "]" << endl;
}
else
break;
}
else
break;
}
close(sockfd);
return 0;
}