本文主要是教大家实现一个具有基本功能的rpc框架。

技术选型

很多RPC框架的实现都是基于protobuf的，包括我们公司服务端引擎实现的rpc框架也是基于protobuf。从前做过一些简单的测试，对相同数据进行序列化，protobuf对数据的压缩率明显高于对比的msgpack、bson和json；相同的对于序列化时间上的比较，msgpack则最快。但是protobuf最让我受不了的就是需要定义固定的协议，并且使用对应的工具生成协议代码，然后才能使用，用起来需要一定的学习成本。
因此我在想，是否可以基于msgpack实现一个不需要预先定义通信协议的rpc框架呢？答案当然是肯定的。

设计原型

class rpc_service {
    void handle_data();
    void register_rpc();
    void remote_call();
};

目前rpc_service这个类只是个原型，我们提供三个接口：

• handle_data: 用于处理原生数据，将原生数据序列化为msgpack
• register_rpc: 用于Server端注册rpc函数
• remote_call: Stub端调用远程rpc函数

msgpack都提供了什么

这里我使用了msgpack c++(https://github.com/msgpack/msgpack-c)。我们可以利用msgpack将参数进行打包和解包。

// 将参数打包成字符串
std::tuple<int, bool, std::string> t(1, false, "hi");
std::stringstream buffer;
msgpack::pack(buffer, t);

buffer.seekg(0);
std::string s(buffer.str());

// 对字符串进行解包
std::tuple<int, bool, std::string> dest;
msgpack::object_handle oh = msgpack::unpack(str.data(), str.size());
oh.get().convert(t);

实现一个简单rpc，我们只需要msgpack上面介绍的功能就可以了。

实现remote_call函数

先从最简单的函数开始实现。

template<typename ... Args>
void remote_call(const std::string& method, Args&& ... args);

思路很简单，我们将method字符串和args打包后的msgpack字符串都存入一个字符串流中。这里的字符串流我们定义为rpc_stream（后续将详细讲解rpc_stream）的实现，这里先简单介绍一下：

class rpc_stream : private asio::noncopyable {
public:
    rpc_stream();
public:
    std::istream& read_stream() { return stream_; }
    std::ostream& write_stream() { return stream_; }
    streambuf_ex& buf() { return buf_; }
    void reset();

    char* c_str();

private:
    streambuf_ex buf_;
    std::iostream stream_;
};

typedef std::shared_ptr<rpc_stream> rpc_stream_ptr;

rpc_stream是我写的服务端引擎的一个副产品，基于asio::streambuf的字节流处理类。接口也比较简单易懂：

• read/write_stream: 读出/输入流
• buf: 数据的缓冲区，提供size接口获得缓冲区大小
• c_str: 将字节流中的数据转化为字符串

到目前为止，就可以动手开始写remote_call函数了。

class rpc_service {
public:
    rpc_service(): stream_ptr_(new hex_engine::rpc::rpc_stream()) {}
    template<typename ... Args>
    void remote_call(const std::string& method, Args&& ... args) {
        // 1.构造tuple
        auto t = std::make_tuple(std::forward<Args>(args)...);
        // 2.总长度
        uint32_t total_sz = 0;
        stream_ptr_->write_stream().write((char*)&total_sz, sizeof(total_sz));
        // 3.函数名称
        uint8_t method_sz = method.size();
        stream_ptr_->write_stream().write((char*)&method_sz, sizeof(method_sz));
        stream_ptr_->write_stream().write(method.c_str(), method_sz);
        // 4.写入参数
        msgpack::pack(stream_ptr_->write_stream(), t);
        total_sz = stream_ptr_->buf().size() - sizeof(total_sz);
        // 5.重新写入总长度
        stream_ptr_->write_stream().seekp(0);
        stream_ptr_->write_stream().write((char*)&total_sz, sizeof(total_sz));
    }
    
    hex_engine::rpc::rpc_stream_ptr stream_ptr() { return stream_ptr_; }
    
private:
    hex_engine::rpc::rpc_stream_ptr stream_ptr_;
};

这样数据便存在rpc_stream的字节流中了，之后可以通过c_str接口将字节流的数据导出，通过网络传输到远端。

接下来

后面一章将讲解Server端Rpc的实现