gRPC 是一个基于 HTTP2 协议设计，语言无关的通用 RPC 框架。借助服务定义，可以生成服务器端骨架（服务器代理）。同时，生成客户端存根（客户端代理）。抽象简化了底层的通信框架，客户端就像调用本地方法那样，远程调用服务接口定义的方法。
附：HTTP 发展
安装 gRPC 1.45.2 版本
安装必要的依赖工具
sudo apt-get install autoconf automake libtool
cmake 最低版本 3.15，这里安装 3.23 版本。
# 卸载原有的 cmake
sudo apt-get autoremove cmake
# 下载解压 cmake 3.23
wget https://cmake.org/files/v3.23/cmake-3.23.0-linux-x86_64.tar.gz
tar xvzf cmake-3.23.0-linux-x86_64.tar.gz
# 创建软链接
sudo mv cmake-3.23.0-linux-x86_64 /opt/cmake-3.23.0
sudo ln -sf /opt/cmake-3.23.0/bin/* /usr/bin/
# 测试
cmake -version
gcc/g++ 版本 6.3，这里安装 7.5
# 安装 gcc/g++ 7
sudo apt-get install -y software-properties-common
sudo add-apt-repository ppa:ubuntu-toolchain-r/test
sudo apt update
sudo apt install g++-7 -y
# 创建软链接
sudo update-alternatives --install /usr/bin/gcc gcc /usr/bin/gcc-7 60 \
--slave /usr/bin/g++ g++ /usr/bin/g++-7
sudo update-alternatives --config gcc
# 测试
gcc -v
g++ -v
# 下载源码
git clone https://github.com/grpc/grpc
# 选择版本 v1.45.2
git tag
git checkout v1.45.2
# 下载第三方依赖
git submodule update --init
# 编译安装: tar -jxvf grpc-v1.45.2.tar.bz2
mkdir -p cmake/build
cd cmake/build
cmake ../..
make
sudo make install
编译 third_party/protobuf 里面编译安装对应的 protobuf
cd third_party/protobuf/
./autogen.sh
./configure --prefix=/usr/local
make
sudo make install
sudo ldconfig # 使得新安装的动态库能被加载
protoc --version # 3.19.4
编译 helloworld
cd grpc/examples/cpp/helloworld/
mkdir build
cd build/
cmake ..
make登录后复制
启动服务和客户端
# 启动服务端，监听在50051端口
./greeter_server
Server listening on 0.0.0.0:50051
# 启动客户端，服务端返回Hello world
./greeter_client
Greeter received: Hello world
构建 grpc 服务首先要定义服务接口。服务就是可以被远程调用的一组方法。
grpc 使用 pb (protocol buffers) 作为 IDL（接口定义语言，interface definition language），来定义服务接口。pb 是一种语言无关、平台无关、可扩展的结构化数据序列化机制。rpc 服务接口在 .proto 文件中定义，并将 rpc 方法参数和返回类型指定为 pb 消息。可以借助 grpc 插件来根据 pb 文件生成代码。
例：
syntax = "proto3"; // 语法
package IM.Login; // 包名
// 定义服务：远程调用方法，参数 Request，返回值 Reply
// pb 规定只能有一个参数，并只能返回一个值，想传多个，定义消息类型。
service ImLogin {
rpc Regist(IMRegistReq) returns (IMRegistRes) {}
rpc Login(IMLoginReq) returns (IMLoginRes) {}
}
// 注册账号
message IMRegistReq{
string user_name = 1; // 用户名
string password = 2; // 密码
}
// 注册返回
message IMRegistRes{
string user_name = 1; // 用户名
uint32 user_id = 2; // 用户 id
uint32 result_code = 3; // 返回0，正常注册
}
// rpc 请求
message IMLoginReq{
string user_name = 1; // 用户名
string password = 2; // 密码
}
// rpc 返回
message IMLoginRes{
uint32 user_id = 1;
uint32 result_code = 2; // 返回0的时候注册注册
}
生成 C++ 代码
# 生成 simple.h 和 simple.cc 文件
protoc -I ./ --cpp_out=. IM.Login.proto
# 生成 simple.grpc.pb.h 和 simple.grpc.pb.cpp 文件，服务框架
protoc -I ./ --grpc_out=. --plugin=protoc-gen-grpc=`which grpc_cpp_plugin` IM.Login.proto
protoc --cpp_out=. --grpc_out=. --plugin=protoc-gen-grpc=/usr/local/bin/grpc_cpp_plugin IM.Login.proto
在服务端，需要实现服务定义，实现远程调用方法；并运行 grpc 服务器绑定该服务。具体来说，服务端需要做好两件事：
例：C++ 流程
#include <iostream>
#include <string>
// grpc 头文件
#include <grpcpp/ext/proto_server_reflection_plugin.h>
#include <grpcpp/grpcpp.h>
#include <grpcpp/health_check_service_interface.h>
// 自定义 proto 文件生成的.h
#include "IM.Login.pb.h"
#include "IM.Login.grpc.pb.h"
// 1、命名空间
// grcp 命名空间
using grpc::Server;
using grpc::ServerBuilder;
using grpc::ServerContext;
using grpc::Status;
// 自定义 proto 文件的命名空间
using IM::Login::ImLogin;
using IM::Login::IMRegistReq;
using IM::Login::IMRegistRes;
using IM::Login::IMLoginReq;
using IM::Login::IMLoginRes;
// 2、重写服务
// 1、定义服务端的类：继承 .grpc.pb.h 文件定义的 grpc 服务
// 2、重写 grpc 服务定义的方法
class IMLoginServiceImpl : public ImLogin::Service {
// 注册
virtual Status Regist(ServerContext* context, const IMRegistReq* request, IMRegistRes* response) override {
std::cout << "Regist user_name: " << request->user_name() << std::endl;
response->set_user_name(request->user_name());
response->set_user_id(10);
response->set_result_code(0);
return Status::OK;
}
// 登录
virtual Status Login(ServerContext* context, const IMLoginReq* request, IMLoginRes* response) override {
std::cout << "Login user_name: " << request->user_name() << std::endl;
response->set_user_id(10);
response->set_result_code(0);
return Status::OK;
}
};
// 3、启动 grpc 服务
void RunServer() {
std::string server_addr("0.0.0.0:50051");
// 创建一个服务类
IMLoginServiceImpl service;
// 创建工厂类
ServerBuilder builder;
// 监听端口地址
builder.AddListeningPort(server_addr, grpc::InsecureServerCredentials());
// 心跳探活
builder.AddChannelArgument(GRPC_ARG_KEEPALIVE_TIME_MS, 5000);
builder.AddChannelArgument(GRPC_ARG_KEEPALIVE_TIMEOUT_MS, 10000);
builder.AddChannelArgument(GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS, 1);
// 多线程：动态调整 epoll 线程数量
builder.SetSyncServerOption(ServerBuilder::MIN_POLLERS, 4);
builder.SetSyncServerOption(ServerBuilder::MAX_POLLERS, 8);
// 注册服务
builder.RegisterService(&service);
// 创建并启动 rpc 服务器
std::unique_ptr<Server> server(builder.BuildAndStart());
std::cout << "Server listening on " << server_addr << std::endl;
// 进入服务事件循环
server->Wait();
}
int main(int argc, const char** argv) {
RunServer();
return 0;
}
在客户端，由服务定义 pb 生成客户端存根 stub（客户端代理），使用通道 channel 连接特定的 grpc 服务端；stub 在 channel 基础上创建而成，通过 stub 真正调用 rpc 请求。
核心代码
class ImLoginClient {
public:
// 使用通道 channel 初始化阻塞式存根 stub
ImLoginClient(std::shared_ptr<Channel> channel)
:stub_(ImLogin::NewStub(channel))
{}
// 使用阻塞式存根调用远程方法
void Regist(const std::string &user_name, const std::string &password) {
// 调用 rpc 接口
Status status = stub_->Regist(&context, request, &response);
}
private:
std::unique_ptr<ImLogin::Stub> stub_; // 存根，客户端代理
};
例：C++ 流程
#include <iostream>
#include <memory>
#include <string>
// grpc 头文件
#include <grpcpp/grpcpp.h>
// 自定义 proto 文件生成的.h
#include "IM.Login.pb.h"
#include "IM.Login.grpc.pb.h"
// 命名空间
// grcp 命名空间
using grpc::Channel;
using grpc::ClientContext;
using grpc::Status;
// 自定义 proto 文件的命名空间
using IM::Login::ImLogin;
using IM::Login::IMRegistReq;
using IM::Login::IMRegistRes;
using IM::Login::IMLoginReq;
using IM::Login::IMLoginRes;
class ImLoginClient {
public:
ImLoginClient(std::shared_ptr<Channel> channel)
:stub_(ImLogin::NewStub(channel))
{}
void Regist(const std::string &user_name, const std::string &password) {
IMRegistReq request;
request.set_user_name(user_name);
request.set_password(password);
IMRegistRes response;
ClientContext context;
std::cout << "-> Regist req" << std::endl;
// 调用 rpc 接口
Status status = stub_->Regist(&context, request, &response);
if(status.ok()) {
std::cout << "user_name:" << response.user_name() << ", user_id:" << response.user_id() << std::endl;
}
else {
std::cout << "user_name:" << response.user_name() << "Regist failed: " << response.result_code()<< std::endl;
}
}
void Login(const std::string &user_name, const std::string &password) {
IMLoginReq request;
request.set_user_name(user_name);
request.set_password(password);
IMLoginRes response;
ClientContext context;
std::cout << "-> Login req" << std::endl;
// 调用 rpc 接口
Status status = stub_->Login(&context, request, &response);
if(status.ok()) {
std::cout << "user_id:" << response.user_id() << " login ok" << std::endl;
}
else {
std::cout << "user_name:" << request.user_name() << "Login failed: " << response.result_code()<< std::endl;
}
}
private:
std::unique_ptr<ImLogin::Stub> stub_; // 存根，客户端代理
};
int main() {
// 服务器的地址
std::string server_addr = "localhost:50051";
// 创建请求通道
ImLoginClient im_login_client(
grpc::CreateChannel(server_addr, grpc::InsecureChannelCredentials())
);
// 测试
std::string user_name = "Jim Hacker";
std::string password = "123456";
im_login_client.Regist(user_name, password);
im_login_client.Login(user_name, password);
return 0;
}
当调用 grpc 服务时，客户端的 grpc 库会使用 pb，并将 rpc 的请求编排 marshal 为 pb 格式，然后将其通过 HTTP/2 进行发送。在服务器端，请求会解排 unmarshal，对应的过程调用会使用 pb 来执行。
grpc 根据消息的数量，将通信模式分为以下四种：
以官方范例 examples/cpp/route_guide/ 为例：pb 定义的服务如下，stream 关键字来定义流
service RouteGuide {
// A simple RPC.
rpc GetFeature(Point) returns (Feature) {}
// A server-to-client streaming RPC.
rpc ListFeatures(Rectangle) returns (stream Feature) {}
// A client-to-server streaming RPC.
rpc RecordRoute(stream Point) returns (RouteSummary) {}
// A Bidirectional streaming RPC.
rpc RouteChat(stream RouteNote) returns (stream RouteNote) {}
}
服务端需要实现 pb 中定义的 rpc，每种 rpc 的实现都需要 ServerContext 参数。
其他参数则与 grpc 通信模式有关。
非流模式：Request 请求，Reply 响应。
// rpc ListFeatures(Rectangle) returns (stream Feature) {}
Status ListFeatures(ServerContext* context, const routeguide::Rectangle* rectangle, ServerWriter<Feature>* writer);
流模式：单向流
ServerReader：读 client 流，通过 Reader->Read() 返回的 bool 型状态，判断流的结束。
// rpc RecordRoute(stream Point) returns (RouteSummary) {}
Status RecordRoute(ServerContext* context, ServerReader<Point>* reader, RouteSummary* summary) {
// 读取请求
while (reader->Read(&point)) {
...
}
}
ServerWriter：写 server 流，通过结束 rpc 函数并返回状态码的方式结束流
// rpc ListFeatures(Rectangle) returns (stream Feature) {}
Status ListFeatures(ServerContext* context, const routeguide::Rectangle* rectangle, ServerWriter<Feature>* writer) {
// 发送响应
writer->Write(f);
...
}
流模式：双向流
ServerReaderWriter：只需要 1 个参数
// rpc RouteChat(stream RouteNote) returns (stream RouteNote) {}
// 注意线程同步
Status RouteChat(ServerContext* context, ServerReaderWriter<RouteNote, RouteNote>* stream) {
// 读取数据
while (stream->Read(¬e)) {
// 写回数据
stream->Write(n);
}
}
客户端均需要传入 ClientContext 参数。
其他参数则与 grpc 通信模式有关。
非流模式：Request 请求，Reply 响应。
// rpc GetFeature(Point) returns (Feature) {}
Status GetFeature(ClientContext* context, const Point& request, Feature* response);
流模式：单向流
ClientReader：读 server 流，通过 Reader->Read() 返回的 bool 型状态，判断流的结束。
// rpc ListFeatures(Rectangle) returns (stream Feature) {}
unique_ptr<ClientReader<Feature>> ListFeatures(ClientContext* context, const Rectangle& request) {
// 创建 reader，读取响应
// 参数：rpc 的 Context, Request
std::unique_ptr<ClientReader<Feature> > reader(stub_->ListFeatures(&context, rect));
// 读取响应
while (reader->Read(&feature)) {
...
}
// 等待返回状态
Status status = reader->Finish();
...
}
ClientWriter：写 client 流，流的结束
writer->WritesDone()：发送结束writer->Finish()：等待对端返回状态// rpc RecordRoute(stream Point) returns (RouteSummary) {}
void RecordRoute() {
// 创建 writer
std::unique_ptr<ClientWriter<Point> > writer(stub_->RecordRoute(&context, &stats));
// 发送请求
writer->Write(f.location()))
// 发送结束
writer->WritesDone();
// 等待返回状态
Status status = writer->Finish();
}
流模式：双向流
ClientReaderWriter：对于 rpc 调用，都是 client 请求后 server 响应，即双向流需要 client 先发送完数据，server 才能结束 rpc。流的结束
stream->WriteDone()stream->Finish()// rpc RouteChat(stream RouteNote) returns (stream RouteNote) {}
// client 需要开启发送线程和接收线程
void RouteChat() {
// 创建 readerwriter，读取写入都是它
std::shared_ptr<ClientReaderWriter<RouteNote, RouteNote> > stream(
stub_->RouteChat(&context));
// 子线程发送请求
std::thread writer([stream]() {
// 发送请求
stream->Write(note);
// 发送结束
stream->WritesDone();
});
...
// 主线程读取响应
// 读取响应
while (stream->Read(&server_note)) {
}
writer.join();
// 等待返回状态
Status status = stream->Finish();
...
}
这里，总结流的结束方式：
Writer->WritesDone() 结束流status code的方式来结束流Reader->Read() 返回的 bool 型状态，来判断流是否结束官方文档：Asynchronous-API tutorial
grpc 通过完成队列 CompletionQueue 来进行异步操作，其通用流程为：
void* Tag 唯一标识请求该 rpc 请求cq->Next()阻塞读取 cq 队列中的下个 rpc 请求异步 server 的逻辑
cq->Next()，处理事件CallData->Proceed()，处理后等待对端返回结果 responder_.Finish（类型：ServerAsyncResponseWriter）
创建 CallData 类：实现 rpc 请求的逻辑和状态。每个 rpc 请求对应一个 CallData 实例。若要实现不同类型的 rpc 请求，可以构造对应的 CallData 子类，子类继承基类 CallData 的通用部分，并实现自己的差异化部分。
例如：文章第 1 部分的案例
class ServerImpl final {
// 实现 rpc 请求的逻辑和状态
class CallData {
public:
// 创建 CallData 类，
// 1、绑定 cq 队列到 rpc 调用
CallData(ImLogin::AsyncService* service, ServerCompletionQueue* cq)
: service_(service), cq_(cq), status_(CREATE) {
Proceed(); // 业务逻辑处理
}
virtual ~CallData(){}
// 虚函数：业务逻辑接口
virtual void Proceed() {}
// 基类部分
// rpc 提供的异步服务
ImLogin::AsyncService* service_;
// 完成队列
ServerCompletionQueue* cq_;
// rpc 上下文
ServerContext ctx_;
// 状态机：描述业务逻辑处理时的状态
enum CallStatus { CREATE, PROCESS, FINISH };
// 当前 rpc 服务的状态
CallStatus status_;
};
// rpc：注册服务
class RegistCallData : public CallData {
...
// 实现注册 rpc 服务的业务逻辑过程处理
void Proceed() override {...}
// 子类成员
IMRegistReq request_;
IMRegistRes reply_;
ServerAsyncResponseWriter<IMRegistRes> responder_;
};
// rpc：登录服务
class LoginCallData : public CallData {
...
void Proceed() override {...}
IMLoginReq request_;
IMLoginRes reply_;
ServerAsyncResponseWriter<IMLoginRes> responder_;
};
...
};
以官方范例 examples/cpp/helloworld 为例，完整代码如下：
#include <iostream>
#include <memory>
#include <string>
#include <thread>
#include <grpc/support/log.h>
#include <grpcpp/grpcpp.h>
#include "examples/protos/helloworld.grpc.pb.h"
using grpc::Server;
using grpc::ServerAsyncResponseWriter;
using grpc::ServerBuilder;
using grpc::ServerCompletionQueue;
using grpc::ServerContext;
using grpc::Status;
using helloworld::Greeter;
using helloworld::HelloReply;
using helloworld::HelloRequest;
class ServerImpl final {
public:
~ServerImpl() {
server_->Shutdown();
cq_->Shutdown();
}
void Run() {
std::string server_address("0.0.0.0:50051");
// 创建工厂类
ServerBuilder builder;
// 监听端口地址，不验证
builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
// 注册服务
builder.RegisterService(&service_);
// 创建完成队列 cq：把要监听的 rpc 对象放入到队列
cq_ = builder.AddCompletionQueue();
// 启动服务
server_ = builder.BuildAndStart();
std::cout << "Server listening on " << server_address << std::endl;
// 启动服务器事件循环：处理 rpc 请求
HandleRpcs();
}
private:
// 实现 rpc 请求的逻辑和状态
class CallData {
public:
// 创建 CallData 类
// 1、绑定 cq 队列到 rpc 调用
CallData(Greeter::AsyncService* service, ServerCompletionQueue* cq)
: service_(service), cq_(cq), responder_(&ctx_), status_(CREATE) {
// 调用业务逻辑处理
Proceed();
}
// 业务逻辑过程处理函数：状态机
void Proceed() {
// 创建状态：把 CallData 实例放入 cq 队列后进入该状态
if (status_ == CREATE) {
// 该 CallData 实例状态推进到 PROCESS
status_ = PROCESS;
// 处理 rpc 请求：CallData 实例的 this 指针作为唯一标识该 rpc 请求的 tag，实现异步返回
service_->RequestSayHello(&ctx_, &request_, &responder_, cq_, cq_, this);
}
// 处理状态
else if (status_ == PROCESS) {
// 创建一个新的 calldata 实例，用于处理新的 rpc 请求
new CallData(service_, cq_);
// 业务逻辑处理
std::string prefix("Hello ");
reply_.set_message(prefix + request_.name());
// 业务逻辑处理结束
// 该 calldata 实例状态推进到 FINISH，并将会在 FINISH 状态中释放其占用的资源
status_ = FINISH;
// 2、等待对端返回状态：this 指针作为 tag 唯一标识 calldata 实例
responder_.Finish(reply_, Status::OK, this);
}
else {
GPR_ASSERT(status_ == FINISH);
// 释放 calldata 内存，即本次 rpc 请求的资源
delete this;
}
}
private:
// rpc 提供的异步服务
Greeter::AsyncService* service_;
// 完成队列
ServerCompletionQueue* cq_;
// rpc 上下文
ServerContext ctx_;
// What we get from the client.
HelloRequest request_;
// What we send back to the client.
HelloReply reply_;
// The means to get back to the client.
ServerAsyncResponseWriter<HelloReply> responder_;
// 状态机：描述业务逻辑处理时的状态
enum CallStatus { CREATE, PROCESS, FINISH };
// 当前 rpc 服务的状态
CallStatus status_;
};
// 服务器事件循环：处理 rpc 请求，可运行在多线程
void HandleRpcs() {
// 创建 calldata 类维护 rpc 请求的逻辑和状态
new CallData(&service_, cq_.get());
// 每个 calldata 请求的唯一标识，指向上面 new calldata 类的地址
void* tag;
bool ok;
while (true) {
// 3、阻塞读取 cq 队列中的下个 rpc 请求
// 通过返回值判断是否有请求到来还是 cq 队列正在关闭
GPR_ASSERT(cq_->Next(&tag, &ok));
GPR_ASSERT(ok);
// 处理业务，可以自定义 proceed
// 改进：扔给线程池去做异步处理
static_cast<CallData*>(tag)->Proceed();
}
}
// 完成队列
std::unique_ptr<ServerCompletionQueue> cq_;
// rpc 异步服务
Greeter::AsyncService service_;
// rpc 服务器
std::unique_ptr<Server> server_;
};
int main(int argc, char** argv) {
ServerImpl server;
server.Run();
return 0;
}
异步 client 的逻辑
CompletionQueue 到 rpc 请求。rpc_.Finish等待对端返回状态cq->Next() 阻塞读取 cq 队列中的下个 rpc 事件以官方范例 examples/cpp/helloworld 为例，完整代码如下
#include <iostream>
#include <memory>
#include <string>
#include <grpc/support/log.h>
#include <grpcpp/grpcpp.h>
#include "examples/protos/helloworld.grpc.pb.h"
using grpc::Channel;
using grpc::ClientAsyncResponseReader;
using grpc::ClientContext;
using grpc::CompletionQueue;
using grpc::Status;
using helloworld::Greeter;
using helloworld::HelloReply;
using helloworld::HelloRequest;
class GreeterClient {
public:
explicit GreeterClient(std::shared_ptr<Channel> channel)
: stub_(Greeter::NewStub(channel)) {}
std::string SayHello(const std::string& user) {
HelloRequest request;
request.set_name(user);
HelloReply reply;
ClientContext context;
CompletionQueue cq;
Status status;
// 1、绑定 cq 到 rpc 请求
std::unique_ptr<ClientAsyncResponseReader<HelloReply> > rpc(
stub_->PrepareAsyncSayHello(&context, request, &cq));
// 初始化 rpc 调用
rpc->StartCall();
// 2、等待对端返回状态
rpc->Finish(&reply, &status, (void*)1);
// 3、阻塞读取 cq 队列中的下个 rpc 事件
void* got_tag;
bool ok = false;
GPR_ASSERT(cq.Next(&got_tag, &ok));
GPR_ASSERT(got_tag == (void*)1);
GPR_ASSERT(ok);
if (status.ok()) {
return reply.message();
} else {
return "RPC failed";
}
}
private:
std::unique_ptr<Greeter::Stub> stub_;
};
int main(int argc, char** argv) {
GreeterClient greeter(grpc::CreateChannel( "localhost:50051", grpc::InsecureChannelCredentials()));
std::string user("world");
std::string reply = greeter.SayHello(user);
std::cout << "Greeter received: " << reply << std::endl;
return 0;
}