我正在开发基于 gRPC 的服务，该服务需要高吞吐量。但目前我的程序在使用 C++ 同步 gRPC 时吞吐量较低。
我已经阅读了 gRPC 文档，但没有找到关于同步/异步 API 之间区别的明确解释。除了 async 可以控制完成队列，而它对同步 API 是透明的。
我想知道同步gRPC是否向TCP层发送消息，并等待它的“ack”，这样下一条消息会被阻塞吗？ 同时，异步 API 会异步发送它们而无需等待后面的消息？
【问题讨论】：
TLDR：是的，异步 API 将异步发送消息而无需等待后面的消息，而同步 API 将在发送/接收一条消息时阻塞整个线程。
gRPC 使用CompletionQueue 进行异步操作。你可以在这里找到官方教程：https://grpc.io/docs/languages/cpp/async/
CompletionQueue 是一个事件队列。这里的“事件”可以是请求数据接收完成，也可以是闹钟（定时器）到期等（基本上是任何异步操作的完成）
以the official gRPC asynchronous APIs example为例，关注CallData类和HandleRpcs()：
void HandleRpcs() {
// Spawn a new CallData instance to serve new clients.
new CallData(&service_, cq_.get());
void* tag; // uniquely identifies a request.
bool ok;
while (true) {
// Block waiting to read the next event from the completion queue. The
// event is uniquely identified by its tag, which in this case is the
// memory address of a CallData instance.
// The return value of Next should always be checked. This return value
// tells us whether there is any kind of event or cq_ is shutting down.
GPR_ASSERT(cq_->Next(&tag, &ok));
GPR_ASSERT(ok);
static_cast<CallData*>(tag)->Proceed();
}
}
HandleRpcs() 是服务器的主循环。这是一个无限循环，通过cq->Next()不断从完成队列中获取下一个事件，并调用它的Proceed()方法（我们自定义的处理不同状态客户端请求的方法）。
CallData 类（其实例代表一个客户端请求的完整处理周期）：
class CallData {
public:
// Take in the "service" instance (in this case representing an asynchronous
// server) and the completion queue "cq" used for asynchronous communication
// with the gRPC runtime.
CallData(Greeter::AsyncService* service, ServerCompletionQueue* cq)
: service_(service), cq_(cq), responder_(&ctx_), status_(CREATE) {
// Invoke the serving logic right away.
Proceed();
}
void Proceed() {
if (status_ == CREATE) {
// Make this instance progress to the PROCESS state.
status_ = PROCESS;
// As part of the initial CREATE state, we *request* that the system
// start processing SayHello requests. In this request, "this" acts are
// the tag uniquely identifying the request (so that different CallData
// instances can serve different requests concurrently), in this case
// the memory address of this CallData instance.
service_->RequestSayHello(&ctx_, &request_, &responder_, cq_, cq_,
this);
} else if (status_ == PROCESS) {
// Spawn a new CallData instance to serve new clients while we process
// the one for this CallData. The instance will deallocate itself as
// part of its FINISH state.
new CallData(service_, cq_);
// The actual processing.
std::string prefix("Hello ");
reply_.set_message(prefix + request_.name());
// And we are done! Let the gRPC runtime know we've finished, using the
// memory address of this instance as the uniquely identifying tag for
// the event.
status_ = FINISH;
responder_.Finish(reply_, Status::OK, this);
} else {
GPR_ASSERT(status_ == FINISH);
// Once in the FINISH state, deallocate ourselves (CallData).
delete this;
}
}
private:
// The means of communication with the gRPC runtime for an asynchronous
// server.
Greeter::AsyncService* service_;
// The producer-consumer queue where for asynchronous server notifications.
ServerCompletionQueue* cq_;
// Context for the rpc, allowing to tweak aspects of it such as the use
// of compression, authentication, as well as to send metadata back to the
// client.
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_;
// Let's implement a tiny state machine with the following states.
enum CallStatus { CREATE, PROCESS, FINISH };
CallStatus status_; // The current serving state.
};
如我们所见，CallData 具有三种状态：CREATE、PROCESS 和 FINISH。
service_->RequestSayHello(&ctx_, &request_, &responder_, cq_, cq_, this) 被调用，这告诉 gRPC 准备接收恰好一个 SayHello 请求。RequestSayHello 的参数告诉 gRPC 将上下文、请求正文和接收到请求的响应者，以及用于通知的完成队列以及应附加到通知事件的标签的位置（在这种情况下， this 用作标签）。HandleRpcs() 阻止 cq->Next()。等待事件发生。一段时间后......
SayHello 请求，gRPC 开始接收并解码该请求。 （IO操作）
一段时间后......
request_ 字段（通过前面提供的指针），然后创建一个事件（以 the pointer to the CallData object 作为标记，正如前面 RequestSayHello 的最后一个参数所询问的那样）。然后 gRPC 将该事件放入完成队列 @987654345@。HandleRpcs() 中的循环接收到事件（之前阻止的对cq->Next() 的调用现在返回），调用CallData::Proceed() 来处理请求。status_ 是PROCESS，所以它执行以下操作：HandleRpcs() 中的循环进入下一次迭代并再次阻塞cq->Next()，等待新事件发生。一段时间后......
cq->Next() 接收事件并返回，CallData::Proceed() 释放 CallData 对象（通过使用delete this;）。 HandleRpcs() 再次循环并阻塞 cq->Next()，等待新事件。该过程可能看起来与同步 API 大致相同，只是对完成队列有额外的访问权限。但是，通过这种方式，在每一个some time later....（通常是等待IO操作完成或等待请求发生），cq->Next()实际上不仅可以接收到该请求的操作完成事件，还可以为其他请求也是如此。
所以如果在第一个请求的时候有一个新的请求进来，比如说，等待回复数据的传输完成，cq->Next() 将获取新请求发出的事件，并开始处理立即并发新请求，而不是等待第一个请求完成传输。
另一方面，同步 API 将始终等待一个请求完全完成（从开始接收到完成回复），然后再开始接收另一个请求。这意味着在接收请求正文数据和发回回复数据（IO 操作）时，CPU 利用率接近 0%。本来可以用来处理其他请求的宝贵 CPU 时间被浪费在等待上。
这真的很糟糕，因为如果一个互联网连接不好的客户端（100 毫秒往返）向服务器发送了一个请求，我们将不得不为此客户端的每个请求花费至少 200 毫秒来积极等待 TCP 传输完成。这将使我们的服务器性能下降到每秒只有约 5 个请求。
如果我们使用异步 API，我们只是不会主动等待任何东西。我们告诉 gRPC：“请将此数据发送给客户端，但我们不会在此处等待您完成。而是在您完成后向完成队列中放一个小信，我们稍后会检查它。”并继续处理其他请求。
您可以看到如何为synchronous APIs 和asynchronous APIs 编写一个简单的服务器
gRPC C++ Performance Nodes 建议的最佳性能做法是生成与 CPU 内核数相等的线程数量，并为每个线程使用一个 CompletionQueue。
【讨论】：