如何用C++ gRPC异步客户端向多服务器流式发送请求？

核心结论 #

完全可以实现你要的逻辑，不需要额外线程，仅用gRPC异步API就能完成，也不需要两个完成队列——单个Completion Queue（CQ）结合标记Tag就能处理所有异步操作，核心是通过Tag区分不同流的不同操作阶段，避免阻塞。

关键概念梳理 #

先明确几个gRPC异步流的核心API作用：

• AsyncMyRequest：初始化客户端流，触发CQ事件标记流就绪
• Write()：异步发送单条消息，无需等待前一条完成（gRPC会自动保证同一流的消息顺序）
• WritesDone()：异步通知服务器“无更多消息发送”，完成后触发CQ事件
• Finish()：异步等待服务器的最终响应，完成后触发CQ事件并返回响应和状态

完整实现示例 #

1. 定义Tag结构体（区分操作类型与流实例） #

用Tag来标记每个异步操作的类型、所属流、关联的响应/状态：

#include <grpcpp/grpcpp.h>
#include <vector>
#include <memory>
#include <iostream>

// 假设你的proto生成的代码包含以下类型
class MyRequest;
class MyResponse;
class MyService;

enum class OperationType {
    STREAM_INIT,    // 流初始化完成
    WRITE_MESSAGE,  // 单条消息发送完成
    WRITES_DONE,    // 所有消息发送完成
    FINISH_STREAM   // 拿到服务器最终响应
};

struct StreamTag {
    OperationType type;
    std::shared_ptr<grpc::ClientAsyncWriter<MyRequest>> writer;
    MyRequest msg;
    MyResponse* response;
    grpc::Status* status;
    int server_idx; // 标记所属服务器，方便后续处理响应
};

2. 主逻辑实现 #

int main() {
    std::vector<std::string> addresses = {"server1:50051", "server2:50052", "server3:50053"};
    std::vector<MyRequest> msgs = {/* 你的消息列表 */};

    grpc::CompletionQueue cq;
    std::vector<std::unique_ptr<MyService::Stub>> stubs;
    std::vector<MyResponse> responses;
    std::vector<grpc::Status> statuses;
    int active_streams = addresses.size();

    // 初始化所有服务器的Stub、响应容器
    for (const auto& addr : addresses) {
        stubs.emplace_back(MyService::NewStub(
            grpc::CreateChannel(addr, grpc::InsecureChannelCredentials())
        ));
        responses.emplace_back();
        statuses.emplace_back();
    }

    // 批量启动所有流的初始化
    for (int i = 0; i < addresses.size(); ++i) {
        grpc::ClientContext context;
        auto* init_tag = new StreamTag{
            OperationType::STREAM_INIT,
            stubs[i]->AsyncMyRequest(&context, &responses[i], &cq),
            MyRequest{},
            &responses[i],
            &statuses[i],
            i
        };
        // 初始化操作会自动触发CQ事件，无需额外调用
    }

    // 轮询CQ处理所有异步事件
    void* tag;
    bool ok;
    while (cq.Next(&tag, &ok)) {
        auto* stream_tag = static_cast<StreamTag*>(tag);
        if (!ok) {
            // 操作失败，清理资源并计数
            delete stream_tag;
            if (--active_streams == 0) break;
            continue;
        }

        switch (stream_tag->type) {
            case OperationType::STREAM_INIT: {
                // 流就绪，批量提交所有消息的异步发送请求
                for (const auto& msg : msgs) {
                    auto* write_tag = new StreamTag{
                        OperationType::WRITE_MESSAGE,
                        stream_tag->writer,
                        msg,
                        stream_tag->response,
                        stream_tag->status,
                        stream_tag->server_idx
                    };
                    stream_tag->writer->Write(msg, write_tag);
                }
                // 提交WritesDone，告知服务器无更多消息
                auto* writes_done_tag = new StreamTag{
                    OperationType::WRITES_DONE,
                    stream_tag->writer,
                    MyRequest{},
                    stream_tag->response,
                    stream_tag->status,
                    stream_tag->server_idx
                };
                stream_tag->writer->WritesDone(writes_done_tag);
                delete stream_tag; // 清理初始化Tag
                break;
            }
            case OperationType::WRITE_MESSAGE: {
                // 单条消息发送完成，直接清理Tag
                delete stream_tag;
                break;
            }
            case OperationType::WRITES_DONE: {
                // 所有消息发送完成，提交Finish等待响应
                auto* finish_tag = new StreamTag{
                    OperationType::FINISH_STREAM,
                    stream_tag->writer,
                    MyRequest{},
                    stream_tag->response,
                    stream_tag->status,
                    stream_tag->server_idx
                };
                stream_tag->writer->Finish(stream_tag->status, finish_tag);
                delete stream_tag;
                break;
            }
            case OperationType::FINISH_STREAM: {
                // 处理服务器返回的响应
                std::cout << "[Server " << stream_tag->server_idx << "] Response: " 
                          << stream_tag->response->DebugString() << std::endl;
                std::cout << "[Server " << stream_tag->server_idx << "] Status: " 
                          << stream_tag->status->ToString() << std::endl;
                delete stream_tag;
                // 所有流完成后退出循环
                if (--active_streams == 0) break;
            }
        }
    }

    std::cout << "All streams processed successfully" << std::endl;
    return 0;
}

针对你的疑问的解答 #

1. 是否需要两个完成队列？
   不需要。单个CQ可以处理所有异步操作，通过Tag区分不同流的不同阶段即可。gRPC会自动将不同的事件分发到对应的Tag上。

2. 能否不借助线程实现？
   完全可以。上述代码全程在主线程执行，所有操作都是异步提交，最后通过cq.Next()轮询处理所有完成的事件，没有使用任何额外线程。

3. WritesDone vs Finish的区别？

• WritesDone()：仅通知服务器“客户端无更多消息要发送”，是流发送阶段的收尾，不会返回响应。
• Finish()：等待服务器的最终响应和状态，是整个流的收尾操作，必须调用才能拿到服务器的结果。

注意事项 #

• Tag的生命周期：必须确保Tag在CQ处理完成前不会被销毁，所以用new创建，处理完成后用delete清理。
• 流的顺序保证：gRPC会自动保证同一流的Write、WritesDone、Finish操作按顺序执行，无需手动同步。
• 错误处理：代码中通过ok参数判断操作是否成功，可根据实际需求添加更详细的错误处理逻辑。

内容的提问来源于stack exchange，提问作者Aristotelis Vontzalidis