gRPC C++ 自定义 EventEngine 实战：为 gRPC 注入应用自有的 I/O 与异步执行引擎

gRPC 的 EventEngine 抽象了全部跨平台底层能力——网络 I/O、定时器、异步任务执行与 DNS 解析。本指南以仓库中的 default_event_engine 示例 为核心骨架，讲解如何在 C++ 应用中实现并注入一个“应用自有”的 EventEngine，以实现对外部事件循环的对接、按通道隔离网络 I/O 等能力。读完本文，你将掌握 EventEngine 接口的全部职责、包装型自定义引擎的写法、SetDefaultEventEngine / ShutdownDefaultEventEngine 的注入与生命周期管理，以及完整的构建与运行流程。

示例概览与项目位置

示例位于仓库的 examples/cpp/default_event_engine 目录，文件构成如下：

文件	作用
wrapping_event_engine.h	自定义 EventEngine 实现：统计 Run 被调用次数，其余所有调用透传给内部 gRPC 默认引擎
greeter_callback_client.cc	基于 Callback API 的异步客户端，在启动时注入上述自定义引擎，并在退出前演示引擎的关闭流程
greeter_callback_server.cc	常规的 Callback 风格 Greeter 服务端（默认监听 0.0.0.0:50051）
BUILD	Bazel 构建脚本

示例的通信协议使用标准的 helloworld proto，定义见 examples/protos/helloworld.proto。服务端与客户端通过回调（Callback）方式完成一次一元 RPC。

什么是 EventEngine：gRPC 的底层平台抽象层

官方接口文档位于 include/grpc/event_engine/README.md，其中明确指出：

An EventEngine handles all cross-platform I/O, task execution, and DNS resolution for gRPC.

也就是说，EventEngine 把原先分散在各平台实现中的底层网络 I/O、异步回调执行、定时器（Timer）与 DNS 解析全部收敛到一组纯 C++ 接口后面。gRPC 自身附带一个跨平台的默认实现，而该接口存在的意义在于：允许外部集成方“自带干粮”，替换掉这套底层机制。

从接口头文件 include/grpc/event_engine/event_engine.h 的注释可以归纳出自定义 EventEngine 的三个典型动机场景：

1. 对接外部事件循环：例如希望把 gRPC 的 I/O 事件并入应用自有的 epoll、libuv 或 UI 事件循环；
2. 按 Channel 或 Server 隔离 I/O：为不同通道使用不同的 EventEngine 实例，把网络 I/O 与回调处理在多个通道之间相互隔离；
3. 在 gRPC 内置引擎不适合的平台上提供自有网络栈。

EventEngine 覆盖的核心能力清单

围绕接口文档，一个完整 EventEngine 实现需要覆盖下列抽象（均定义于 include/grpc/event_engine/event_engine.h）：

• 异步任务执行：Run(Closure*) 与 Run(absl::AnyInvocable<void()>)，在合适时机“尽快”执行闭包。两种重载的归属权语义不同：Closure* 版本的所有权永远在调用方，引擎执行完不会删除闭包；而 AnyInvocable 版本执行完毕后由引擎负责销毁。
• 定时器：RunAfter(Duration, ...) 用于在指定延时后执行回调，返回的 TaskHandle 可通过 Cancel(TaskHandle) 取消；Duration 定义为纳秒精度的 std::chrono::duration<int64_t, std::nano>。
• 网络连接：Connect(...) 发起客户端连接，CancelConnect(ConnectionHandle) 取消连接尝试，成功连接后通过 OnConnectCallback 回调交出 Endpoint。
• 网络监听：CreateListener(...) 创建服务端监听器，返回的 Listener 支持在 Start() 前绑定多个地址/端口（Bind），每建立一个连接就异步触发一次 AcceptCallback。
• DNS 解析：GetDNSResolver(...) 返回 DNSResolver，提供 LookupHostname、LookupSRV、LookupTXT 三类异步查询；ResolverOptions::dns_server 可指定自定义 DNS 服务器（IP:port 格式，为空则使用系统默认）。
• 线程工具：IsWorkerThread() 用于判断当前线程是否为引擎的工作线程。

对实现者的四条硬性期望

同一文档还给出实现 EventEngine 时必须遵守的期望，编写自定义引擎前务必理解：

1. 自备 I/O 线程：引擎必须内部创建执行 I/O 与回调所需的线程，例如可拆分“轮询线程池”与“回调执行线程池”。
2. 通过 Slice 分配数据缓冲：引擎内部的读写缓冲区内存应通过 SliceAllocator 分配为 Slice 形式，使 gRPC 的 ResourceQuota 内存配额系统能在应用设定阈值下回收内存、优雅降级。
3. 明确回调执行语义：部分回调可能较昂贵，引擎应确定并文档化回调执行是否会阻塞轮询，便于上层决定是否把重回调搬运到独立线程。
4. 支持并发调用：gRPC 可能跨多个线程并发使用同一 EventEngine，所有方法须保证线程安全。

此外 include/grpc/event_engine/event_engine.h 特别提示：不要在 EventEngine 回调里做大量阻塞性工作——内置实现扩容线程池需要时间，而用户自研引擎可能完全不具备抗饥饿能力。另有两点实现纪律需要遵守：

• 每个 Endpoint 同一时刻至多允许一个未完成的 Read 或 Write，违规调用引擎必须 abort；
• 引擎析构时不得存在未完成任务、活跃监听器或端点，否则属于非法使用（undefined behavior），责任在应用侧。

示例核心：包装型（Wrapping）EventEngine

wrapping_event_engine.h 展示了在不重写整个网络栈的前提下“接管 gRPC 异步执行面”的最简可行路径：继承 EventEngine，计数后把一切调用委托给一个内部持有的 gRPC 默认引擎实例。

// 见 examples/cpp/default_event_engine/wrapping_event_engine.h
#include <grpc/event_engine/event_engine.h>

namespace my_application {

class WrappingEventEngine
    : public grpc_event_engine::experimental::EventEngine {
 public:
  WrappingEventEngine()
      : wrapped_engine_(grpc_event_engine::experimental::CreateEventEngine()) {}
  ~WrappingEventEngine() override = default;

  // 唯一“被自定义”的行为：统计 Run 调用次数
  void Run(Closure* closure) override {
    run_count_++;
    wrapped_engine_->Run(closure);
  }

  void Run(absl::AnyInvocable<void()> closure) override {
    run_count_++;
    wrapped_engine_->Run(std::move(closure));
  }

  int get_run_count() { return run_count_.load(); }

  // 其余全部为透传（passthrough）……
 private:
  std::shared_ptr<EventEngine> wrapped_engine_;
  std::atomic<int> run_count_{0};
};
}  // namespace my_application

要点拆解：

• 命名空间与类型：公共接口位于 grpc_event_engine::experimental 命名空间——experimental 前缀意味着该 API 仍处于演进期，未来可能调整。
• 引擎创建：构造函数调用自由函数 CreateEventEngine()（声明见 include/grpc/event_engine/event_engine.h），得到 gRPC 内置默认引擎，并以 std::shared_ptr 持有。gRPC 本身也通过 std::shared_ptr 共享持有 EventEngine，保证引擎在其仍被使用时始终存活。
• 两次 Run 重载都需覆盖：Closure*（非拷贝接口类，所有权在调用方）与 absl::AnyInvocable<void()>（引擎代管销毁）语义不同，派生类必须分别实现。示例在两处都执行 run_count_++ 后转发给内部引擎。
• 计数用原子变量：std::atomic<int> run_count_{0}——因为引擎方法可能被多个线程并发调用。
• 透传并非复制粘贴无意义代码：它保住了完整接口契约。透传清单覆盖 CreateListener、Connect、CancelConnect、IsWorkerThread、GetDNSResolver、RunAfter（两种重载）、Cancel。这正是“在既有引擎之上做切面/观测/限量”的通用模式——真实产品中你可以在此插入指标采集、线程模型改造或流量控制逻辑。

注入引擎：SetDefaultEventEngine 的两种用法

示例采用的方式：设置全局默认引擎

客户端在 main() 入口处完成注入（见 greeter_callback_client.cc）：

// Create some EventEngine of your choosing, likely your own.
auto custom_engine = std::make_shared<my_application::WrappingEventEngine>();
// Provide this engine to gRPC. Now there are 2 refs to this engine: one here,
// and one owned by gRPC.
grpc_event_engine::experimental::SetDefaultEventEngine(custom_engine);
{
  // gRPC 对象（Channel、Client 等）在作用域内创建并使用
  std::string target_str = absl::GetFlag(FLAGS_target);
  my_application::GreeterClient greeter(
      grpc::CreateChannel(target_str, grpc::InsecureChannelCredentials()));
  std::string reply = greeter.SayHello("EventEngine");
  std::cout << "Greeter received: " << reply << std::endl;
}
LOG(INFO) << "My EventEngine ran " << custom_engine->get_run_count()
          << " closures";
// Release the application's ownership of the EventEngine.
custom_engine.reset();
// Block until gRPC is done using the engine, and the engine is destroyed.
grpc_event_engine::experimental::ShutdownDefaultEventEngine();

这段代码完整演示了官方要求的生命周期纪律，语义逐条对应接口注释（include/grpc/event_engine/event_engine.h）：

• SetDefaultEventEngine(std::shared_ptr<EventEngine>)：把自定义实例设为 gRPC 全库默认引擎。gRPC 会持有该引擎一个引用（外加应用自己持有的一个，共两个引用）；若传入 nullptr，gRPC 只丢弃手中引用而不设置新值。
• 先析构 gRPC 对象，再关引擎：客户端、Channel 等对象被放在独立作用域内，保证它们在 ShutdownDefaultEventEngine() 之前析构、释放引用——避免“引擎还欠着活跃 I/O/任务就析构”的未定义行为。
• custom_engine.reset()：释放应用侧的引用，此刻引擎仅由 gRPC 持有。
• ShutdownDefaultEventEngine()：让 gRPC 回落到内置默认引擎（对所有新的 GetDefaultEventEngine() 请求），并阻塞直至当前默认引擎的全部引用被释放（引擎被销毁）。官方注释强调：凡调用过 SetDefaultEventEngine 的程序，必须在结束时调用 ShutdownDefaultEventEngine 或 SetDefaultEventEngine(nullptr)，否则自定义引擎永远不会被销毁；若想不等待旧引擎引用释放就立即回落内置引擎，则应改用 SetDefaultEventEngine(nullptr)。
• 验证手段：通过 custom_engine->get_run_count() 打印自定义引擎累计执行的闭包数量——由于 RPC 的异步回调与 I/O 驱动都会经由引擎，一次成功的 SayHello 会观察到非零的计数，直观证明引擎确实被 gRPC 使用。若把打印挪到 ShutdownDefaultEventEngine() 之后读取，读到的是已被销毁对象的悬垂访问——这正是注释提醒“先在作用域内统计、后关闭”的原因。

注：示例是整个进程范围内注入同一默认引擎的方式。若读者需要更细粒度控制，接口头文件还给出了按 Channel/Server 注入的示例（含 ChannelArguments::SetEventEngine / ServerBuilder::SetEventEngine 用法）。不过从源码结构看，这部分按通道注入的 API 在示例中标注为 Not yet implemented（见 include/grpc/event_engine/event_engine.h），实践中以全局默认引擎注入为主。

历史接口与一次性作用域

头文件中还保留了已废弃（DEPRECATED）的工厂函数 SetEventEngineFactory / EventEngineFactoryReset——它们通过“工厂回调”在需要时创建引擎，官方标注将随全部已知用户迁移完毕而移除，新代码不应再使用。仓库内部则提供了 RAII 形式的便捷封装 src/core/lib/event_engine/default_event_engine.h：

class DefaultEventEngineScope {
 public:
  explicit DefaultEventEngineScope(std::shared_ptr<EventEngine> engine) {
    SetDefaultEventEngine(std::move(engine));
  }
  ~DefaultEventEngineScope() { ShutdownDefaultEventEngine(); }
};

它把 SetDefaultEventEngine 与 ShutdownDefaultEventEngine 配对成作用域守卫，在测试代码与希望临时替换默认引擎的场景中可直接套用。同样位于 src/core/lib/event_engine 目录的还有 RegisterEventEngineChannelArgPreconditioning，它负责在 Channel 参数进入 gRPC 前做“预置”：确保每个通道参数中都预置好默认 EventEngine 实例。

客户端与服务端的 RPC 流程

注入引擎后的通信本身仍是标准的 helloworld 模式，服务端无需任何 EventEngine 感知（默认引擎也会被全局替换为自定义实例）。服务端实现见 greeter_callback_server.cc，它注册了健康检查服务与反射插件，并用 CallbackService 提供 SayHello：

class GreeterServiceImpl final : public Greeter::CallbackService {
  ServerUnaryReactor* SayHello(CallbackServerContext* context,
                               const HelloRequest* request,
                               HelloReply* reply) override {
    std::string prefix("Hello ");
    reply->set_message(prefix + request->name());
    ServerUnaryReactor* reactor = context->DefaultReactor();
    reactor->Finish(Status::OK);
    return reactor;
  }
};

服务端通过 ServerBuilder 监听 0.0.0.0:PORT（端口由 --port 标志控制，默认 50051，见 greeter_callback_server.cc）。

客户端则展示了 Callback 风格异步一元 RPC 的写法（greeter_callback_client.cc）：通过 stub_->async()->SayHello(...) 发起请求，以 std::mutex + std::condition_variable + done 标志让主线程阻塞等待完成回调，收到 Status 后校验 status.ok() 并输出 reply.message()。地址由 --target 标志指定，默认 localhost:50051（greeter_callback_client.cc）。

值得留意的是客户端代码中的作用域设计——从创建 Channel、发起 RPC 到打印回复都在一个匿名块内完成，这正是为了在调用 ShutdownDefaultEventEngine() 前确保所有 gRPC 对象（尤其是引用着 EventEngine 的通道与调用栈）先被销毁。

构建与运行

示例的 Bazel 构建配置见 BUILD：greeter_callback_client 与 greeter_callback_server 两个 cc_binary 均依赖 //:grpc++ 与 //examples/protos:helloworld_cc_grpc（由 helloworld.proto 生成的代码），并通过 defines = ["BAZEL_BUILD"] 使源码走 #ifdef BAZEL_BUILD 分支、包含 examples/protos/helloworld.grpc.pb.h；wrapping_event_engine 被打包为独立 cc_library 供客户端链接。非 Bazel 构建（如 CMake）下，预生成的 proto 头文件位于构建输出目录，对应源码中的 #else 分支包含路径。

构建与运行的完整步骤在官方 C++ Quick Start 文档中有系统说明（本仓库对应指南可见于 examples/cpp/helloworld 与根目录 BUILDING.md），核心流程如下：

# 1. 构建 gRPC 与示例（以 CMake 为例）
#    先按 BUILDING.md 安装依赖并编译安装 gRPC，
#    再对 examples/cpp 目录执行 CMake 配置与构建。

# 2. 终端 A：启动服务端（默认 0.0.0.0:50051）
./greeter_callback_server

# 3. 终端 B：运行注入自定义 EventEngine 的客户端
./greeter_callback_client --target=localhost:50051

预期输出形如：

Greeter received: Hello EventEngine
My EventEngine ran <N> closures

服务端打印 Server listening on 0.0.0.0:50051。其中第二行来自客户端对 custom_engine->get_run_count() 的统计输出，<N> 为本次 RPC 生命周期内引擎实际执行的闭包数量，是自定义引擎确实生效的直接证据。服务端还可选 --port 参数修改监听端口（例如 ./greeter_callback_server --port=50052 配合客户端 --target=localhost:50052）。

源码路径速查

• 示例 README：examples/cpp/default_event_engine/README.md
• 自定义引擎实现：examples/cpp/default_event_engine/wrapping_event_engine.h
• 客户端（注入与关闭流程）：examples/cpp/default_event_engine/greeter_callback_client.cc
• 服务端：examples/cpp/default_event_engine/greeter_callback_server.cc
• EventEngine 公共接口与注入/关闭函数声明：include/grpc/event_engine/event_engine.h
• EventEngine 设计说明：include/grpc/event_engine/README.md
• 端点配置接口（int/string/void* 三类配置取值）：include/grpc/event_engine/endpoint_config.h
• 默认引擎 RAII 封装与 Channel 参数预置：src/core/lib/event_engine/default_event_engine.h
• 构建配置：examples/cpp/default_event_engine/BUILD
• proto 定义：examples/protos/helloworld.proto