proto文件跟同步的相同

完整代码见：grpc_example: 4种RPC模式的c++同步、异步编写模板 (gitee.com)

一、服务端异步代码

1. 单cq

/*
 * 单cq异步服务端：oop格式继承Calldata，推荐使用
 */

#include <iostream>
#include <memory>
#include <string>

#include <grpc/support/log.h>
#include <grpcpp/grpcpp.h>

#ifdef BAZEL_BUILD
#include "examples/protos/example.grpc.pb.h"
#else
#include "./build/example.grpc.pb.h"
#endif

using example::ExampleService;
using example::Request;
using example::Response;
using grpc::Server;
using grpc::ServerBuilder;
using grpc::ServerCompletionQueue;
using grpc::ServerContext;
using grpc::Status;

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_ = builder.AddCompletionQueue();
    server_ = builder.BuildAndStart();
    std::cout << "Async server listening on " << server_address << std::endl;

    HandleRpcs();
  }

private:
  // 基类：统一事件接口，提供构造函数初始化公共成员
  class CallData
  {
  public:
    CallData(ExampleService::AsyncService *service, ServerCompletionQueue *cq)
        : service_(service), cq_(cq), status_(CREATE) {}
    virtual ~CallData() = default;
    virtual void Proceed(bool ok) = 0;

  protected:
    ExampleService::AsyncService *service_;
    ServerCompletionQueue *cq_;
    ServerContext ctx_;
    enum CallStatus
    {
      CREATE,
      PROCESS,
      READING,
      WRITING,
      FINISH
    };
    CallStatus status_;
  };

  // 1. 一元RPC：修复responder_初始化问题
  class UnaryCallData : public CallData
  {
  public:
    // 核心修复：在初始化列表中调用基类构造函数 + 初始化responder_
    UnaryCallData(ExampleService::AsyncService *service, ServerCompletionQueue *cq)
        : CallData(service, cq), responder_(&ctx_)
    {
      Proceed(true);
    }

    void Proceed(bool ok) override
    {
      // std::cout << "UnaryCallData status:" << status_ << std::endl;
      if (status_ == CREATE)
      {
        status_ = PROCESS;
        service_->RequestUnaryCall(&ctx_, &req_, &responder_, cq_, cq_, this);
      }
      else if (status_ == PROCESS)
      {
        new UnaryCallData(service_, cq_);
        res_.set_data("Async Server Unary: " + req_.data());
        status_ = FINISH;
        responder_.Finish(res_, Status::OK, this);
      }
      else
      {
        delete this;
      }
    }

  private:
    Request req_;
    Response res_;
    grpc::ServerAsyncResponseWriter<Response> responder_;
  };

  // 2. 服务端流式RPC：修复responder_初始化问题
  class ServerStreamCallData : public CallData
  {
  public:
    // 核心修复：在初始化列表中初始化responder_
    ServerStreamCallData(ExampleService::AsyncService *service, ServerCompletionQueue *cq)
        : CallData(service, cq), responder_(&ctx_), send_count_(0)
    {
      Proceed(true);
    }

    void Proceed(bool ok) override
    {
      // std::cout << "ServerStreamCallData status:" << status_ << std::endl;

      if (status_ == CREATE)
      {
        status_ = PROCESS;
        service_->RequestServerStream(&ctx_, &req_, &responder_, cq_, cq_, this);
      }
      else if (status_ == PROCESS)
      {
        new ServerStreamCallData(service_, cq_);
        status_ = WRITING;
        WriteNext();
      }
      else if (status_ == WRITING)
      {
        if (send_count_ < 3)
        {
          WriteNext();
        }
        else
        {
          status_ = FINISH;
          responder_.Finish(Status::OK, this);
        }
      }
      else
      {
        delete this;
      }
    }

  private:
    void WriteNext()
    {
      res_.set_data("ServerStream " + std::to_string(send_count_++) + ": " + req_.data());
      responder_.Write(res_, this);
    }

    Request req_;
    Response res_;
    grpc::ServerAsyncWriter<Response> responder_;
    int send_count_;
  };

  // 3. 客户端流式RPC
  class ClientStreamCallData : public CallData
  {
  public:
    // 在初始化列表中初始化responder_
    ClientStreamCallData(ExampleService::AsyncService *service, ServerCompletionQueue *cq)
        : CallData(service, cq), responder_(&ctx_)
    {
      Proceed(true);
    }

    void Proceed(bool ok) override
    {
      // std::cout << "ClientStreamCallData status:" << status_ << std::endl;
      if (status_ == CREATE)
      {
        status_ = READING;
        service_->RequestClientStream(&ctx_, &responder_, cq_, cq_, this);
      }
      else if (status_ == READING)
      {
        if (ok)
        {
          new ClientStreamCallData(service_, cq_);
          status_ = PROCESS;
          responder_.Read(&req_, this);
        }
        else
        {
          status_ = FINISH;
          delete this;
          return;
        }
      }
      else if (status_ == PROCESS)
      {
        if (ok)
        {
          combined_data_ += req_.data() + " ";
          responder_.Read(&req_, this);
        }
        else
        {
          res_.set_data("ClientStream Combined: " + combined_data_);
          status_ = FINISH;
          responder_.Finish(res_, Status::OK, this);
        }
      }
      else
      {
        delete this;
      }
    }

  private:
    Request req_;
    Response res_;
    std::string combined_data_;
    grpc::ServerAsyncReader<Response, Request> responder_;
  };

  // 4. 双向流式RPC
  class BidiStreamCallData : public CallData
  {
  public:
    // 在初始化列表中初始化stream_
    BidiStreamCallData(ExampleService::AsyncService *service, ServerCompletionQueue *cq)
        : CallData(service, cq), stream_(&ctx_)
    {
      Proceed(true);
    }
#if 1
    // 不管连接是否正常关闭
    void Proceed(bool ok) override
    {
      // std::cout << "BidiStreamCallData status:" << status_ << std::endl;
      if (status_ == CREATE)
      {
        service_->RequestBidiStream(&ctx_, &stream_, cq_, cq_, this);
        status_ = PROCESS;
      }
      else if (status_ == PROCESS)
      {
        if (ok)
        {
          new BidiStreamCallData(service_, cq_);
          stream_.Read(&req_, this);
          status_ = READING;
        }
        else
        {
          status_ = FINISH;
          stream_.Finish(Status::OK, this);
        }
      }
      else if (status_ == WRITING) // 表示刚刚写完数据触发
      {
        if (ok)
        {
          stream_.Read(&req_, this);
          status_ = READING;
        }
        else
        {
          stream_.Finish(Status::OK, this);
          status_ = FINISH;
        }
      }
      else if (status_ == READING) // 表示刚刚读完数据触发
      {
        if (ok)
        {
          res_.set_data("BidiStream Echo: " + req_.data());
          stream_.Write(res_, this);
          status_ = WRITING;
        }
        else
        {

          stream_.Finish(Status::OK, this);
          status_ = FINISH;
        }
      }
      else if (status_ == FINISH)
      {

        delete this;
        std::cout << "delete" << std::endl;
      }
      // std::cout << "BidiStreamCallData status:" << status_ << std::endl;
    }
#else
    // 在意连接是否正常关闭
    void Proceed(bool ok) override
    {
      // std::cout << "BidiStreamCallData status:" << status_ << std::endl;
      if (status_ == CREATE)
      {
        service_->RequestBidiStream(&ctx_, &stream_, cq_, cq_, this);
        status_ = PROCESS;
      }
      else if (status_ == PROCESS)
      {
        if (ok)
        {
          new BidiStreamCallData(service_, cq_);
          stream_.Read(&req_, this);
          status_ = READING;
        }
        else
        {
          status_ = FINISH;
          rpc_status_ = Status(grpc::StatusCode::CANCELLED, "PROCESS stage ok=false"); // ✅ 赋值异常状态
          stream_.Finish(rpc_status_, this);
        }
      }
      else if (status_ == WRITING) // 表示刚刚写完数据触发
      {
        if (ok)
        {
          stream_.Read(&req_, this);
          status_ = READING;
        }
        else
        {
          rpc_status_ = Status(grpc::StatusCode::INTERNAL, "Write operation failed"); // ✅ 赋值异常状态
          stream_.Finish(rpc_status_, this);
          status_ = FINISH;
        }
      }
      else if (status_ == READING) // 表示刚刚读完数据触发
      {
        if (ok)
        {
          res_.set_data("BidiStream Echo: " + req_.data());
          stream_.Write(res_, this);
          status_ = WRITING;
        }
        else
        {
          // 正常关闭（客户端发完数据，Read返回false）：赋值正常状态
          rpc_status_ = Status::OK; // ✅ 正常状态
          stream_.Finish(rpc_status_, this);
          status_ = FINISH;
        }
      }
      else if (status_ == FINISH)
      {
        // ✅ 核心：判断是否异常关闭
        if (rpc_status_.ok())
        {
          std::cout << "BidiStream正常关闭 | 最后响应：" << res_.data() << std::endl;
        }
        else
        {
          // 打印异常原因（错误码+错误信息），生产环境建议用日志框架（如glog/logger）
          std::cerr << "BidiStream异常关闭 | 错误码：" << rpc_status_.error_code()
                    << " | 错误信息：" << rpc_status_.error_message() << std::endl;
        }
        delete this;
        std::cout << "delete" << std::endl;
      }
    }
#endif
  private:
    Request req_;
    Response res_;
    grpc::ServerAsyncReaderWriter<Response, Request> stream_;
    Status rpc_status_;
  };

  void HandleRpcs()
  {
    new UnaryCallData(&service_, cq_.get());
    new ServerStreamCallData(&service_, cq_.get());
    new ClientStreamCallData(&service_, cq_.get());
    new BidiStreamCallData(&service_, cq_.get());

    void *tag;
    bool ok;
    while (cq_->Next(&tag, &ok))
    {
      static_cast<CallData *>(tag)->Proceed(ok);
    }
  }

  std::unique_ptr<ServerCompletionQueue> cq_;
  ExampleService::AsyncService service_;
  std::unique_ptr<Server> server_;
};

int main(int argc, char **argv)
{
  ServerImpl server;
  server.Run();
  return 0;
}

代码核心解析

• 准备阶段

  • 定义基类 CallData，包含公共成员：service_、cq_ (完成队列指针)、ctx_、状态机。
  • 声明纯虚函数 Proceed(bool ok)。

• 核心循环（CQ 驱动）

  • 在 HandleRpcs() 中：
    • 创建初始对象：new UnaryCallData(...) 等，构造函数中触发首次逻辑。
    • 事件循环：while (cq_->Next(&tag, &ok)) —— CQ (Completion Queue) 在此阻塞等待事件。
    • 分发事件：static_cast<CallData*>(tag)->Proceed(ok) —— 通过 Tag (this 指针) 找回对象并处理。

• RPC 处理逻辑（以任意派生类为例）

  • 状态 CREATE：
    • 注册请求：service_->RequestXXX(&ctx_, ..., cq_, cq_, this) —— 向框架注册 Request 监听器，并将 Tag (this) 绑定到该请求。
    • 切换状态。
  • 状态 PROCESS/READING/WRITING：
    • 业务逻辑处理。
    • 调用 Read/Write/Finish 等异步 API，同样传入 Tag (this)。
  • 状态 FINISH：
    • delete this；清理资源

2. 多cq

/*
 * 多cq异步服务端：oop格式继承Calldata，推荐使用
 */

#include <iostream>
#include <memory>
#include <string>
#include <thread>
#include <vector>
#include <atomic>

#include <grpc/support/log.h>
#include <grpcpp/grpcpp.h>

#ifdef BAZEL_BUILD
#include "examples/protos/example.grpc.pb.h"
#else
#include "./build/example.grpc.pb.h"
#endif

using example::ExampleService;
using example::Request;
using example::Response;
using grpc::Server;
using grpc::ServerBuilder;
using grpc::ServerCompletionQueue;
using grpc::ServerContext;
using grpc::Status;

// 前向声明
class ServerImpl;

class ServerImpl final
{
public:
    ~ServerImpl()
    {
        if (server_)
        {
            server_->Shutdown();
        }
        for (auto &cq : cqs_)
        {
            if (cq)
            {
                cq->Shutdown();
            }
        }
        for (auto &t : cq_threads_)
        {
            if (t.joinable())
            {
                t.join();
            }
        }
        std::cout << "Multi-CQ server shutdown completed" << std::endl;
    }

    void Run(int num_cqs = std::thread::hardware_concurrency())
    {
        std::string server_address("0.0.0.0:50051");
        num_cqs_ = num_cqs;
        next_cq_idx_ = 0;

        ServerBuilder builder;
        builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
        builder.RegisterService(&service_);

        for (int i = 0; i < num_cqs_; ++i)
        {
            cqs_.push_back(builder.AddCompletionQueue());
        }

        server_ = builder.BuildAndStart();
        std::cout << "Multi-CQ async server listening on " << server_address
                  << " with " << num_cqs_ << " CQs/threads" << std::endl;

        for (int i = 0; i < num_cqs_; ++i)
        {
            cq_threads_.emplace_back(&ServerImpl::HandleRpcs, this, cqs_[i].get());
        }

        // 初始化CallData，传递this指针
        new UnaryCallData(&service_, GetNextCQ(), this);
        new ServerStreamCallData(&service_, GetNextCQ(), this);
        new ClientStreamCallData(&service_, GetNextCQ(), this);
        new BidiStreamCallData(&service_, GetNextCQ(), this);

        while (true)
        {
            std::this_thread::sleep_for(std::chrono::seconds(1));
        }
    }

private:
    // 基类：新增ServerImpl*参数
    class CallData
    {
    public:
        CallData(ExampleService::AsyncService *service, ServerCompletionQueue *cq, ServerImpl *server)
            : service_(service), cq_(cq), server_(server), status_(CREATE) {}
        virtual ~CallData() = default;
        virtual void Proceed(bool ok) = 0;

    protected:
        ExampleService::AsyncService *service_;
        ServerCompletionQueue *cq_;
        ServerImpl *server_; // 持有ServerImpl指针
        ServerContext ctx_;
        enum CallStatus
        {
            CREATE,
            PROCESS,
            READING,
            WRITING,
            FINISH
        };
        CallStatus status_;
    };

    // 1. 一元RPC：修改构造函数，传递ServerImpl*
    class UnaryCallData : public CallData
    {
    public:
        UnaryCallData(ExampleService::AsyncService *service, ServerCompletionQueue *cq, ServerImpl *server)
            : CallData(service, cq, server), responder_(&ctx_)
        {
            Proceed(true);
        }

        void Proceed(bool ok) override
        {
            if (status_ == CREATE)
            {
                status_ = PROCESS;
                service_->RequestUnaryCall(&ctx_, &req_, &responder_, cq_, cq_, this);
            }
            else if (status_ == PROCESS)
            {
                // 使用server_->GetNextCQ()获取下一个CQ
                new UnaryCallData(service_, server_->GetNextCQ(), server_);
                res_.set_data("Async Server Unary: " + req_.data());
                status_ = FINISH;
                responder_.Finish(res_, Status::OK, this);
            }
            else
            {
                delete this;
            }
        }

    private:
        Request req_;
        Response res_;
        grpc::ServerAsyncResponseWriter<Response> responder_;
    };

    // 2. 服务端流式RPC：修改构造函数，传递ServerImpl*
    class ServerStreamCallData : public CallData
    {
    public:
        ServerStreamCallData(ExampleService::AsyncService *service, ServerCompletionQueue *cq, ServerImpl *server)
            : CallData(service, cq, server), responder_(&ctx_), send_count_(0)
        {
            Proceed(true);
        }

        void Proceed(bool ok) override
        {
            if (status_ == CREATE)
            {
                status_ = PROCESS;
                service_->RequestServerStream(&ctx_, &req_, &responder_, cq_, cq_, this);
            }
            else if (status_ == PROCESS)
            {
                new ServerStreamCallData(service_, server_->GetNextCQ(), server_);
                status_ = WRITING;
                WriteNext();
            }
            else if (status_ == WRITING)
            {
                if (send_count_ < 3)
                {
                    WriteNext();
                }
                else
                {
                    status_ = FINISH;
                    responder_.Finish(Status::OK, this);
                }
            }
            else
            {
                delete this;
            }
        }

    private:
        void WriteNext()
        {
            res_.set_data("ServerStream " + std::to_string(send_count_++) + ": " + req_.data());
            responder_.Write(res_, this);
        }

        Request req_;
        Response res_;
        grpc::ServerAsyncWriter<Response> responder_;
        int send_count_;
    };

    // 3. 客户端流式RPC：修改构造函数，传递ServerImpl*
    class ClientStreamCallData : public CallData
    {
    public:
        ClientStreamCallData(ExampleService::AsyncService *service, ServerCompletionQueue *cq, ServerImpl *server)
            : CallData(service, cq, server), responder_(&ctx_)
        {
            Proceed(true);
        }

        void Proceed(bool ok) override
        {
            if (status_ == CREATE)
            {
                status_ = READING;
                service_->RequestClientStream(&ctx_, &responder_, cq_, cq_, this);
            }
            else if (status_ == READING)
            {
                if (ok)
                {
                    new ClientStreamCallData(service_, server_->GetNextCQ(), server_);
                    status_ = PROCESS;
                    responder_.Read(&req_, this);
                }
                else
                {
                    status_ = FINISH;
                    delete this;
                    return;
                }
            }
            else if (status_ == PROCESS)
            {
                if (ok)
                {
                    combined_data_ += req_.data() + " ";
                    responder_.Read(&req_, this);
                }
                else
                {
                    res_.set_data("ClientStream Combined: " + combined_data_);
                    status_ = FINISH;
                    responder_.Finish(res_, Status::OK, this);
                }
            }
            else
            {
                delete this;
            }
        }

    private:
        Request req_;
        Response res_;
        std::string combined_data_;
        grpc::ServerAsyncReader<Response, Request> responder_;
    };

    // 4. 双向流式RPC：修改构造函数，传递ServerImpl*
    class BidiStreamCallData : public CallData
    {
    public:
        BidiStreamCallData(ExampleService::AsyncService *service, ServerCompletionQueue *cq, ServerImpl *server)
            : CallData(service, cq, server), stream_(&ctx_)
        {
            Proceed(true);
        }

#if 1
    // 不管连接是否正常关闭
    void Proceed(bool ok) override
    {
      // std::cout << "BidiStreamCallData status:" << status_ << std::endl;
      if (status_ == CREATE)
      {
        service_->RequestBidiStream(&ctx_, &stream_, cq_, cq_, this);
        status_ = PROCESS;
      }
      else if (status_ == PROCESS)
      {
        if (ok)
        {
          new BidiStreamCallData(service_, server_->GetNextCQ(), server_);
          stream_.Read(&req_, this);
          status_ = READING;
        }
        else
        {
          status_ = FINISH;
          stream_.Finish(Status::OK, this);
        }
      }
      else if (status_ == WRITING) // 表示刚刚写完数据触发
      {
        if (ok)
        {
          stream_.Read(&req_, this);
          status_ = READING;
        }
        else
        {
          stream_.Finish(Status::OK, this);
          status_ = FINISH;
        }
      }
      else if (status_ == READING) // 表示刚刚读完数据触发
      {
        if (ok)
        {
          res_.set_data("BidiStream Echo: " + req_.data());
          stream_.Write(res_, this);
          status_ = WRITING;
        }
        else
        {

          stream_.Finish(Status::OK, this);
          status_ = FINISH;
        }
      }
      else if (status_ == FINISH)
      {

        delete this;
        std::cout << "delete" << std::endl;
      }
      // std::cout << "BidiStreamCallData status:" << status_ << std::endl;
    }
#else
        // 在意连接是否正常关闭
        void Proceed(bool ok) override
        {
            // std::cout << "BidiStreamCallData status:" << status_ << std::endl;
            if (status_ == CREATE)
            {
                service_->RequestBidiStream(&ctx_, &stream_, cq_, cq_, this);
                status_ = PROCESS;
            }
            else if (status_ == PROCESS)
            {
                if (ok)
                {
                    new BidiStreamCallData(service_, server_->GetNextCQ(), server_);
                    stream_.Read(&req_, this);
                    status_ = READING;
                }
                else
                {
                    status_ = FINISH;
                    rpc_status_ = Status(grpc::StatusCode::CANCELLED, "PROCESS stage ok=false"); // ✅ 赋值异常状态
                    stream_.Finish(rpc_status_, this);
                }
            }
            else if (status_ == WRITING) // 表示刚刚写完数据触发
            {
                if (ok)
                {
                    stream_.Read(&req_, this);
                    status_ = READING;
                }
                else
                {
                    rpc_status_ = Status(grpc::StatusCode::INTERNAL, "Write operation failed"); // ✅ 赋值异常状态
                    stream_.Finish(rpc_status_, this);
                    status_ = FINISH;
                }
            }
            else if (status_ == READING) // 表示刚刚读完数据触发
            {
                if (ok)
                {
                    res_.set_data("BidiStream Echo: " + req_.data());
                    stream_.Write(res_, this);
                    status_ = WRITING;
                }
                else
                {
                    // 正常关闭（客户端发完数据，Read返回false）：赋值正常状态
                    rpc_status_ = Status::OK; // ✅ 正常状态
                    stream_.Finish(rpc_status_, this);
                    status_ = FINISH;
                }
            }
            else if (status_ == FINISH)
            {
                // ✅ 核心：判断是否异常关闭
                if (rpc_status_.ok())
                {
                    std::cout << "BidiStream正常关闭 | 最后响应：" << res_.data() << std::endl;
                }
                else
                {
                    // 打印异常原因（错误码+错误信息），生产环境建议用日志框架（如glog/logger）
                    std::cerr << "BidiStream异常关闭 | 错误码：" << rpc_status_.error_code()
                              << " | 错误信息：" << rpc_status_.error_message() << std::endl;
                }
                delete this;
                std::cout << "delete" << std::endl;
            }
        }
#endif

    private:
        Request req_;
        Response res_;
        grpc::ServerAsyncReaderWriter<Response, Request> stream_;
        Status rpc_status_;
    };

    // 核心工具函数：线程安全的轮询获取下一个CQ
    ServerCompletionQueue *GetNextCQ()
    {
        int idx = next_cq_idx_++ % num_cqs_;
        return cqs_[idx].get();
    }

    // 单个CQ的事件循环
    void HandleRpcs(ServerCompletionQueue *cq)
    {
        void *tag;
        bool ok;
        while (cq->Next(&tag, &ok))
        {
            static_cast<CallData *>(tag)->Proceed(ok);
        }
        std::cout << "CQ thread exited (CQ ptr: " << cq << ")" << std::endl;
    }

    ExampleService::AsyncService service_;
    std::unique_ptr<Server> server_;
    int num_cqs_;
    std::vector<std::unique_ptr<ServerCompletionQueue>> cqs_;
    std::vector<std::thread> cq_threads_;
    std::atomic<int> next_cq_idx_{0};
};

int main(int argc, char **argv)
{
    ServerImpl server;
    server.Run();
    return 0;
}

多 CQ 与单 CQ 的编写差别

1. 成员变量

   • 单 CQ：单个 std::unique_ptr<ServerCompletionQueue> cq_
   • 多 CQ：std::vector 存储多个 CQ、线程、原子索引 next_cq_idx_

2. Run() 函数

   • 多 CQ：循环创建多个 CQ，为每个 CQ 启动一个独立线程运行 HandleRpcs

3. CallData 基类

   • 多 CQ：新增 ServerImpl* server_ 成员，构造函数需传入该指针

4. 派生类（Unary/ServerStream/ClientStream/BidiStream）

   • 构造函数新增 ServerImpl* 参数
   • 创建新 CallData 时，通过 server_->GetNextCQ() 轮询获取下一个 CQ

5. 新增工具函数

   • 多 CQ：GetNextCQ()，线程安全地轮询返回 CQ

6. HandleRpcs() 函数

   • 多 CQ：接受 ServerCompletionQueue* 作为参数，每个线程处理专属 CQ

7. 析构函数

   • 多 CQ：循环 Shutdown 所有 CQ，Join 所有线程

二、客户端异步代码

1. 单cq

/*
 * 单cq异步客户端：oop格式继承Calldata，推荐使用
 */
#include <iostream>
#include <memory>
#include <string>
#include <thread>
#include <vector>

#include <grpc/support/log.h>
#include <grpcpp/grpcpp.h>

#ifdef BAZEL_BUILD
#include "examples/protos/example.grpc.pb.h"
#else
#include "./build/example.grpc.pb.h"
#endif

using example::ExampleService;
using example::Request;
using example::Response;
using grpc::Channel;
using grpc::ClientContext;
using grpc::CompletionQueue;
using grpc::Status;

class ClientImpl final
{
public:
    explicit ClientImpl(std::shared_ptr<Channel> channel)
        : stub_(ExampleService::NewStub(channel))
    {
        // 启动事件循环线程（对应服务端的 HandleRpcs）
        cq_thread_ = std::thread(&ClientImpl::HandleEvents, this);
    }

    ~ClientImpl()
    {
        cq_.Shutdown();
        if (cq_thread_.joinable())
        {
            cq_thread_.join();
        }
    }

    // 对外提供的 RPC 发起接口（对应服务端的 RequestXXX）
    void AsyncUnary(const std::string &msg)
    {
        new UnaryCallData(stub_.get(), &cq_, msg);
    }

    void AsyncServerStream(const std::string &msg)
    {
        new ServerStreamCallData(stub_.get(), &cq_, msg);
    }

    void AsyncClientStream(const std::vector<std::string> &msgs)
    {
        new ClientStreamCallData(stub_.get(), &cq_, msgs);
    }

    void AsyncBidiStream(const std::vector<std::string> &msgs)
    {
        new BidiStreamCallData(stub_.get(), &cq_, msgs);
    }

private:
    // 基类：统一事件接口，对应服务端的 CallData
    class CallData
    {
    public:
        CallData(ExampleService::Stub *stub, CompletionQueue *cq)
            : stub_(stub), cq_(cq), status_(CREATE) {}
        virtual ~CallData() = default;
        virtual void Proceed(bool ok) = 0;

    protected:
        ExampleService::Stub *stub_;
        CompletionQueue *cq_;
        ClientContext ctx_;
        enum CallStatus
        {
            CREATE,
            PROCESS,
            READING,
            WRITING,
            WRITES_DONE,
            FINISH
        };
        CallStatus status_;

        Request req_;
        Response res_;
        Status rpc_status_;
    };

    // 1. 一元RPC：对应服务端的 UnaryCallData
    class UnaryCallData : public CallData
    {
    public:
        UnaryCallData(ExampleService::Stub *stub, CompletionQueue *cq, const std::string &msg)
            : CallData(stub, cq)
        {
            req_.set_data(msg);
            Proceed(true); // 对应服务端构造函数里的 Proceed
        }

        void Proceed(bool ok) override
        {
            if (status_ == CREATE)
            {
                status_ = PROCESS;
                // 客户端异步一元RPC三步法
                reader_ = stub_->PrepareAsyncUnaryCall(&ctx_, req_, cq_);
                reader_->StartCall();
                reader_->Finish(&res_, &rpc_status_, this);
            }
            else if (status_ == PROCESS)
            {
                GPR_ASSERT(ok);
                if (rpc_status_.ok())
                {
                    std::cout << "[Unary] Received: " << res_.data() << std::endl;
                }
                else
                {
                    std::cout << "[Unary] Failed: " << rpc_status_.error_message() << std::endl;
                }
                status_ = FINISH;
                delete this; // 对应服务端的 delete this
            }
        }

    private:
        std::unique_ptr<grpc::ClientAsyncResponseReader<Response>> reader_;
    };

    // 2. 服务端流式RPC：对应服务端的 ServerStreamCallData
    class ServerStreamCallData : public CallData
    {
    public:
        ServerStreamCallData(ExampleService::Stub *stub, CompletionQueue *cq, const std::string &msg)
            : CallData(stub, cq)
        {
            req_.set_data(msg);
            Proceed(true);
        }

        void Proceed(bool ok) override
        {


            if (status_ == CREATE)
            {
                status_ = PROCESS;
                reader_ = stub_->PrepareAsyncServerStream(&ctx_, req_, cq_);
                reader_->StartCall(this);
            }
            else if (status_ == PROCESS)
            {
                status_ = READING;
                reader_->Read(&res_, this);
            }
            else if (status_ == READING)
            {
                if (ok)
                {
                    std::cout << "[ServerStream] Received: " << res_.data() << std::endl;
                    reader_->Read(&res_, this);
                }
                else
                {
                    status_ = FINISH;
                    reader_->Finish(&rpc_status_, this);
                }
            }
            else if (status_ == FINISH)
            {
                if (rpc_status_.ok())
                {
                    std::cout << "[ServerStream] Completed" << std::endl;
                }
                else
                {
                    std::cout << "[ServerStream] Failed: " << rpc_status_.error_message() << std::endl;
                }
                delete this;
            }
        }

    private:
        std::unique_ptr<grpc::ClientAsyncReader<Response>> reader_;
    };

    // 3. 客户端流式RPC：对应服务端的 ClientStreamCallData
    class ClientStreamCallData : public CallData
    {
    public:
        ClientStreamCallData(ExampleService::Stub *stub, CompletionQueue *cq, const std::vector<std::string> &msgs)
            : CallData(stub, cq), send_msgs_(msgs), send_idx_(0)
        {
            Proceed(true);
        }

        void Proceed(bool ok) override
        {
        

            if (status_ == CREATE)
            {
                status_ = PROCESS;
                writer_ = stub_->PrepareAsyncClientStream(&ctx_, &res_, cq_);
                writer_->StartCall(this);
            }
            else if (status_ == PROCESS)
            {
                status_ = WRITING;
                req_.set_data(send_msgs_[send_idx_++]);
                writer_->Write(req_, this);
            }
            else if (status_ == WRITING)
            {
                if (ok && send_idx_ < send_msgs_.size())
                {
                    req_.set_data(send_msgs_[send_idx_++]);
                    writer_->Write(req_, this);
                }
                else
                {
                    status_ = WRITES_DONE;
                    writer_->WritesDone(this);
                }
            }
            else if (status_ == WRITES_DONE)
            {
                status_ = FINISH;
                writer_->Finish(&rpc_status_, this);
            }
            else if (status_ == FINISH)
            {
                if (rpc_status_.ok())
                {
                    std::cout << "[ClientStream] Received: " << res_.data() << std::endl;
                }
                else
                {
                    std::cout << "[ClientStream] Failed: " << rpc_status_.error_message() << std::endl;
                }
                std::cout << "[ClientStream]: complete" << std::endl;
                delete this;
            }
            
        }

    private:
        std::unique_ptr<grpc::ClientAsyncWriter<Request>> writer_;
        std::vector<std::string> send_msgs_;
        size_t send_idx_;
    };

    // 4. 双向流式RPC：修正状态机，先写后读，确保流正常推进
    class BidiStreamCallData : public CallData
    {
    public:
        BidiStreamCallData(ExampleService::Stub *stub, CompletionQueue *cq, const std::vector<std::string> &msgs)
            : CallData(stub, cq), send_msgs_(msgs), send_idx_(0)
        {
            Proceed(true);
        }

#if 1
        void Proceed(bool ok) override
        {

            if (status_ == CREATE)
            {
                status_ = PROCESS;
                stream_ = stub_->PrepareAsyncBidiStream(&ctx_, cq_);
                stream_->StartCall(this);
            }
            else if (status_ == PROCESS)
            {
                if (ok)
                {
                    // 先发起Write写数据，再Read读响应，打破死锁
                    req_.set_data(send_msgs_[send_idx_++]);
                    stream_->Write(req_, this);
                    status_ = WRITING;
                }
                else
                {
                    stream_->Finish(&rpc_status_, this);
                    status_ = FINISH;
                }
            }
            else if (status_ == WRITING)
            {
                if (ok)
                {
                    stream_->Read(&res_, this);
                    status_ = READING;
                }
                else
                {
                    // 写失败，尝试关闭流
                    stream_->Finish(&rpc_status_, this);
                    status_ = FINISH;
                }
            }
            else if (status_ == READING)
            {
                if (ok)
                {
                    // 读到响应，处理并继续读
                    std::cout << "[BidiStream] Received: " << res_.data() << std::endl;
                    if (send_idx_ < send_msgs_.size())
                    {
                        // 还有数据，继续写
                        req_.set_data(send_msgs_[send_idx_++]);
                        stream_->Write(req_, this);
                        status_ = WRITING;
                    }
                    else
                    {
                        // 写完所有数据，发送WritesDone
                        stream_->WritesDone(this);
                        status_ = WRITES_DONE;
                    }
                }
                else
                {
                    // 关闭流
                    status_ = FINISH;
                    stream_->Finish(&rpc_status_, this);
                }
            }
            else if (status_ == WRITES_DONE)
            {
                if (ok)
                {
                    stream_->Read(&res_, this);
                    status_ = READING;
                }
                else
                {
                    // 写失败，尝试关闭流
                    stream_->Finish(&rpc_status_, this);
                    status_ = FINISH;
                }
            }
            else if (status_ == FINISH)
            {
                if (rpc_status_.ok())
                {
                    std::cout << "[BidiStream] Completed" << std::endl;
                }
                else
                {
                    std::cout << "[BidiStream] Failed: " << rpc_status_.error_message() << std::endl;
                }
                delete this;
            }
        }
#else
        void Proceed(bool ok) override
        {
            if (status_ == CREATE)
            {
                status_ = PROCESS;
                stream_ = stub_->PrepareAsyncBidiStream(&ctx_, cq_);
                stream_->StartCall(this);
            }
            else if (status_ == PROCESS)
            {
                // 先发起Write写数据，再Read读响应，打破死锁
                status_ = WRITING;
                req_.set_data(send_msgs_[send_idx_++]);
                stream_->Write(req_, this);
            }
            else if (status_ == WRITING)
            {
                if (ok)
                {
                    if (send_idx_ < send_msgs_.size())
                    {
                        // 还有数据，继续写
                        req_.set_data(send_msgs_[send_idx_++]);
                        stream_->Write(req_, this);
                    }
                    else
                    {
                        // 写完所有数据，发送WritesDone
                        status_ = WRITES_DONE;
                        stream_->WritesDone(this);
                    }
                }
                else
                {
                    // 关闭流
                    status_ = FINISH;
                    stream_->Finish(&rpc_status_, this);
                    
                }
            }
            else if (status_ == WRITES_DONE)
            {
                // WritesDone完成后，开始读服务端响应
                status_ = READING;
                stream_->Read(&res_, this);
            }
            else if (status_ == READING)
            {
                if (ok)
                {
                    // 读到响应，处理并继续读
                    std::cout << "[BidiStream] Received: " << res_.data() << std::endl;
                    res_.clear_data();
                    stream_->Read(&res_, this);
                }
                else
                {
                    // 关闭流
                    status_ = FINISH;
                    stream_->Finish(&rpc_status_, this);
                }
            }
            else if (status_ == FINISH)
            {
                if (rpc_status_.ok())
                {
                    std::cout << "[BidiStream] Completed" << std::endl;
                }
                else
                {
                    std::cout << "[BidiStream] Failed: " << rpc_status_.error_message() << std::endl;
                }
                delete this;
            }
        }
#endif

    private:
        std::unique_ptr<grpc::ClientAsyncReaderWriter<Request, Response>> stream_;
        std::vector<std::string> send_msgs_;
        size_t send_idx_;
    };

    // 对应服务端的 HandleRpcs：事件循环
    void HandleEvents()
    {
        void *tag;
        bool ok;
        while (cq_.Next(&tag, &ok))
        {
            static_cast<CallData *>(tag)->Proceed(ok);
        }
    }

    std::unique_ptr<ExampleService::Stub> stub_;
    CompletionQueue cq_;
    std::thread cq_thread_;
};

int main(int argc, char **argv)
{
    ClientImpl client(grpc::CreateChannel(
        "localhost:50051", grpc::InsecureChannelCredentials()));

    // 发起RPC调用（对应服务端启动时创建CallData）
    client.AsyncUnary("Hello Async");
    client.AsyncServerStream("Hi Stream");
    client.AsyncClientStream({"A", "B", "C"});
    client.AsyncBidiStream({"X", "Y", "Z"});

    std::cout << "Press control-c to quit" << std::endl
              << std::endl;
    // 阻塞主线程
    while (true)
    {
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }

    return 0;
}

• 准备阶段

  • 定义基类 CallData，包含公共成员：stub_、cq_ (完成队列指针)、ctx_、状态机、req_/res_/rpc_status_。
  • 声明纯虚函数 Proceed(bool ok)。

• 核心循环（CQ 驱动）

  • 在构造函数中启动独立线程运行 HandleEvents()。
  • HandleEvents() 内：
    • while (cq_->Next(&tag, &ok)) —— CQ (Completion Queue) 在此阻塞等待事件。
    • static_cast<CallData*>(tag)->Proceed(ok) —— 通过 Tag (this 指针) 找回对象并处理。

• 对外 RPC 发起接口

  • AsyncUnary/AsyncServerStream/AsyncClientStream/AsyncBidiStream：new 对应类型的 CallData 对象，传入 stub_、cq_ 及业务参数。

• RPC 处理逻辑（以派生类为例）

  • 状态 CREATE：
    • Prepare：调用 stub_->PrepareAsyncXXX(...) 准备 RPC。
    • StartCall：调用 StartCall() 或 StartCall(this) 启动 RPC，传入 Tag (this)。
    • 切换状态。
  • 状态 PROCESS/READING/WRITING：
    • 业务逻辑处理。
    • 调用 Read/Write/WritesDone/Finish 等异步 API，传入 Tag (this)。
  • 状态 FINISH：
    • 处理 rpc_status_。
    • delete this；清理资源。

• main 函数

  • 创建 ClientImpl 对象。
  • 调用对外接口发起 RPC。
  • 阻塞主线程。

2. cq

/*
* 多cq异步客户端：oop格式继承Calldata，推荐使用
*/

#include <iostream>
#include <memory>
#include <string>
#include <thread>
#include <vector>
#include <atomic>  // 新增：原子变量用于轮询分配CQ

#include <grpc/support/log.h>
#include <grpcpp/grpcpp.h>

#ifdef BAZEL_BUILD
#include "examples/protos/example.grpc.pb.h"
#else
#include "./build/example.grpc.pb.h"
#endif

using example::ExampleService;
using example::Request;
using example::Response;
using grpc::Channel;
using grpc::ClientContext;
using grpc::CompletionQueue;
using grpc::Status;

// 前向声明
class ClientImpl;

class ClientImpl final
{
public:
    // 改造1：新增num_cqs参数，默认使用硬件并发数
    explicit ClientImpl(std::shared_ptr<Channel> channel, int num_cqs = std::thread::hardware_concurrency())
        : stub_(ExampleService::NewStub(channel)), num_cqs_(num_cqs), next_cq_idx_(0)
    {
        // 改造2：创建多个CQ并启动对应线程
        for (int i = 0; i < num_cqs_; ++i) {
            cqs_.emplace_back(std::make_unique<CompletionQueue>());
            cq_threads_.emplace_back(&ClientImpl::HandleEvents, this, cqs_[i].get());
        }
        std::cout << "Multi-CQ client initialized with " << num_cqs_ << " CQs/threads" << std::endl;
    }

    // 改造3：析构函数关闭所有CQ并join线程
    ~ClientImpl()
    {
        // 关闭所有CQ
        for (auto& cq : cqs_) {
            if (cq) {
                cq->Shutdown();
            }
        }
        // 等待所有CQ线程退出
        for (auto& t : cq_threads_) {
            if (t.joinable()) {
                t.join();
            }
        }
        std::cout << "Multi-CQ client shutdown completed" << std::endl;
    }

    // 改造4：对外接口使用GetNextCQ获取轮询的CQ
    void AsyncUnary(const std::string &msg)
    {
        new UnaryCallData(stub_.get(), GetNextCQ(), this, msg);
    }

    void AsyncServerStream(const std::string &msg)
    {
        new ServerStreamCallData(stub_.get(), GetNextCQ(), this, msg);
    }

    void AsyncClientStream(const std::vector<std::string> &msgs)
    {
        new ClientStreamCallData(stub_.get(), GetNextCQ(), this, msgs);
    }

    void AsyncBidiStream(const std::vector<std::string> &msgs)
    {
        new BidiStreamCallData(stub_.get(), GetNextCQ(), this, msgs);
    }

private:
    // 改造5：基类新增ClientImpl指针，用于获取CQ
    class CallData
    {
    public:
        CallData(ExampleService::Stub *stub, CompletionQueue *cq, ClientImpl* client)
            : stub_(stub), cq_(cq), client_(client), status_(CREATE) {}
        virtual ~CallData() = default;
        virtual void Proceed(bool ok) = 0;

    protected:
        ExampleService::Stub *stub_;
        CompletionQueue *cq_;
        ClientImpl* client_;  // 持有ClientImpl指针，用于后续操作（如果需要）
        ClientContext ctx_;
        enum CallStatus
        {
            CREATE,
            PROCESS,
            READING,
            WRITING,
            WRITES_DONE,
            FINISH
        };
        CallStatus status_;

        Request req_;
        Response res_;
        Status rpc_status_;
    };

    // 1. 一元RPC：修改构造函数，新增ClientImpl*参数
    class UnaryCallData : public CallData
    {
    public:
        UnaryCallData(ExampleService::Stub *stub, CompletionQueue *cq, ClientImpl* client, const std::string &msg)
            : CallData(stub, cq, client)
        {
            req_.set_data(msg);
            Proceed(true);
        }

        void Proceed(bool ok) override
        {
            if (status_ == CREATE)
            {
                status_ = PROCESS;
                // 客户端异步一元RPC三步法（使用当前CQ）
                reader_ = stub_->PrepareAsyncUnaryCall(&ctx_, req_, cq_);
                reader_->StartCall();
                reader_->Finish(&res_, &rpc_status_, this);
            }
            else if (status_ == PROCESS)
            {
                GPR_ASSERT(ok);
                if (rpc_status_.ok())
                {
                    std::cout << "[Unary] Received: " << res_.data() << std::endl;
                }
                else
                {
                    std::cout << "[Unary] Failed: " << rpc_status_.error_message() << std::endl;
                }
                status_ = FINISH;
                delete this;
            }
        }

    private:
        std::unique_ptr<grpc::ClientAsyncResponseReader<Response>> reader_;
    };

    // 2. 服务端流式RPC：修改构造函数，新增ClientImpl*参数
    class ServerStreamCallData : public CallData
    {
    public:
        ServerStreamCallData(ExampleService::Stub *stub, CompletionQueue *cq, ClientImpl* client, const std::string &msg)
            : CallData(stub, cq, client)
        {
            req_.set_data(msg);
            Proceed(true);
        }

        void Proceed(bool ok) override
        {

            if (status_ == CREATE)
            {
                status_ = PROCESS;
                reader_ = stub_->PrepareAsyncServerStream(&ctx_, req_, cq_);
                reader_->StartCall(this);
            }
            else if (status_ == PROCESS)
            {
                status_ = READING;
                reader_->Read(&res_, this);
            }
            else if (status_ == READING)
            {
                if (ok)
                {
                    std::cout << "[ServerStream] Received: " << res_.data() << std::endl;
                    reader_->Read(&res_, this);
                }
                else
                {
                    status_ = FINISH;
                    reader_->Finish(&rpc_status_, this);
                }
            }
            else if (status_ == FINISH)
            {
                if (rpc_status_.ok())
                {
                    std::cout << "[ServerStream] Completed" << std::endl;
                }
                else
                {
                    std::cout << "[ServerStream] Failed: " << rpc_status_.error_message() << std::endl;
                }
                delete this;
            }
        }

    private:
        std::unique_ptr<grpc::ClientAsyncReader<Response>> reader_;
    };

    // 3. 客户端流式RPC：修改构造函数，新增ClientImpl*参数
    class ClientStreamCallData : public CallData
    {
    public:
        ClientStreamCallData(ExampleService::Stub *stub, CompletionQueue *cq, ClientImpl* client, const std::vector<std::string> &msgs)
            : CallData(stub, cq, client), send_msgs_(msgs), send_idx_(0)
        {
            Proceed(true);
        }

        void Proceed(bool ok) override
        {
           

            if (status_ == CREATE)
            {
                status_ = PROCESS;
                writer_ = stub_->PrepareAsyncClientStream(&ctx_, &res_, cq_);
                writer_->StartCall(this);
            }
            else if (status_ == PROCESS)
            {
                status_ = WRITING;
                req_.set_data(send_msgs_[send_idx_++]);
                writer_->Write(req_, this);
            }
            else if (status_ == WRITING)
            {
                if (ok && send_idx_ < send_msgs_.size())
                {
                    req_.set_data(send_msgs_[send_idx_++]);
                    writer_->Write(req_, this);
                }
                else
                {
                    status_ = WRITES_DONE;
                    writer_->WritesDone(this);
                }
            }
            else if (status_ == WRITES_DONE)
            {
                status_ = FINISH;
                writer_->Finish(&rpc_status_, this);
            }
            else if (status_ == FINISH)
            {
                if (rpc_status_.ok())
                {
                    std::cout << "[ClientStream] Received: " << res_.data() << std::endl;
                }
                else
                {
                    std::cout << "[ClientStream] Failed: " << rpc_status_.error_message() << std::endl;
                }
                std::cout << "[ClientStream] Completed" << std::endl;
                delete this;
            }
        }

    private:
        std::unique_ptr<grpc::ClientAsyncWriter<Request>> writer_;
        std::vector<std::string> send_msgs_;
        size_t send_idx_;
    };

    // 4. 双向流式RPC：修改构造函数，新增ClientImpl*参数（保留原有一写一读/全部写完再读逻辑）
    class BidiStreamCallData : public CallData
    {
    public:
        BidiStreamCallData(ExampleService::Stub *stub, CompletionQueue *cq, ClientImpl* client, const std::vector<std::string> &msgs)
            : CallData(stub, cq, client), send_msgs_(msgs), send_idx_(0)
        {
            Proceed(true);
        }
#if 1
        void Proceed(bool ok) override
        {

            if (status_ == CREATE)
            {
                status_ = PROCESS;
                stream_ = stub_->PrepareAsyncBidiStream(&ctx_, cq_);
                stream_->StartCall(this);
            }
            else if (status_ == PROCESS)
            {
                if (ok)
                {
                    // 先发起Write写数据，再Read读响应，打破死锁
                    req_.set_data(send_msgs_[send_idx_++]);
                    stream_->Write(req_, this);
                    status_ = WRITING;
                }
                else
                {
                    stream_->Finish(&rpc_status_, this);
                    status_ = FINISH;
                }
            }
            else if (status_ == WRITING)
            {
                if (ok)
                {
                    stream_->Read(&res_, this);
                    status_ = READING;
                }
                else
                {
                    // 写失败，尝试关闭流
                    stream_->Finish(&rpc_status_, this);
                    status_ = FINISH;
                }
            }
            else if (status_ == READING)
            {
                if (ok)
                {
                    // 读到响应，处理并继续读
                    std::cout << "[BidiStream] Received: " << res_.data() << std::endl;
                    if (send_idx_ < send_msgs_.size())
                    {
                        // 还有数据，继续写
                        req_.set_data(send_msgs_[send_idx_++]);
                        stream_->Write(req_, this);
                        status_ = WRITING;
                    }
                    else
                    {
                        // 写完所有数据，发送WritesDone
                        stream_->WritesDone(this);
                        status_ = WRITES_DONE;
                    }
                }
                else
                {
                    // 关闭流
                    status_ = FINISH;
                    stream_->Finish(&rpc_status_, this);
                }
            }
            else if (status_ == WRITES_DONE)
            {
                if (ok)
                {
                    stream_->Read(&res_, this);
                    status_ = READING;
                }
                else
                {
                    // 写失败，尝试关闭流
                    stream_->Finish(&rpc_status_, this);
                    status_ = FINISH;
                }
            }
            else if (status_ == FINISH)
            {
                if (rpc_status_.ok())
                {
                    std::cout << "[BidiStream] Completed" << std::endl;
                }
                else
                {
                    std::cout << "[BidiStream] Failed: " << rpc_status_.error_message() << std::endl;
                }
                delete this;
            }
        }
#else
        void Proceed(bool ok) override
        {
            if (status_ == CREATE)
            {
                status_ = PROCESS;
                stream_ = stub_->PrepareAsyncBidiStream(&ctx_, cq_);
                stream_->StartCall(this);
            }
            else if (status_ == PROCESS)
            {
                // 先发起Write写数据，再Read读响应，打破死锁
                status_ = WRITING;
                req_.set_data(send_msgs_[send_idx_++]);
                stream_->Write(req_, this);
            }
            else if (status_ == WRITING)
            {
                if (ok)
                {
                    if (send_idx_ < send_msgs_.size())
                    {
                        // 还有数据，继续写
                        req_.set_data(send_msgs_[send_idx_++]);
                        stream_->Write(req_, this);
                    }
                    else
                    {
                        // 写完所有数据，发送WritesDone
                        status_ = WRITES_DONE;
                        stream_->WritesDone(this);
                    }
                }
                else
                {
                    // 关闭流
                    status_ = FINISH;
                    stream_->Finish(&rpc_status_, this);
                    
                }
            }
            else if (status_ == WRITES_DONE)
            {
                // WritesDone完成后，开始读服务端响应
                status_ = READING;
                stream_->Read(&res_, this);
            }
            else if (status_ == READING)
            {
                if (ok)
                {
                    // 读到响应，处理并继续读
                    std::cout << "[BidiStream] Received: " << res_.data() << std::endl;
                    res_.clear_data();
                    stream_->Read(&res_, this);
                }
                else
                {
                    // 关闭流
                    status_ = FINISH;
                    stream_->Finish(&rpc_status_, this);
                }
            }
            else if (status_ == FINISH)
            {
                if (rpc_status_.ok())
                {
                    std::cout << "[BidiStream] Completed" << std::endl;
                }
                else
                {
                    std::cout << "[BidiStream] Failed: " << rpc_status_.error_message() << std::endl;
                }
                delete this;
            }
        }
#endif

    private:
        std::unique_ptr<grpc::ClientAsyncReaderWriter<Request, Response>> stream_;
        std::vector<std::string> send_msgs_;
        size_t send_idx_;
    };

    // 改造6：核心工具函数 - 线程安全的轮询获取下一个CQ
    CompletionQueue* GetNextCQ() {
        int idx = next_cq_idx_++ % num_cqs_;
        return cqs_[idx].get();
    }

    // 改造7：单个CQ的事件循环（接收CQ指针，对应服务端的HandleRpcs）
    void HandleEvents(CompletionQueue* cq)
    {
        void *tag;
        bool ok;
        while (cq->Next(&tag, &ok))
        {
            static_cast<CallData *>(tag)->Proceed(ok);
        }
        std::cout << "CQ thread exited (CQ ptr: " << cq << ")" << std::endl;
    }

    // 改造8：成员变量替换为多CQ相关
    std::unique_ptr<ExampleService::Stub> stub_;
    int num_cqs_;                                  // CQ数量
    std::atomic<int> next_cq_idx_{0};              // 轮询索引（原子变量保证线程安全）
    std::vector<std::unique_ptr<CompletionQueue>> cqs_; // 多个CQ对象
    std::vector<std::thread> cq_threads_;          // 每个CQ对应的事件循环线程
};

int main(int argc, char **argv)
{
    // 改造9：创建客户端时指定CQ数量（默认用硬件并发数，也可手动指定如2/4）
    ClientImpl client(grpc::CreateChannel(
        "localhost:50051", grpc::InsecureChannelCredentials()),
        std::thread::hardware_concurrency()); // 可选：改为固定值如4 → , 4

    // 发起RPC调用（自动分配到不同CQ）
    client.AsyncUnary("Hello Async");
    client.AsyncServerStream("Hi Stream");
    client.AsyncClientStream({"A", "B", "C"});
    client.AsyncBidiStream({"X", "Y", "Z"});

    std::cout << "Press control-c to quit" << std::endl
              << std::endl;
    // 阻塞主线程
    while (true)
    {
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }

    return 0;
}

多 CQ 与单 CQ 异步客户端的编写差别

1. 成员变量

   • 单 CQ：单个 std::unique_ptr<CompletionQueue> cq_、std::thread cq_thread_
   • 多 CQ：std::vector 存储多个 CQ、线程，新增 num_cqs_、原子索引 next_cq_idx_

2. 构造函数

   • 多 CQ：循环创建多个 CQ，为每个 CQ 启动独立线程运行 HandleEvents

3. 析构函数

   • 多 CQ：循环 Shutdown 所有 CQ，Join 所有线程

4. 对外 RPC 接口

   • 多 CQ：通过 GetNextCQ() 轮询获取下一个 CQ 传给 CallData

5. CallData 基类

   • 多 CQ：新增 ClientImpl* client_ 成员，构造函数需传入该指针

6. 派生类构造函数

   • 多 CQ：新增 ClientImpl* 参数

7. 新增工具函数

   • 多 CQ：GetNextCQ()，线程安全地轮询返回 CQ

8. HandleEvents() 函数

   • 多 CQ：接受 CompletionQueue* 作为参数，每个线程处理专属 CQ

9. main 函数

   • 多 CQ：创建客户端时可指定 CQ 数量（默认硬件并发数）

三、客户端异步的另一种写法（不推荐，不易拓展和维护）

1. 单cq

/*
 * 单cq异步客户端：非oop，采用枚举区分不同的rpc。不推荐使用，不容易拓展
 */

#include <iostream>
#include <memory>
#include <string>
#include <thread>
#include <vector>

#include <grpc/support/log.h>
#include <grpcpp/grpcpp.h>

#ifdef BAZEL_BUILD
#include "examples/protos/example.grpc.pb.h"
#else
#include "./build/example.grpc.pb.h"
#endif

using example::ExampleService;
using example::Request;
using example::Response;
using grpc::Channel;
using grpc::ClientContext;
using grpc::CompletionQueue;
using grpc::Status;

// 枚举：标记异步操作的类型，精准处理每一个事件
enum OpType
{
  OP_START_CALL,
  OP_READ,
  OP_WRITE,
  OP_WRITES_DONE,
  OP_FINISH
};

// 枚举：标记RPC类型
enum RpcType
{
  RPC_UNARY,
  RPC_SERVER_STREAM,
  RPC_CLIENT_STREAM,
  RPC_BIDI_STREAM
};

// 统一异步调用结构体，无任何虚函数，彻底杜绝纯虚调用
struct AsyncCall
{
  RpcType rpc_type;
  OpType current_op;

  Request req;
  Response res;
  ClientContext ctx;
  Status rpc_status;

  // 不同RPC的读写器
  std::unique_ptr<grpc::ClientAsyncResponseReader<Response>> unary_reader;
  std::unique_ptr<grpc::ClientAsyncReader<Response>> server_stream_reader;
  std::unique_ptr<grpc::ClientAsyncWriter<Request>> client_stream_writer;
  std::unique_ptr<grpc::ClientAsyncReaderWriter<Request, Response>> bidi_stream;

  // 流式消息存储
  std::vector<std::string> send_msgs;
  size_t send_idx = 0;
  bool writes_done = false;
  bool read_done = false;
};

class ExampleClient
{
public:
  explicit ExampleClient(std::shared_ptr<Channel> channel)
      : stub_(ExampleService::NewStub(channel)) {}

  // 1. 一元RPC：完全对齐官方示例
  void AsyncUnary(const std::string &msg)
  {
    AsyncCall *call = new AsyncCall();
    call->rpc_type = RPC_UNARY;
    call->req.set_data(msg);

    // 严格官方三步法，唯一异步操作是Finish
    call->unary_reader = stub_->PrepareAsyncUnaryCall(&call->ctx, call->req, &cq_);
    call->unary_reader->StartCall();
    call->unary_reader->Finish(&call->res, &call->rpc_status, call);
  }

  // 2. 服务端流式RPC
  void AsyncServerStream(const std::string &msg)
  {
    AsyncCall *call = new AsyncCall();
    call->rpc_type = RPC_SERVER_STREAM;
    call->current_op = OP_START_CALL;
    call->req.set_data(msg);

    call->server_stream_reader = stub_->PrepareAsyncServerStream(&call->ctx, call->req, &cq_);
    // 第一个异步操作：StartCall，tag=call
    call->server_stream_reader->StartCall(call);
  }

  // 3. 客户端流式RPC
  void AsyncClientStream(const std::vector<std::string> &msgs)
  {
    AsyncCall *call = new AsyncCall();
    call->rpc_type = RPC_CLIENT_STREAM;
    call->current_op = OP_START_CALL;
    call->send_msgs = msgs;

    call->client_stream_writer = stub_->PrepareAsyncClientStream(&call->ctx, &call->res, &cq_);
    // 第一个异步操作：StartCall，tag=call
    call->client_stream_writer->StartCall(call);
  }

  // 4. 双向流式RPC
  void AsyncBidiStream(const std::vector<std::string> &msgs)
  {
    AsyncCall *call = new AsyncCall();
    call->rpc_type = RPC_BIDI_STREAM;
    call->current_op = OP_START_CALL;
    call->send_msgs = msgs;

    call->bidi_stream = stub_->PrepareAsyncBidiStream(&call->ctx, &cq_);
    // 第一个异步操作：StartCall，tag=call
    call->bidi_stream->StartCall(call);
  }

  // 事件循环：严格处理每一个异步操作的事件，绝对不提前释放对象
  void AsyncEventLoop()
  {
    void *tag;
    bool ok;

    while (cq_.Next(&tag, &ok))
    {
      AsyncCall *call = static_cast<AsyncCall *>(tag);

      // ========== 一元RPC处理 ==========
      if (call->rpc_type == RPC_UNARY)
      {
        GPR_ASSERT(ok);
        if (call->rpc_status.ok())
        {
          std::cout << "[Unary] Received: " << call->res.data() << std::endl;
        }
        else
        {
          std::cout << "[Unary] Failed: " << call->rpc_status.error_message() << std::endl;
        }
        delete call;
        continue;
      }

      // ========== 服务端流式RPC处理 ==========
      if (call->rpc_type == RPC_SERVER_STREAM)
      {
        switch (call->current_op)
        {
        case OP_START_CALL:
          if (!ok)
          {
            std::cout << "[ServerStream] StartCall failed" << std::endl;
            delete call;
            continue;
          }
          // StartCall完成，发起第一次Read
          call->current_op = OP_READ;
          call->server_stream_reader->Read(&call->res, call);
          break;

        case OP_READ:
          if (ok)
          {
            // 收到数据，打印后继续读
            std::cout << "[ServerStream] Received: " << call->res.data() << std::endl;
            call->server_stream_reader->Read(&call->res, call);
          }
          else
          {
            // 流结束，发起Finish
            call->current_op = OP_FINISH;
            call->server_stream_reader->Finish(&call->rpc_status, call);
          }
          break;

        case OP_FINISH:
          // Finish完成，释放对象
          if (call->rpc_status.ok())
          {
            std::cout << "[ServerStream] Completed" << std::endl;
          }
          else
          {
            std::cout << "[ServerStream] Failed: " << call->rpc_status.error_message() << std::endl;
          }
          delete call;
          break;

        default:
          delete call;
          break;
        }
        continue;
      }

      // ========== 客户端流式RPC处理 ==========
      if (call->rpc_type == RPC_CLIENT_STREAM)
      {
        switch (call->current_op)
        {
        case OP_START_CALL:
          if (!ok)
          {
            std::cout << "[ClientStream] StartCall failed" << std::endl;
            delete call;
            continue;
          }
          // StartCall完成，发起第一次Write

          call->req.set_data(call->send_msgs[call->send_idx++]);
          call->client_stream_writer->Write(call->req, call);
          call->current_op = OP_WRITE;
          break;

        case OP_WRITE:
          if (ok && call->send_idx < call->send_msgs.size())
          {
            // 上一次Write完成，继续写
            call->req.set_data(call->send_msgs[call->send_idx++]);
            call->client_stream_writer->Write(call->req, call);
          }
          else
          {
            // 所有数据写完，发起WritesDone
            call->current_op = OP_WRITES_DONE;
            call->client_stream_writer->WritesDone(call);
          }
          break;

        case OP_WRITES_DONE:
          // WritesDone完成，发起Finish
          call->current_op = OP_FINISH;
          call->client_stream_writer->Finish(&call->rpc_status, call);
          break;

        case OP_FINISH:
          // Finish完成，打印结果、释放对象
          if (call->rpc_status.ok())
          {
            std::cout << "[ClientStream] Received: " << call->res.data() << std::endl;
          }
          else
          {
            std::cout << "[ClientStream] Failed: " << call->rpc_status.error_message() << std::endl;
          }
          std::cout << "[ClientStream]: complete" << std::endl;
          delete call;
          break;

        default:
          delete call;
          break;
        }
        continue;
      }

      // ========== 双向流式RPC处理 ==========
      if (call->rpc_type == RPC_BIDI_STREAM)
      {
        switch (call->current_op)
        {
        case OP_START_CALL:
          if (!ok)
          {
            std::cout << "[BidiStream] StartCall failed" << std::endl;
            delete call;
            continue;
          }

          call->req.set_data(call->send_msgs[call->send_idx++]);
          call->bidi_stream->Write(call->req, call);
          call->current_op = OP_WRITE;
          break;

        case OP_READ:

          if (ok)
          {
            std::cout << "[BidiStream] Received: " << call->res.data() << std::endl;
            if (call->send_idx < call->send_msgs.size())
            {         
              // 上一次Write完成，继续写
              call->req.set_data(call->send_msgs[call->send_idx++]);
              call->bidi_stream->Write(call->req, call);
              call->current_op = OP_WRITE;
            }
            else
            {
              // 所有数据写完，发起WritesDone
              call->current_op = OP_WRITES_DONE;
              call->bidi_stream->WritesDone(call);
            }
          }

          else
          {
            call->current_op = OP_FINISH;
            call->bidi_stream->Finish(&call->rpc_status, call);
          }

          break;

        case OP_WRITE:

          if (ok)
          {

            call->bidi_stream->Read(&call->res, call);
            call->current_op = OP_READ;
          }
          else
          {
            // 读结束，标记完成
            call->current_op = OP_FINISH;
            call->bidi_stream->Finish(&call->rpc_status, call);
          }

          break;

        case OP_WRITES_DONE:
          if (ok)
          {
            call->current_op = OP_READ;
            call->bidi_stream->Read(&call->res, call);
          }
          else
          {
            call->current_op = OP_FINISH;
            call->bidi_stream->Finish(&call->rpc_status, call);
          }

          break;

        case OP_FINISH:
          // Finish完成，释放对象
          if (call->rpc_status.ok())
          {
            std::cout << "[BidiStream] Completed" << std::endl;
          }
          else
          {
            std::cout << "[BidiStream] Failed: " << call->rpc_status.error_message() << std::endl;
          }
          delete call;
          break;

        default:
          delete call;
          break;
        }
        continue;
      }

      // 异常情况，安全释放
      delete call;
    }
  }

  void Shutdown()
  {
    cq_.Shutdown();
  }

private:
  std::unique_ptr<ExampleService::Stub> stub_;
  CompletionQueue cq_;
};

int main(int argc, char **argv)
{
  ExampleClient client(grpc::CreateChannel(
      "localhost:50051", grpc::InsecureChannelCredentials()));

  // 独立线程运行事件循环
  std::thread event_thread = std::thread(&ExampleClient::AsyncEventLoop, &client);

  // 发起所有RPC调用
  client.AsyncUnary("Hello Async");
  client.AsyncServerStream("Hi Stream");
  client.AsyncClientStream({"A", "B", "C"});
  client.AsyncBidiStream({"X", "Y", "Z"});

  std::cout << "Press control-c to quit" << std::endl
            << std::endl;
  event_thread.join();
  client.Shutdown();

  return 0;
}

2. 非 OOP 异步客户端代码编写流程

1. 定义枚举

   • OpType：标记异步操作类型（START_CALL/READ/WRITE/WRITES_DONE/FINISH）。
   • RpcType：标记 RPC 类型（UNARY/SERVER_STREAM/CLIENT_STREAM/BIDI_STREAM）。

2. 定义统一结构体 AsyncCall

   • 包含所有 RPC 类型的公共成员：rpc_type、current_op、req/res/ctx/rpc_status。
   • 包含所有 RPC 类型的读写器（unary_reader/server_stream_reader/client_stream_writer/bidi_stream）。
   • 包含流式 RPC 所需的辅助成员（send_msgs/send_idx/writes_done/read_done）。

3. 对外 RPC 发起接口

   • new AsyncCall()，设置 rpc_type 和业务参数。
   • 调用 PrepareAsyncXXX 准备 RPC。
   • 调用 StartCall() 或 Finish(..., this)，传入 Tag (this)。

4. 事件循环（核心）

   • while (cq_.Next(&tag, &ok)) —— CQ 阻塞等待事件。
   • static_cast<AsyncCall*>(tag) —— 通过 Tag 找回结构体。
   • 大 switch/case：先判断 rpc_type，再判断 current_op，处理对应逻辑并发起下一个异步操作（传入 this）。
   • 处理完 FINISH 后 delete call。

5. main 函数

   • 创建客户端对象，启动独立线程运行事件循环。
   • 调用对外接口发起 RPC。
   • Join 线程并 Shutdown。

3. 与 OOP 写法对比

维度	非 OOP 写法（枚举区分）	OOP 写法（继承 CallData）
可扩展性	❌ 差。新增 RPC 类型需修改：1. 枚举 RpcType2. 结构体 AsyncCall3. 事件循环的大 switch/case	✅ 好。新增 RPC 类型只需：新增一个继承 CallData 的派生类，实现 Proceed()
代码可读性	❌ 差。所有逻辑挤在一个大 switch/case 里，结构体包含所有类型的成员（大部分时间是冗余的）	✅ 好。每个 RPC 类型逻辑独立在自己的派生类里，成员按需定义
维护成本	❌ 高。修改一个 RPC 类型的逻辑，容易误改其他类型的代码	✅ 低。各派生类独立，修改互不影响
性能	⚠️ 几乎无差异。但大 switch/case 可能有微小的分支预测开销	⚠️ 几乎无差异。虚函数调用开销可忽略
推荐度	❌ 不推荐。仅适合简单、固定的场景	✅ 强烈推荐。符合代码规范，易于扩展和维护

OOP 写法（继承 CallData）更好。它通过多态将不同 RPC 类型的逻辑解耦，极大提升了代码的可扩展性和可维护性。