gRPC C 自定义 EventEngine 实战为 gRPC 注入应用自有的 I/O 与异步执行引擎【免费下载链接】grpcC based gRPC (C, Python, Ruby, Objective-C, PHP, C#)项目地址: https://gitcode.com/GitHub_Trending/gr/grpcgRPC 的 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:50051BUILDBazel 构建脚本示例的通信协议使用标准的 helloworld proto定义见 examples/protos/helloworld.proto。服务端与客户端通过回调Callback方式完成一次一元 RPC。什么是 EventEnginegRPC 的底层平台抽象层官方接口文档位于 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 的三个典型动机场景对接外部事件循环例如希望把 gRPC 的 I/O 事件并入应用自有的 epoll、libuv 或 UI 事件循环按 Channel 或 Server 隔离 I/O为不同通道使用不同的 EventEngine 实例把网络 I/O 与回调处理在多个通道之间相互隔离在 gRPC 内置引擎不适合的平台上提供自有网络栈。EventEngine 覆盖的核心能力清单围绕接口文档一个完整 EventEngine 实现需要覆盖下列抽象均定义于 include/grpc/event_engine/event_engine.h异步任务执行Run(Closure*)与Run(absl::AnyInvocablevoid())在合适时机“尽快”执行闭包。两种重载的归属权语义不同Closure*版本的所有权永远在调用方引擎执行完不会删除闭包而AnyInvocable版本执行完毕后由引擎负责销毁。定时器RunAfter(Duration, ...)用于在指定延时后执行回调返回的TaskHandle可通过Cancel(TaskHandle)取消Duration定义为纳秒精度的std::chrono::durationint64_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 时必须遵守的期望编写自定义引擎前务必理解自备 I/O 线程引擎必须内部创建执行 I/O 与回调所需的线程例如可拆分“轮询线程池”与“回调执行线程池”。通过 Slice 分配数据缓冲引擎内部的读写缓冲区内存应通过SliceAllocator分配为 Slice 形式使 gRPC 的ResourceQuota内存配额系统能在应用设定阈值下回收内存、优雅降级。明确回调执行语义部分回调可能较昂贵引擎应确定并文档化回调执行是否会阻塞轮询便于上层决定是否把重回调搬运到独立线程。支持并发调用gRPC 可能跨多个线程并发使用同一 EventEngine所有方法须保证线程安全。此外 include/grpc/event_engine/event_engine.h 特别提示不要在 EventEngine 回调里做大量阻塞性工作——内置实现扩容线程池需要时间而用户自研引擎可能完全不具备抗饥饿能力。另有两点实现纪律需要遵守每个Endpoint同一时刻至多允许一个未完成的Read或Write违规调用引擎必须 abort引擎析构时不得存在未完成任务、活跃监听器或端点否则属于非法使用undefined behavior责任在应用侧。示例核心包装型WrappingEventEnginewrapping_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::AnyInvocablevoid() closure) override { run_count_; wrapped_engine_-Run(std::move(closure)); } int get_run_count() { return run_count_.load(); } // 其余全部为透传passthrough…… private: std::shared_ptrEventEngine wrapped_engine_; std::atomicint 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::AnyInvocablevoid()引擎代管销毁语义不同派生类必须分别实现。示例在两处都执行run_count_后转发给内部引擎。计数用原子变量std::atomicint 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_sharedmy_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 applications 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.hSetDefaultEventEngine(std::shared_ptrEventEngine)把自定义实例设为 gRPC 全库默认引擎。gRPC 会持有该引擎一个引用外加应用自己持有的一个共两个引用若传入nullptrgRPC 只丢弃手中引用而不设置新值。先析构 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.hclass DefaultEventEngineScope { public: explicit DefaultEventEngineScope(std::shared_ptrEventEngine 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提供SayHelloclass 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::mutexstd::condition_variabledone标志让主线程阻塞等待完成回调收到Status后校验status.ok()并输出reply.message()。地址由--target标志指定默认localhost:50051greeter_callback_client.cc。值得留意的是客户端代码中的作用域设计——从创建 Channel、发起 RPC 到打印回复都在一个匿名块内完成这正是为了在调用ShutdownDefaultEventEngine()前确保所有 gRPC 对象尤其是引用着 EventEngine 的通道与调用栈先被销毁。构建与运行示例的 Bazel 构建配置见 BUILDgreeter_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.hwrapping_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 --targetlocalhost: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 --port50052配合客户端--targetlocalhost:50052。源码路径速查示例 READMEexamples/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.ccEventEngine 公共接口与注入/关闭函数声明include/grpc/event_engine/event_engine.hEventEngine 设计说明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/BUILDproto 定义examples/protos/helloworld.proto【免费下载链接】grpcC based gRPC (C, Python, Ruby, Objective-C, PHP, C#)项目地址: https://gitcode.com/GitHub_Trending/gr/grpc创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考