C++简单实现RPC网络通讯的示例详解

2023-04-28 11:51 开发作者： Thomas_Lbw

RPC是远程调用系统简称，它允许程序调用运行在另一台计算机上的过程，就像调用本地的过程一样。RPC 实现了网络编程的“过程调用”模型，让程序员可以像调用本地函数一样调用远程函数。最近在做的也是远程调用过程，所以通过重新梳理RPC来整理总结一下。

项目来源：

github - qicosmos/rest_rpc: modern C++(C++11), simple, easy to use rpc framework

一、RPC简介

1.1 简介

RPC指的是计算机A的进程调用另外一台计算机B的进程，A上的进程被挂起，B上被调用的进程开始执行，当B执行完毕后将执行结果返回给A，A的进程继续执行。调用方可以通过使用参数将信息传送给被调用方，然后通过传回的结果得到信息。这些传递的信息都是被加密过或者其他方式处理。这个过程对开发人员是透明的，因此RPC可以看作是本地过程调用的一种扩展，使被调用过程不必与调用过程位于同一物理机中。

C++简单实现RPC网络通讯的示例详解

RPC可以用于构建基于B/S模式的分布式应用程序：请求服务是一个客户端、而服务提供程序是一台服务器。和常规和本地的调用过程一样，远程过程调用是同步操作，在结果返回之前，需要暂时中止请求程序。

RPC的优点：

支持面向过程和面向线程的模型；
内部消息传递机制对用户隐藏；
基于 RPC 模式的开发可以减少代码重写；
可以在本地环境和分布式环境中运行；

1.2 本地调用和远程调用的区别

以ARM环境为例，我们拆解本地调用的过程，以下面代码为例：

int selfIncrement(int a)
{
    return a + 1;
}
int a = 10;

当执行到selfIncrement(a)时，首先把a存入寄存器R0，之后转到函数地址selfIncrement，执行函数内的指令 ADD R0,#1。跳转到函数的地址偏移量在编译时确定。

但是如果这是一个远程调用，selfIncrement函数存在于其他机器，为了实现远程调用，请求方和服务方需要提供需要解决以下问题：

1. 网络传输。

本地调用的参数存放在寄存器或栈中，在同一块内存中，可以直接访问到。远程过程调用需要借助网络来传递参数和需要调用的函数 ID。

2. 编解码

请求方需要将参数转化为字节流，服务提供方需要将字节流转化为参数。

3. 函数映射表

服务提供方的函数需要有唯一的 ID 标识，请求方通过 ID 标识告知服务提供方需要调用哪个函数。

以上三个功能即为 RPC 的基本框架所必须包含的功能。

1.3 RPC运行的流程

一次 RPC 调用的运行流程大致分为如下七步，具体如下图所示。

1.客户端调用客户端存根程序，将参数传入；

2.客户端存根程序将参数转化为标准格式，并编组进消息；

3.客户端存根程序将消息发送到传输层，传输层将消息传送至远程服务器；

4.服务器的传输层将消息传递到服务器存根程序，存根程序对阐述进行解包，并使用本地调用的机制调用所需的函数；

5.运算完成之后，将结果返回给服务器存根，存根将结果编组为消息，之后发送给传输层;

6.服务器传输层将结果消息发送给客户端传输层；

7.客户端存根对返回消息解包，并返回给调用方。

C++简单实现RPC网络通讯的示例详解

服务端存根和客户端存根可以看做是被封装起来的细节，这些细节对于开发人员来说是透明的，但是在客户端层面看到的是 “本地” 调用了 selfIncrement() 方法，在服务端层面，则需要封装、网络传输、解封装等等操作。因此 RPC 可以看作是传统本地过程调用的一种扩展，其使得被调用过程不必与调用过程位于同一物理机中。

1.4 小结

RPC 的目标是做到在远程机器上调用函数与本地调用函数一样的体验。为了达到这个目的，需要实现网络传输、序列化与反序列化、函数映射表等功能，其中网络传输可以使用socket或其他，序列化和反序列化可以使用protobuf，函数映射表可以使用std::function。

lambda与std::function内容可以看：

C++11 匿名函数lambda的使用

C++11 std::function 基础用法

lambda 表达式和 std::function 的功能是类似的，lambda 表达式可以转换为 std::function，一般情况下，更多使用 lambda 表达式，只有在需要回调函数的情况下才会使用 std::function。

二、RPC简单实现

2.1 客户端实现代码

#include <IOStream>
#include <memory>
#include <thread>
#include <functional>
#include <cstring>
 
class RPCClient
{
public:
    using RPCCallback = std::function<void(const std::string&)>;
    RPCClient(const std::string& server_address) : server_address_(server_address) {}
    ~RPCClient() {}
 
    void Call(const std::string& method, const std::string& request, RPCCallback callback)
    {
        // 序列化请求数据
        std::stringpython data = Serialize(method, request);
        // 发送请求
        SendRequest(data);
        // 开启线程接收响应
        std::thread t([this, callback]() {
            std::string response = RecvResponse();
            // 反序列化响应数据
            std::string result = Deserialize(response);
            callback(resujavascriptlt);
        });
        t.detach();
    }
 
private:
    std::string Serialize(const std::string& method, const std::string& request)
    {
        // 省略序列化实现
    }
 
    void SendRequest(const std::string& data)
    {
        // 省略网络发送实现
    }
 
    std::string RecvResponse()
    {
        // 省略网络接收实现
    }
 
    std::string Deserialize(const std::string& response)
    {
        // 省略反序列化实现
    }
 
private:
    std::string server_address_;
};
 
int main()
{
    std::shared_ptr<RPCClient> client(new RPCClient("127.0.0.1:8000"));
    client->Call("Add", "1,2", [](const std::string& result) {
        std::cout << "Result: " << result << std::endl;
    });
    return 0;
}

这段代码定义了RPCClient类来处理客户端的请求任务，用到了lambda和std::function来处理函数调用，在Call中使用多线程技术。main中使用智能指针管理Rpcclient类，并调用了客户端的Add函数。

127.0.0.1为本地地址，对开发来说需要使用本地地址自测，端口号为8000，需要选择一个空闲端口来通信。

2.2 服务端代码

下面是服务端的实现

#include <iostream>
#include <map>
#include <functional>
#include <memory>
#include <thread>
#include <mutex>
// 使用第三方库实现序列化和反序列化
#include <boost/serialization/serialization.hpp>
#include <boost/serialization/map.hpp>
using namespacandroide std;
// 定义RPC函数类型
using RPCCallback = std::function<std::string(const std::string&)>;
class RPCHandler {
public:
    void registerCallback(const std::string& name, RPCCallback callback) {
        std::unique_lock<std::mutex> lock(mtx_);
        callbacks_[name] = callback;
    }
    std::string handleRequest(const std::string& request) {
        // 反序列化请求
        std::map<std::string, std::string> requestMap;
        std::istringstream is(request);
        boost::archive::text_iarchive ia(is);
        ia >> requestMap;
        // 查找并调用对应的回调函数
        std::string name = requestMap["name"];
        std::string args = requestMap["args"];
        std::unique_lock<std::mutex> lock(mtx_);
        auto it = callbacks_.find(name);
        if (it == callbacks_.end()) {
            return "Error: Unknown function";
        }
        RPCCallback callback = it->second;
        return callback(args);
    }
private:
    std::map<std::string, RPCCallback> callbacks_;
    std::mutex mtx_;
};
int main() {
    RPCHandler rpcHandler;
    // 注册回调函数
    rpcHandler.registerCallback("add", [](const std::string& args) {
        std::istringstream is(args);
        int a, b;
        is >> a >> b;
        int result = a + b;
        std::ostringstream os;
        os << result;
        return os.str();
    });
    rpcHandler.registerCallback("sub", [](const std::string& args) {
        std::istringstream is(args);
        int a, b;
        is >> a >> b;
        int result = a - b;
        std::ostringstream os;
        os << result;
        return os.str
    });
    // 创建处理请求的线程
    std::thread requestThread([&]() {
        while (true) {
            std::string request;
            std::cin >> request;
            std::string response = rpcHandler.handleRequest(request);
            std::cout << response << std::endl;
        }
    });
    requestThread.join();
    return 0;
}

上面的代码实现了一个简单的C++ RPC服务端。主要实现了以下功能：

1.定义了RPC函数类型 RPCCallback，使用std::function<std::string(const std::string&)>表示。

2.RPCHandler类实现了注册函数和处理请求的功能。

3.在main函数中创建了一个RPCHandler对象，并注册了两个函数"add" 和 "sub"。这些函数通过lambda表达式实现，并在被调用时通过std::istringstream读取参数并返回结果。

4.创建了一个新线程requestThread来处理请求。在这个线程中，通过std::cin读取请求，然后调用RPCHandler的handleRequest函数并使用std::cout输出响应。

C++简单实现RPC网络通讯的示例详解

注意，这套代码是最简单的RPC机制，只能调用本地的资源，他还存在以下缺点：

1.代码并没有处理错误处理，如果请求格式不正确或函数不存在，服务端将会返回“Error: Unknown function”。

2.没有使用网络库进行通信，所以只能在本机上使用。

3.没有提供高效的并发性能，所有请求都在单独的线程中处理。

4.没有考虑RPC服务的可用性和高可用性，如果服务端崩溃或不可用，客户端将无法继续使用服务。

5.没有考虑RPC服务的可扩展性，如果有大量请求需要处理，可能会导致性能问题。

6.使用了第三方库Boost.Serialization来实现序列化和反序列化，如果不想使用第三方库，可能需要自己实现序列化的功能。

下面我们一步一步完善它。

三、加强版RPC（以“RPC简单实现”为基础）

3.1 加入错误处理

下面是 RPCHandler 类中加入错误处理的代码示例:

class RPCHandler {
public:
    // 其他代码...
    std::string handleRequest(const std::string& request) {
        // 反序列化请求
        std::map<std::string, std::string> requestMap;
        std::istringstream is(request);
        boost::archive::text_iarchive ia(is);
        ia >> requestMap;
        // 查找并调用对应的回调函数
        std::string name = requestMap["name"];
        std::string args = requestMap["args"];
        std::unique_lock<std::mutex> lock(mtx_);
        auto it = callbacks_.find(name);
        if (it == callbacks_.end()) {
            return "Error: Unknown function";
        }
        RPCCallback callback = it->second;
        try {
            return callback(args);
        } catch (const std::exception& e) {
            return "Error: Exception occurred: " + std::string(e.what());
        } catch (...) {
            return "Error: Unknown exception occurred";
        }
    }
};

上面的代码在 RPCHandler 类的 handleRequest 函数中加入了错误处理的代码，它使用了 try-catch 语句来捕获可能发生的异常。如果找不到对应的函数或发生了异常，会返回错误信息。这样，如果请求格式不正确或函数不存在，服务端将会返回相应的错误信息。

3.2 加入网络连接（socket）

加入网络连接不需要动服务端的实现，只需要在main里创造套接字去链接就好：

int main() 
{
    io_context ioc;
    ip::tcp::acceptor acceptor(ioc, ip::tcp::endpoint(ip::tcp::v4(), 8080));
    RPCHandler rpcHandler;
    // 注册函数
    rpcHandler.registerCallback("add", [](const std::string& args) {
        std::istringstream is(args);
        int a, b;
        is >> a >> b;
        int result = a + b;
        std::ostringstream os;
        os << result;
        return os.str();
    });
    rpcHandler.registerCallback("sub", [](const std::string& args) {
        std::istjsringstream is(args);
        int a, b;
        is >> a >> b;
        int result = a - b;
        std::ostringstream os;
        os << result;
        return os.str();
    });
    // 等待连接
    while (true) {
        ip::tcp::socket socket(ioc);
        acceptor.accept(socket);
        // 创建线程处理请求
        std::thread requestThread([&](ip::tcp::socket socket) {
            while (true) {
                // 读取请求
                boost::asio::streambuf buf;
                read_until(socket, buf, '\n');
                std::string request = boost::asio::bjsuffer_cast<const char*>(buf.data());
                request.pop_back();
                // 处理请求
                std::string response = rpcHandler.handleRequest(request);
                // 发送响应
                write(socket, buffer(response + '\n'));
            }
        }, std::move(socket));
        requestThread.detach();
    }
    return 0;
}

这是一个使用Boost.Asio库实现的RPC服务端代码示例。它使用了TCP协议监听8080端口，等待客户端的连接。当有客户端连接时，创建一个新线程来处理请求。请求和响应通过网络传输。

3.3 加强并发性

使用并发和异步机制，忽略重复代码，实现如下：

class RPCHandler {
public:
    // ...
    void handleConnection(ip::tcp::socket socket) {
        while (true) {
            // 读取请求
            boost::asio::streambuf buf;
            read_until(socket, buf, '\n');
            std::string request = boost::asio::buffer_cast<const char*>(buf.data());
            request.pop_back();
            // 使用并行执行处理请求
            std::vector<std::future<std::string>> futures;
            for (int i = 0; i < request.size(); i++) {
                futures.emplace_back(std::async(std::launch::async, &RPCHandler::handleRequest, this, request[i]));
            }
            // 等待所有请求处理完成并发送响应
            for (auto& f : futures) {
                std::string response = f.get();
                write(socket, buffer(response + '\n'));
            }
        }
    }
};

这样，请求会被分成多个部分并行处理，可以利用多核 CPU 的优势提高服务端的并发性能。

main（）：

int main() {
    io_context ioc;
    ip::tcp::acceptor acceptor(ioc, ip::tcp::endpoint(ip::tcp::v4(), 8080));
    RPCHandler rpcHandler;
    // 注册函数
    rpcHandler.registerCallback("add", [](const std::string& args) {
        std::istringstream is(args);
        int a, b;
        is >> a >> b;
        int result 开发者_JAVA开发= a + b;
        std::ostringstream os;
        os << result;
        return os.str();
    });
    rpcHandler.registerCallback("sub", [](const std::string& args) {
        std::istringstream is(args);
        int a, b;
        is >> a >> b;
        int result = a - b;
        std::ostringstream os;
        os << result;
        return os.str();
    });
    // 创建线程池
    boost::thread_pool::executor pool(10);
    // 等待连接
    while (true) {
        ip::tcp::socket socket(ioc);
        acceptor.accept(socket);
        // 将请求添加到线程池中处理
        pool.submit(boost::bind(&RPCHandler::handleConnection, &rpcHandler, std::move(socket)));
    }
    return 0;
}

在 main 函数中可以使用 boost::thread_pool::executor 来管理线程池，在线程池中提交任务来处理请求。这里的线程池大小设置为10，可以根据实际情况调整。

3.4 加入容错机制（修改客户端部分）

在其中使用了重试机制来保证客户端能够重新连接服务端：

class RPCClient {
public:
    RPCClient(const std::string& address, int port) : address_(address), port_(port), socket_(io_context_) {
        connect();
    }
    std::string call(const std::string& name, const std::string& args) {
        // 序列化请求
        std::ostringstream os;
        boost::archive::text_oarchive oa(os);
        std::map<std::string, std::string> request;
        request["name"] = name;
        request["args"] = args;
        oa << request;
        std::string requestStr = os.str();
        // 发送请求
        write(socket_, buffer(requestStr + '\n'));
        // 读取响应
        boost::asio::streambuf buf;
        read_until(socket_, buf, '\n');
        std::string response = boost::asio::buffer_cast<const char*>(buf.data());
        response.pop_back();
        return response;
    }
private:
    void connect() {
        bool connected = false;
        while (!connected) {
            try {
                socket_.connect(ip::tcp::endpoint(ip::address::from_string(address_), port_));
                connected = true;
            } catch (const std::exception& e) {
                std::cerr << "Error connecting to server: " << e.what() << std::endl;
                std::this_thread::sleep_for(std::chrono::seconds(1));
            }
        }
    }
    std::string address_;
    int port_;
    io_context io_context_;
    ip::tcp::socket socket_;
};

在这个示例中，当连接服务端失败时，客户端会在一定的时间间隔后重试连接，直到成功连接上服务端为止。

以上就是C++简单实现RPC网络通讯的示例详解的详细内容，更多关于C++实现RPC网络通讯的资料请关注我们其它相关文章！

继续阅读：C++ RPC网络通讯 C++ 网络通讯 C++实现RPC网络通讯

C++简单实现RPC网络通讯的示例详解

目录

一、RPC简介

1.1 简介

1.2 本地调用和远程调用的区别

1.3 RPC运行的流程

1.4 小结

二、RPC简单实现

2.1 客户端实现代码

2.2 服务端代码

三、加强版RPC（以“RPC简单实现”为基础）

3.1 加入错误处理

3.2 加入网络连接（socket）

3.3 加强并发性

3.4 加入容错机制（修改客户端部分）

更多精彩内容

精彩评论

最新开发

SpringBoot整合Liquibase实现对数据库管理和迁移

java8升级java21方法步骤(完整pom文件升级及代码修改)

javacv ffmpeg使用原生ffmpeg命令方式

golang进行xml文件解析的操作方法

如何使用go实现创建WebSocket服务器

开发排行榜

springboot后端存储富文本内容的思路与步骤(含图片内容)

PyCharm运行python测试,报错“没有发现测试”/“空套件”的解决

return base64.b64encode(b).decode(

基于C语言实现钻石棋游戏的示例代码

Sublime Text 3解决中文乱码问题（实测可用）