手写一个 RPC 远程调用(C++)

版权声明

个人作品，禁止转载

参考资料

Build Remote Procedure Calls (RPC) - from scratch in C(https://www.udemy.com/course/linuxrpc/)
Linux manual page(https://man7.org/linux/man-pages/)
Cpp reference(https://en.cppreference.com/w/)

代码下载

https://github.com/541380000/write_rpc_from_scratch.git
或者从网盘下载
链接：https://pan.baidu.com/s/1BjTK3aH_T80umXsF9Yzqmg?pwd=5mof
提取码：5mof

Overview

这是一个学习型项目，旨在学习 RPC 如何运行
这是一个 C++项目，只是用 C++标准库，不使用第三方库
各个步骤纯手写，代码无法用于商业项目
本项目不考虑机器大小端

预备知识

基础的 c/c++知识
基础的 socket 编程
基础的 debug 能力

RPC 的步骤

RPC 客户端，序列化 RPC 头和 RPC 参数
RPC 客户端，将 RPC 头和参数打包成数据包，通过网络（等其他通信手段）发送
RPC 服务器，反序列化 RPC 头和 RPC 参数
RPC 服务器，执行远程调用
RPC 服务器，将 RPC 头和返回值打包成数据包，发送给客户端
RPC 客户端，反序列化 RPC 头和返回值，得到 RPC 结果

序列化和反序列化

基础数据类型:SerBuffer

结构体定义

SerBuffer是数据序列化和反序列化的缓冲区
```
using byte=char;
struct SerBuffer{vector<byte> buffer;
};
```

方法

创建新的 SerBuffer，预留若干字节的空间

unique_ptr<SerBuffer> creatSerBuffer(const uint32_t buffer_size=DEFAULT_INIT_SERIALIZE_BUFFER_SIZE){auto ptr = make_unique<SerBuffer>();ptr->buffer.reserve(buffer_size);return move(ptr);
}

将 nBytes 的数据复制到 buffer 中

void serializeData(const unique_ptr<SerBuffer>& buffer, const byte* data, const uint32_t nBytes){copy(data, data+nBytes, back_inserter(buffer->buffer));
}

跳过 n 字节，nBytes 为负则从缓冲区删除数据，为正则追加 nBytes 个 0

void serializeBufferSkip(const unique_ptr<SerBuffer>& buffer, int32_t nBytes){if (nBytes < 0)while(nBytes < 0) buffer->buffer.pop_back(), nBytes ++;else if (nBytes > 0)while(nBytes > 0) buffer->buffer.emplace_back('0'), nBytes --;
}

清空缓冲区

void resetSerBuffer(const unique_ptr<SerBuffer>& buffer){buffer->buffer.clear();
}

填充一个空指针，空指针用 0xFFFFFFFF 表示

void fillNull(const unique_ptr<SerBuffer>& buffer){uint32_t sentinel = 0xFFFFFFFF;serializeData(buffer, (byte*)&sentinel, 4);
}

将 RPC 头序列化到 buffer 中(RPC 头定义见下文)

void serializeRPCHeader(const unique_ptr<SerBuffer>& buffer, const RPCHeader* const header){serializeData(buffer, (char*)header, sizeof(header));
}

简单对象(反)序列化

简单对象定义为：只包含基础数据类型的结构体（不包含指针和其他结构体）

对象定义

using byte=char;
struct SimplePerson{byte name[30];uint32_t age;uint32_t weight;
};

序列化方法

将数据拷贝到内存中，不考虑内存对齐。
对于 SimplePerson，内存中应该包含：30 字节的 name，4 字节的 age，4 字节的 weight，共 38 字节
如果传入参数为 NULL，则填充空指针，表示为 0xFFFFFFFF，否则填充数据

void serialize_simple_person(const SimplePerson* const obj, const unique_ptr<SerBuffer>& buffer){if(!obj){fillNull(buffer);return ;}serializeData(buffer, obj->name, sizeof(obj->name));serializeData(buffer, (char*)&obj->age, sizeof(obj->age));serializeData(buffer, (char*)&obj->weight, sizeof(obj->weight));
}

反序列化方法

相反的过程，将数据提取出来即可
offset 为：从 buffer 的第几个字节开始，反序列化出来一个对象


unique_ptr<SimplePerson> deserialize_simple_person(const unique_ptr<SerBuffer>& buffer, uint32_t offset = 0){if (*(uint32_t*)buffer->buffer.data() == 0xFFFFFFFF)return {nullptr};auto ptr = make_unique<SimplePerson>();auto& person = *ptr;copy(buffer->buffer.begin() + offset, buffer->buffer.begin() + sizeof(person) + offset, person.name);offset += sizeof(person.name);person.age = *(decltype(person.age)*)(buffer->buffer.data() + offset);offset += sizeof(person.age);person.weight = *(decltype(person.weight)*)(buffer->buffer.data() + offset);offset += sizeof(person.weight);return move(ptr);
}

测试

void testSimplePerson(){SimplePerson p{"1234", 1, 1};auto buf = make_unique<SerBuffer>();printf("-------------------> SimplePerson\n");serialize_simple_person(&p, buf);printf("Full size of person after serialize: %zu, expected is %d\n", buf->buffer.size(), 30+4+4);cout << "Before  serialize: " << p << "\nAfter deserialize: " << *deserialize_simple_person(buf) << endl << endl;
}

嵌套对象(反)序列化

简单对象定义为：只包含基础数据类型和其他结构体的结构体（不包含指针）

对象定义

struct Occupation{char title[30];uint32_t id;
}
struct NestedPerson{char name[30];Occupation occupation;uint32_t age;uint32_t weight;
}

Occupation 的(反)序列化方法（简单对象序列化）

void serialize_simple_occupation(const Occupation* const obj, const unique_ptr<SerBuffer>& buffer){serializeData(buffer, obj->title, sizeof(obj->title));serializeData(buffer, (char*)&obj->id, sizeof(obj->id));
}unique_ptr<Occupation> deserialize_simple_occupation(const unique_ptr<SerBuffer>& buffer, uint32_t offset = 0){auto ptr = make_unique<Occupation>();auto& occupation = *ptr;copy(buffer->buffer.begin() + offset, buffer->buffer.begin() + offset + sizeof(occupation.title), occupation.title);offset += sizeof(occupation.title);occupation.id = *(decltype(occupation.id)*)(buffer->buffer.data() + offset);return move(ptr);
}

序列化与反序列化方法

如果遇到嵌套结构体，则调用嵌套结构体的序列化方法，将内层结构体序列化到 buffer 中

void serialize_nested_person(const NestedPerson* const obj, const unique_ptr<SerBuffer>& buffer){if (!obj){fillNull(buffer);return ;}serializeData(buffer, obj->name, sizeof(obj->name));serialize_simple_occupation(&obj->occupation, buffer);serializeData(buffer, (char*)&obj->age, sizeof(obj->age));serializeData(buffer, (char*)&obj->weight, sizeof(obj->weight));
}unique_ptr<NestedPerson> deserialize_nested_person(const unique_ptr<SerBuffer>& buffer, uint32_t offset = 0){if (*(uint32_t*)buffer->buffer.data() == 0xFFFFFFFF)return {nullptr};auto ptr = make_unique<NestedPerson>();auto& person = *ptr;copy(buffer->buffer.begin() + offset, buffer->buffer.begin() + sizeof(person) + offset, person.name);offset += sizeof(person.name);person.occupation = *deserialize_simple_occupation(buffer, offset);offset += sizeof(person.occupation.title) + sizeof(person.occupation.id);person.age = *(decltype(person.age)*)(buffer->buffer.data() + offset);offset += sizeof(person.age);person.weight = *(decltype(person.weight)*)(buffer->buffer.data() + offset);offset += sizeof(person.weight);return move(ptr);
}

测试

void testNestedPerson(){NestedPerson p{"1234", {"doctor", 294}, 1, 1};auto buf = make_unique<SerBuffer>();printf("-------------------> NestedPerson\n");serialize_nested_person(&p, buf);printf("Full size of nested person after serialize: %zu, expected is %d\n", buf->buffer.size(), 30+4+4+30+4);cout << "Before  serialize: " << p << "\nAfter deserialize: " << *deserialize_nested_person(buf) << endl << endl;
}

含有指针的结构体的(反)序列化

结构体定义

struct PointerPerson{char name[30];Occupation* occupation;uint32_t age;uint32_t weight;
}

序列化与反序列化方法

如果有指针

指针为空，则填充 0xFFFFFFFF(如你所见，这会导致问题)
指针不为空，调用内层对象的序列化方法，将内层对象序列化到 buffer 中

void serialize_pointer_person(const PointerPerson* const obj, const unique_ptr<SerBuffer>& buffer){if (!obj){fillNull(buffer);return ;}serializeData(buffer, obj->name, sizeof(obj->name));serialize_simple_occupation(obj->occupation, buffer);serializeData(buffer, (char*)&obj->age, sizeof(obj->age));serializeData(buffer, (char*)&obj->weight, sizeof(obj->weight));
}unique_ptr<PointerPerson> deserialize_pointer_person(const unique_ptr<SerBuffer>& buffer, uint32_t offset = 0){if (*(uint32_t*)buffer->buffer.data() == 0xFFFFFFFF)return {nullptr};auto ptr = make_unique<PointerPerson>();auto& person = *ptr;copy(buffer->buffer.begin() + offset, buffer->buffer.begin() + sizeof(person) + offset, person.name);offset += sizeof(person.name);person.occupation = deserialize_simple_occupation(buffer, offset).release();offset += sizeof(person.occupation->title) + sizeof(person.occupation->id);person.age = *(decltype(person.age)*)(buffer->buffer.data() + offset);offset += sizeof(person.age);person.weight = *(decltype(person.weight)*)(buffer->buffer.data() + offset);offset += sizeof(person.weight);return move(ptr);
}

测试

void testPointerPerson(){Occupation occ{"doctor", 294};PointerPerson p{"1234", &occ, 1, 1};auto buf = make_unique<SerBuffer>();printf("-------------------> PointerPerson\n");serialize_pointer_person(&p, buf);printf("Full size of pointer person after serialize: %zu, expected is %d\n", buf->buffer.size(), 30+4+4+30+4);cout << "Before  serialize: " << p << "\nAfter deserialize: " << *deserialize_pointer_person(buf) << endl << endl;
}

还有其他很多数据结构需要序列化，但是不应该这样手动实现

RPC 头

RPC 头应该包含如下信息

远程调用 id，这里用 uint32_t 表示
payload 大小，即参数大小，用 uint32_t 表示
定义为：

struct RPCHeader{uint32_t id;    // id of procedureuint32_t size;  // size of payload(argument)
};

RPC 服务器实现

下方的 id 为 1 的 RPC，实现如下功能：接受两个 SimplePerson 对象，给他们办结婚证
，返回值的 name 字段为 name1 -Merry- name2，age 最大值，weight 取最小值

// socket相关代码省略
// 从socket_fd读取指定字节的数据，存入buffer
int read_n_bytes(int connfd, char* buffer, int n_bytes){int tmp = 0;int cnt = 0;while (cnt < n_bytes) {if ((tmp = read(connfd, buffer + cnt, n_bytes - cnt)) == -1) {cout << "failed reading" << endl;return -1;}cnt += tmp;}return 0;
}
void handle_rpc(int connfd)
{char buff[BUFFER_SIZE];RPCHeader header{};for (;;) {bzero(buff, BUFFER_SIZE);// 读取RPC头if (read_n_bytes(connfd, buff, sizeof(RPCHeader))){printf("Error reading nbytes header\n");exit(-1);}memcpy(&header, buff, sizeof(RPCHeader));printf("Get header: id: %d\t\tpayload: %d\n", header.id, header.size);// 根据RPC头中指示的payload大小，读取参数bzero(buff, BUFFER_SIZE);if (read_n_bytes(connfd, buff, header.size)) {printf("Error reading nbytes payload\n");exit(-1);}auto buffer = make_unique<SerBuffer>();buffer->buffer.assign(buff, buff + header.size);cout << "Get header: " << header << endl;// 根据不同的远程调用id，执行远程调用switch (header.id) {// get merry of two personcase 1:{// 解析两个参数auto p1 = deserialize_simple_person(buffer);auto p2 = deserialize_simple_person(buffer, header.size / 2);// 产生新的对象auto res = string(p1->name) + string(" -Merry- ") + string(p2->name);auto resp_buf = make_unique<SerBuffer>();SimplePerson p{};memcpy(p.name, res.c_str(), res.length());p.age = max(p1->age, p2->age);p.weight = min(p1->weight, p2->weight);// 产生RPC头，id为1的RPC请求，其返回的id为100+1=101RPCHeader resp_header{100+header.id, sizeof(p.weight) + sizeof(p.name) + sizeof(p.age)};// 序列化返回值并发送serializeRPCHeader(resp_buf, &resp_header);serialize_simple_person(&p, resp_buf);// send it backif (send(connfd, resp_buf->buffer.data(), resp_buf->buffer.size(), 0) == -1){return ;}break;}case 2:{// .... do something}default:{printf("Invalid rpc id: %d\n", header.id);break;}}}
}

RPC 客户端实现

如下代码将调用 id 为 1 的 RPC，并接收结果

// socket相关代码省略
void func(int sockfd)
{char buff[BUFFER_SIZE];int n;bzero(buff, sizeof(buff));/*** to run rpc, we will follow the steps:* 1. serialize RPC header to a buffer* 2. serialize RPC argument to a buffer* 3. send RPC package through socket* 4. receive the response* 5. deserialize RPC package*/// 1. serialize RPC header to a bufferRPCHeader rpcHeader{};rpcHeader.id = 1;// TODO: size will be assigned after argument serialize// 2. serialize RPC argument to a bufferSimplePerson p1{"Tony", 23, 160};SimplePerson p2{"Jenny", 21, 100};auto argumentBuffer = make_unique<SerBuffer>();serialize_simple_person(&p1, argumentBuffer);serialize_simple_person(&p2, argumentBuffer);rpcHeader.size = argumentBuffer->buffer.size();// combine 1 & 2auto buffer = make_unique<SerBuffer>();serializeRPCHeader(buffer, &rpcHeader);copy(argumentBuffer->buffer.begin(), argumentBuffer->buffer.end(), back_inserter(buffer->buffer));cout << "send request -> p1: " << p1 << "\t\tp2: " << p2 << endl;// cout << "buf sent: " << buffer->buffer.size() << endl;send(sockfd, buffer->buffer.data(), buffer->buffer.size(), 0);rpcHeader = {};read_n_bytes(sockfd, (char*)&rpcHeader, sizeof(rpcHeader));// printf("From Server : %s", buff);read_n_bytes(sockfd, buff, rpcHeader.size);resetSerBuffer(argumentBuffer);copy(buff, buff+rpcHeader.size, back_inserter(argumentBuffer->buffer));auto resp = deserialize_simple_person(argumentBuffer, 0);cout << "Get response -> " << *resp << endl;
}

手写一个 RPC 远程调用(C++)相关推荐

【RPC框架、RPC框架必会的基本知识、手写一个RPC框架案例、优秀的RPC框架Dubbo、Dubbo和SpringCloud框架比较】
一.RPC框架必会的基本知识 1.1 什么是RPC? RPC(Remote Procedure Call --远程过程调用),它是一种通过网络从远程计算机程序上请求服务,而不需要了解底层网络的技术. ...
自己手写一个RPC，实现远程调用功能（基于netty、反射和代理）
emmm,昨天蘑菇街一面,我感觉面试官特别好,最后的时候给了我一些建议和方向,感觉启发很大.面试过程中,问了我几个相对开放的问题,没怎么问基础.但是我感觉我答的不很好,第一次面大公司有点紧张.希望过过 ...
手写一个RPC框架，理解更透彻（附源码）
作者:烟味i www.cnblogs.com/2YSP/p/13545217.html 一.前言前段时间看到一篇不错的文章<看了这篇你就会手写RPC框架了>,于是便来了兴趣对着实现了一遍 ...
面试官让我手写一个RPC框架
如今,分布式系统大行其道,RPC 有着举足轻重的地位.Dubbo.Thrift.gRpc 等框架各领风骚,学习RPC是新手也是老鸟的必修课.本文带你手撸一个rpc-spring-starter,深入学 ...
如何手写一个RPC(面试要知道)
文章目录一.RPC 到底是什么? 二.场景的模拟三.rpc-core项目四.谁应该实现接口? 五.室友室友这个类来计算答案六.对自己的改造(关键的一步) 七.使用注册中心来解决室友端口变化的问 ...
【手写一个RPC框架】simpleRPC-04
目录前言实现项目创建配置依赖 common service server client 文件结构运行本项目所有代码可见:https://github.com/weiyu-zeng/Simp ...
【手写一个RPC框架】simpleRPC-05
目录前言实现项目创建依赖配置 common service codec client server 文件结构运行本项目所有代码可见:https://github.com/weiyu-zen ...
手写一个简易版本的RPC
前言在1024程序员节前夕,学习技术是对节日最好的庆祝. 手写一个简易版的RPC,可以把模糊抽象的概念具像化,属于落地层面了. 1. RPC基本原理 RPC原理 2. 四个版本的迭代注:api表示 ...
徒手撸框架--实现 RPC 远程调用
微服务,已经是每个互联网开发者必须掌握的一项技术.而 RPC 框架,是构成微服务最重要的组成部分之一.趁最近有时间.又看了看 dubbo 的源码.dubbo 为了做到灵活和解耦,使用了大量的设计模式和 ...

手写一个 RPC 远程调用(C++)

手写一个 RPC 远程调用(C++)

版权声明

参考资料

代码下载

推荐进阶工具

Overview

预备知识

RPC 的步骤

序列化和反序列化

基础数据类型:SerBuffer

结构体定义

方法

创建新的 SerBuffer，预留若干字节的空间

将 nBytes 的数据复制到 buffer 中

跳过 n 字节，nBytes 为负则从缓冲区删除数据，为正则追加 nBytes 个 0

清空缓冲区

填充一个空指针，空指针用 0xFFFFFFFF 表示

将 RPC 头序列化到 buffer 中(RPC 头定义见下文)

简单对象(反)序列化

对象定义

序列化方法

反序列化方法

测试

嵌套对象(反)序列化

对象定义

Occupation 的(反)序列化方法（简单对象序列化）

序列化与反序列化方法

测试

含有指针的结构体的(反)序列化

结构体定义

序列化与反序列化方法

测试

还有其他很多数据结构需要序列化，但是不应该这样手动实现

RPC 头

RPC 服务器实现

RPC 客户端实现

手写一个 RPC 远程调用(C++)相关推荐

最新文章

热门文章