实验知识点
msgpack基本用法asio异步读写asio同步读写
安装 msgpack 库，并进入 /home/shiyanlou/rpc/code5 目录。
git clone --depth 1 https:///msgpack/msgpack-c.git
cd msgpack-c
cmake -DMSGPACK_C11=ON .
sudo make install
cd ..msgpack 是一种高效的二进制序列化格式，像 json ，但是相比 json 速度快得多，体积也小得多，它支持多种语言。
本节我们将使用 msgpack 对数据进行序列化和反序列化。
序列化与反序列化
我们的 rpc 通信库将使用 msgpack 作为传输格式。在 code5 目录下新建 code1.cpp 文件。
#include <msgpack.hpp>
#include <string>
#include <iostream>
#include <sstream>
int main()
{
msgpack::type::tuple<bool, char, std::string> src(true, 'i', "shiyanlou");
std::stringstream buffer;
// 序列化
msgpack::pack(buffer, src);
std::string str(buffer.str());
// 反序列化
msgpack::object_handle oh = msgpack::unpack(str.data(), str.size());
msgpack::object deserialized = oh.get();
std::cout << deserialized << std::endl;
// 两种把 msgpack::object_handle 转化为 msgpack::type::tuple 的方法
msgpack::type::tuple<bool, char, std::string> dst;
deserialized.convert(dst);
msgpack::type::tuple<bool, char, std::string> dst2 = deserialized.as<msgpack::type::tuple<bool, char, std::string> >();
return 0;
}
编译和运行代码：在 build 目录下执行
g++ ../code1.cpp -o code1 -std=c++11 -I msgpack-c/include && ./code1
输出：
[true,105,"shiyanlou"]
code1.cpp 实现了如何序列化和反序列化一个 tuple ， msgpack::pack 将 tuple 序列化到 buffer 里，后面再通过 msgpack::unpack 来反序列化，序列化的结果放到 msgpack::unpacked 对象中，再调用 as<T> 将该对象转换为某个具体的对象。这里需要注意的是如果 as<T> 转换失败了则会抛异常，我们需要捕获异常做错误处理。
最后再来强调一下序列化反序列化过程中数据类型的变化。
序列化：msgpack::type::tuple 类型 -> std::stringstream 类型 -> std::string 类型。
反序列化：std::string 类型 -> msgpack::object_handle 类型 -> msgpack::object 类型 -> msgpack::type::tuple 类型。
序列化与反序列化自定义对象
只需要在自定义类中添加一句声明 MSGPACK_DEFINE()，msgpack 就可以序列化自定义对象。在 code5 目录下新建 code2.cpp
#include <iostream>
#include <string>
#include <sstream>
#include <msgpack.hpp>
class person {
public:
//person() :name("") { age = 0; id = 0; };
person(int id_ = 0, std::string name_ = "", int age_ = 0) :name(name_) { age = age_; id = id_; };
int id;
std::string name;
int age;
MSGPACK_DEFINE(id, name, age); // 申明这个类需要序列化
void disply() {
std::cout << id << " " << name << " " << age << std::endl;
};
};
void test() {
person src(1, "tom", 20 );
std::stringstream buffer;
msgpack::pack(buffer, src); // 将自定义类序列化
std::string str(buffer.str());
msgpack::object_handle oh = msgpack::unpack(str.data(), str.size()); // 反序列化
msgpack::object deserialized = oh.get();
try {
person dst = deserialized.as<person>(); // 得到类的实例
dst.disply(); // 调用类的方法
}
catch (...)
{
throw std::invalid_argument("Args not match!");
}
}
int main(void)
{
test();
return 0;
}
编译和运行代码：在 build 目录下执行
g++ ../code2.cpp -o code2 -std=c++11 -I msgpack-c/include && ./code2
输出：
1 tom 20
对于要序列化的自定义对象 person 我们需要定义一个 MSGPACK_DEFINE ，然后就可以调用 msgpack::pack 和 msgpack::unpack 对 person 对象进行序列化和反序列化。
需要注意的是，可以将数据反序列化为对象的前提是通信的双发都有该类的定义。如果接收方没有该类的定义，最终无法填写 deserialized.as< >() 中的模板参数，也就无法得到类的实例。
在 asio 框架下使用 msgpack 封装数据
tcp 流传输数据时，我们需要约定一个流的解析协议，一种常见的方式是包头加包体，固定长度的一个包头，这个包头里的内容就是包体的长度，我们读 tcp 流时先读固定长度的包头，然后解析出包体的长度，再把包体读完，读完之后重新读包头，如此循环。
我们可以用一个 char head[4] 来表示包头，std::vecotr<char> 来表示变长的包体。
整个流程为：
- 客户端对数据进行序列化，得到包体;
- 客户计算包体的长度并写入包头字段;
- 客户端将包从网口发出;
- 服务器收到包;
- 服务器从数据流中读取前 4 字节，这个就是包头的值，也是包的长度;
- 服务器从数据流中读取相应长度的字节，反序列化得到原数据;
- 服务器打印数据。
服务端
在 code5 目录下新建 code3.cpp
#include<string>
#include<iostream>
#include<boost/asio/io_service.hpp>
#include<boost/asio/ip/tcp.hpp>
#include<boost/bind.hpp>
#include<boost/shared_ptr.hpp>
#include<boost/enable_shared_from_this.hpp>
#include<boost/asio/streambuf.hpp>
#include<boost/asio/placeholders.hpp>
#include<boost/asio.hpp>
using boost::asio::ip::tcp;
using boost::asio::ip::address;
#include <msgpack.hpp>
#define NOTAPPLICATED -3000
#define MAXPACKSIZE 1024
class session
: public boost::enable_shared_from_this<session> {
public:
session(boost::asio::io_service &io_service) : io_service_(io_service),socket_(io_service)
{
buffer = std::make_shared<std::array<char, MAXPACKSIZE>>();
*len_ = '\0'; // 初始化成员变量
*opt_ = '\0';
}
void start() {
static tcp::no_delay option(true);
socket_.set_option(option); // 设置 socket 为无延时模式
start_chains(); // 开始 读取头部 -> 读取 msgpack 包 -> 读取头部 的循环
}
tcp::socket& socket() {
return socket_;
}
private:
void start_chains()
{
read_msgpack_len();
}
void read_msgpack_len() // 读取包的长度
{
auto self = this->shared_from_this();
auto async_buffer = buffer;
boost::asio::async_read(socket_, boost::asio::buffer(len_, 4),
[this, self, async_buffer](const boost::system::error_code& ec, std::size_t size)
{
std::cout << socket_.remote_endpoint().address() << ":" << socket_.remote_endpoint().port() << " len 数据接收完成" << std::endl;
std::cout << socket_.remote_endpoint().address() << ":" << socket_.remote_endpoint().port() << " 原始数据 " << len_ << std::endl;
len = boost::asio::detail::socket_ops::network_to_host_long(int(*(int*)len_)); // 转换为主机字节序
std::cout << socket_.remote_endpoint().address() << ":" << socket_.remote_endpoint().port() << " len " << opt << std::endl;
read_msgpack(); // 读取 msgpack 包
});
}
void read_msgpack()
{
auto self = this->shared_from_this();
auto async_buffer = buffer;
boost::asio::async_read(socket_, boost::asio::buffer(async_buffer->data(), len),
[this, self, async_buffer](const boost::system::error_code& ec, std::size_t size)
{
if (ec)
{
std::cout << ec.message() << std::endl;
return;
}
std::cout << socket_.remote_endpoint().address() << ":" << socket_.remote_endpoint().port() << " magpack 数据接收完成" << std::endl;
std::cout << socket_.remote_endpoint().address() << ":" << socket_.remote_endpoint().port() << " 原始数据 " << async_buffer->data() << std::endl;
msg = msgpack::unpack(async_buffer->data(), len); // 反序列化
send_to_client();
});
}
void send_to_client()
{
std::cout << "进入 send_to_client" << std::endl;
auto tp = msg.get().as<std::tuple<int, int> >();
std::cout << socket_.remote_endpoint().address() << ":" << socket_.remote_endpoint().port() << " magpack " << std::get<0>(tp) << " " << std::get<1>(tp) << std::endl;
int result = 23333; // 要发送的数据
auto self = this->shared_from_this();
auto async_buffer = buffer;
std::tuple<int> src(result);
std::stringstream sbuffer;
msgpack::pack(sbuffer, src); // 序列化
std::string strbuff(sbuffer.str());
memcpy(async_buffer->data(), strbuff.data(), strbuff.size());
std::cout << socket_.remote_endpoint().address() << ":" << socket_.remote_endpoint().port() << " 服务端序列化完成" << std::endl;
boost::asio::async_write(socket_, boost::asio::buffer(async_buffer->data(), async_buffer->size()),//
[this, self, async_buffer](const boost::system::error_code& ec, std::size_t size)
{
if (ec)
{
std::cout << ec.message() << std::endl;
return;
}
std::cout << socket_.remote_endpoint().address()<< ":" <<socket_.remote_endpoint().port()<<" 服务端发送成功" << std::endl;
});
}
private:
boost::asio::io_service& io_service_;
tcp::socket socket_;
boost::asio::streambuf sbuf_;
std::shared_ptr<std::array<char, MAXPACKSIZE>> buffer;
char len_[4];
int len;
char opt_[4];
int opt;
msgpack::object_handle msg;
};
typedef boost::shared_ptr<session> session_ptr;
class server {
public:
server(boost::asio::io_service& io_service, tcp::endpoint& endpoint)
: io_service_(io_service), acceptor_(io_service, endpoint)
{
session_ptr new_session(new session(io_service_)); // 指向 session 类
acceptor_.async_accept(new_session->socket(), // 异步接收连接，如果有连接就调用 handle_accept() 成员函数
boost::bind(&server::handle_accept,
this,
new_session,
boost::asio::placeholders::error));
}
void handle_accept(session_ptr new_session, const boost::system::error_code& error) {
if (error) {
return;
}
new_session->start(); // 调用 session 类的 start() 成员函数
new_session.reset(new session(io_service_));
acceptor_.async_accept(new_session->socket(), boost::bind(&server::handle_accept, this, new_session,
boost::asio::placeholders::error));
io_service_.run();
}
void run() {
io_service_.run();
}
private:
boost::asio::io_service& io_service_;
tcp::acceptor acceptor_;
};
int main(int argc, char* argv[])
{
boost::asio::io_service io_service; // 定义 io_service
tcp::endpoint endpoint(tcp::v4(), 2019); // 设置协议与端口号
server s(io_service, endpoint); // 实例化服务器，并开始运行
s.run();
return 0;
}
这是服务端的代码，包括两个类： connection 和 server 。 connection 表示一个连接， server 接受客户端连接之后就启动 connection , 让 connection 去读和写。接下来我们来分析一下 connection 类的实现。
connection 类成员包含一个 socket 句柄，通过它来读 tcp 流。它还包括一个 4 字节的 head_ ，和一个 vector<char> data_ ，正如前面所说，这个固定长度的 head 是用来获得包体长度的，读到包头之后继续读包体，包体数据就存放在 data_ 中。
接下来我们看看 read_head 函数，内部调用了 boost::asio::async_read 来读 tcp 流，当读满 head_ 之后就会进入到回调的 lambda ，注意，这里我们调用了 shared_from_this() ，它会生成 this 的 shared_ptr ，通过它可以保证安全的异步回调，这一点，前面已经讲过。
在回调的 lambda 中我们先获取包体长度 len ，然后给 data_ 做了 resize(len) ，接下来就调用 read_body 函数去读完整的包体。这里要注意的时，如果发生错误了就需要记录日志和关闭 socket 。
接着再看 read_body 函数，和 read_head 类似，也是异步回调中处理业务，在回调中我们通过 msgpack 解析读入的包体，然后把解析后的内容打印出来。
server 就比较简单，它主要就是监听端口不断地等待新的连接过来。
客户端
接下来我们写一个客户端来测试一下 asio 读写数据，由客户端向服务器发送数据，服务器收到并解析并打印客户端发送的内容。
在 code5 目录下新建 code4.cpp。
#include<boost/asio/io_service.hpp>
#include<boost/asio/ip/tcp.hpp>
#include<boost/bind.hpp>
#include<boost/shared_ptr.hpp>
#include<boost/enable_shared_from_this.hpp>
#include<boost/asio/streambuf.hpp>
#include<boost/asio/placeholders.hpp>
#include<boost/asio.hpp>
using boost::asio::ip::tcp;
using boost::asio::ip::address;
#include <msgpack.hpp>
#define NOTAPPLICATED -3000
#define MAXPACKSIZE 1024
#include<string>
#include<iostream>
class client : public boost::enable_shared_from_this<client> {
public:
client(boost::asio::io_service& io_service, tcp::endpoint& endpoint)
: io_service_(io_service), socket_(io_service), endpoint_(endpoint)
{
buffer = std::make_shared<std::array<char, MAXPACKSIZE>>();
result = 0;
}
int start(int a, int b) { // 整个流程为：传入两个整形数据，发送给服务端，服务端再返回一个数据。
boost::system::error_code ec;
socket_.connect(endpoint_, ec); // 连接服务端
if (!ec)
{
static tcp::no_delay option(true);
socket_.set_option(option); // 设置 socket 为无时延 socket
construct_rpc_data( a , b); // 构造发送给服务端的数据，包含包的长度和序列化之后的数据。整个包存储在成员变量 buffer 里面
send_recive_rpc_data(ec); // 接受来自服务端的信息，存储在成员变量 result 里面
std::cout << "send_recive_rpc_data返回值：" << result << std::endl;
return result;
}
else
{
std::cerr << boost::system::system_error(ec).what() << std::endl;
}
return NOTAPPLICATED;
}
private:
void construct_rpc_data(int a, int b) // 构造 msgpack 包，获取 msgpack 的长度，将以上两个信息写入成员变量 buffer
{
std::tuple<int, int> src(1,2);
std::stringstream sbuffer;
msgpack::pack(sbuffer, src);
std::string strbuf(sbuffer.str());
std::cout << " len " << strbuf.size() << std::endl;
size_t len_bigend = boost::asio::detail::socket_ops::host_to_network_long(strbuf.size());
memcpy(buffer->data(), &len_bigend, 4);
memcpy(buffer->data() + 4, strbuf.data(), strbuf.size());
}
void send_recive_rpc_data(const boost::system::error_code& error) // 将数据发送给服务端，并接受来自服务端的信息
{
auto self = this->shared_from_this();
auto async_buffer = buffer;
boost::asio::async_write(socket_, boost::asio::buffer(*async_buffer, MAXPACKSIZE), // 异步发送数据
[this,self, async_buffer](const boost::system::error_code& ec, std::size_t size)
{
recive_rpc_data(ec); // 接收数据
io_service_.stop();
});
io_service_.run(); // 同步等待上述事件完成
}
void recive_rpc_data(const boost::system::error_code& error) {
std::cout << "发送完毕，开始接受数据" << std::endl;
auto self = this->shared_from_this();
auto async_buffer = buffer;
boost::asio::async_read(socket_, boost::asio::buffer(*async_buffer, async_buffer->size()),
[this, self, async_buffer](const boost::system::error_code& ec, std::size_t size)
{
std::cout << "数据读取完成" << std::endl;
handle_rpc_data(ec); // 对读到的数据进行处理
io_service_.stop(); // 终止 asio 事件循环
});
io_service_.run(); // 同步等待上述时间完成
}
void handle_rpc_data(const boost::system::error_code& error) {
std::cout << "读到数据：" << buffer->data() << std::endl;
msgpack::object_handle msg = msgpack::unpack(buffer->data(), buffer->size());
auto tp = msg.get().as<std::tuple<int>>();
std::cout << " magpack " << std::get<0>(tp) << std::endl;
result = std::get<0>(tp);
}
private:
boost::asio::io_service& io_service_;
tcp::socket socket_;
tcp::endpoint& endpoint_;
std::shared_ptr<std::array<char, MAXPACKSIZE>> buffer;
int result;
};
typedef boost::shared_ptr<client> client_ptr;
int main()
{
boost::asio::io_service io_service; // 定义 io_service
tcp::endpoint endpoint(address::from_string("127.0.0.1"), 2019); // 定义远端地址
client_ptr new_session(new client(io_service, endpoint)); // 初始化 client 类，与服务器建立连接
new_session->start(1,2); // 向服务端发送数据
io_service.run();
return 0;
}
客户端主要有两个函数一个是连接 connect 函数，一个是发送消息的 send 函数。
注意这里的 connect 和 write 都是同步方式的， asio 异步网络接口都会有 asyc_ 前缀。
连接函数比较简单，传入服务端的 IP 和端口的字符串即可实现连接，如果连接失败会抛异常。
发送数据的函数调用了 boost::asio::write 函数，主要参数为 socket 和 std::vector<boost::asio::const_buffer> 消息，这里可以通过错误码判断发送是否成功。
我们测试的时候把 std::tuple<int, std::string> src(20, "hello tom") 序列化为了 msgpack ，服务器收到客户端的消息会反序列化这个消息并打印出内容。
编译客户端和服务端代码：在 build 目录下执行
g++ ../code3.cpp -o code3 -std=c++11 -I msgpack-c/include -lboost_system && g++ ../code4.cpp -o code4 -std=c++11 -I msgpack-c/include -lboost_system -lpthread
运行一下：
./code3 & ./code4
输出如下图所示：
本节实验介绍了 msgpack 的基本用法，重点介绍 msgpack 作为通信协议， asio 异步和同步读写。
序列化反序列化过程数据类型的变化如下：
序列化：msgpack::type::tuple 类型 -> std::stringstream 类型 -> std::string 类型。
反序列化：std::string 类型 -> msgpack::object_handle 类型 -> msgpack::object 类型 -> msgpack::type::tuple 类型。
最后一小节使用 msgpack 封装数据，asio 网络库搭建通信框架已经具备了 RPC 的基本特征，但是耦合度过高，客户端需要显式连接服务端，自己手动封装数据，没有达到 RPC 所期望的“远程执行代码和本地执行代码感觉一样”的特点。下一章我们将针对这几个问题进行优化，完成一个看起来是本地调用实际上是远程调用的客户服务器模型，并简单介绍开源 RPC 框架 thrift 的设计思想。
