Envoy Mobile Python 库实战用 asyncio 与 httpx 集成 Envoy 高性能网络引擎【免费下载链接】envoyCloud-native high-performance edge/middle/service proxy项目地址: https://gitcode.com/GitHub_Trending/en/envoy导读Envoy Mobile Python 库 为 Python 应用提供了 Envoy 的原生绑定bindings让你不必亲自启动一个独立的 Envoy 进程就能在进程内复用 Envoy 的高性能网络能力——包括 HTTP/2、HTTP/3QUIC、连接池、DNS 缓存与丰富的指标。本文围绕该库的官方 README 展开完整讲解AsyncClient高层异步 API 与直接操作Engine/Stream的底层用法并深入其源码mobile/library/python/envoy_mobile与集成测试mobile/test/python说明请求头规范化、显式流控、httpx transport 适配等实现细节帮助你快速在自己的 Python 项目中落地 Envoy Mobile。功能特性总览按照 README 的描述该库提供四大核心能力高层异步 APIHigh-level Async API基于 Python 标准库asyncio实现AsyncClient提供熟悉的非阻塞 HTTP 请求接口与httpx的调用风格接近Envoy 引擎集成Envoy Engine Integration可直接访问 Envoy 引擎进行高级配置与性能调优连接超时、DNS 刷新、QUIC 提示、gzip/brotli 解压、socket tagging 等现代协议支持Modern Protocols原生支持 HTTP/2 与 HTTP/3QUIC可观测性Observable集成 Envoy 丰富的指标stats与日志logging。从包结构上看envoy_mobile/init.py 对外导出AsyncClient、Engine、EngineBuilder、LogLevel、Stream、StreamPrototype、EnvoyError、ErrorCode、StreamIntel、FinalStreamIntel以及三组 httpx transportAsyncEnvoyClientTransport、EnvoyClientTransport、EnvoyTransportFactory分层清晰envoy_enginepybind11 生成的 C 绑定模块.soasync_client/纯 Python 的高层AsyncClient及其Response、Executor、请求规范化工具*_client_transport.py、transport_factory.py面向 httpx 生态的 transport 适配层。使用 AsyncClient 发起异步请求AsyncClient是官方推荐的入口。它“每个客户端实例独占一个 Envoy 引擎和 executor”request()在流完成后返回Response对象整个操作借助底层asyncio事件循环与AsyncioExecutor完全非阻塞见 client.py 的类文档。最小可用示例README 中的示例完整可运行其核心流程是构造EngineBuilder→ 配置日志级别 → 将 builder 交给AsyncClient作为上下文管理器 → 发起请求 → 读取响应import asyncio from envoy_mobile import AsyncClient, EngineBuilder, LogLevel async def main(): # Configure the engine builder EngineBuilder().set_log_level(LogLevel.info) # Use AsyncClient as a context manager async with AsyncClient(builder) as client: # Make a request response await client.request( methodGET, urlhttps://api.github.com/repos/envoyproxy/envoy-mobile ) print(fStatus: {response.status_code}) body await response.body() print(fBody length: {len(body)}) if __name__ __main__: asyncio.run(main())生命周期与引擎就绪等待AsyncClient作为异步上下文管理器使用时__aenter__内部做了三件事见 client.py创建一个asyncio.Event作为引擎就绪信号并用AsyncioExecutor(loopasyncio.get_running_loop())绑定当前事件循环调用builder.set_on_engine_running(...)注册回调后build()引擎——这是为了让回调能安全地跨线程调度回 asyncio 循环AsyncioExecutor.wrap内部使用loop.call_soon_threadsafe见 executor.pyawait self._engine_running.wait()阻塞直到引擎真正跑起来DNS 等子系统就绪。退出时__aexit__调用engine.terminate()释放资源__del__也做了兜底清理避免引擎泄漏。请求参数与响应读取request(method, url, **kwargs)是全部 HTTP 动词方法get/post/put/delete/head/options/patch/trace的统一实现支持json、data、headers、timeout四个扩展参数client.pyjson序列化为 JSON 字节并自动设置content-type: application/json与data同时传入会抛出ValueErrordata字符串按 UTF-8 编码dict/list 会被urlencode为表单并设置application/x-www-form-urlencoded同时自动计算content-lengthheadersdict 形式值为字符串或字符串列表timeoutint/float/timedelta 均可最终换算为毫秒并写入x-envoy-upstream-rq-timeout-ms头见 utils.py。Response对象response.py暴露了丰富的读取能力response.status_code/response.headers/response.trailers从:status伪头解析状态码头信息由回调逐条填充await response.body读取完整响应体并缓存重复调用直接返回缓存await response.text以 UTF-8 解码响应体当前未根据 charset 头动态解码属 TODO 项await response.json()将响应体解析为 JSONresponse.content.read(n)StreamReader流式读取接口可每次读取 n 字节避免大响应整体缓冲默认_READ_SIZE 1024见 response.pyresponse.ok状态码小于 400 即为真raise_for_status()在非 2xx/3xx 时抛ClientResponseErrorasync with response或response.close()可取消未完成的底层流。Response内部通过attach()注册on_headers/on_data/on_trailers/on_complete/on_error/on_cancel六个回调且开启explicit_flow_controlTrue显式流控即每次通过stream.read_data(n)主动向 Envoy 请求数据块response.py。并发请求同一个AsyncClient可并发发起多个请求。集成测试 async_client_fetch_test.py 展示了用asyncio.gather同时发送 GET/POST 并统一等待响应的写法测试还覆盖了自定义请求头、流式逐字节读取、JSON 请求、json/data冲突报错、404 时raise_for_status等场景可作为最佳实践参考。直接使用 Engine 与 Stream 底层 API对于需要精细控制的高级场景README 提供了绕过AsyncClient、直接与Engine和Stream交互的方式。该方式不依赖 asyncio适合在纯回调/线程模型中使用from envoy_mobile import EngineBuilder, LogLevel def on_data(data, end_stream): print(fReceived data: {data}) builder EngineBuilder().set_log_level(LogLevel.debug) engine builder.build() # Create a stream and send a request stream engine.get_stream_client().new_stream_prototype() \ .on_data(on_data) \ .start() stream.send_headers({method: GET, scheme: https, authority: google.com, path: /}, True)注意当前仓库 API 与 README 示例的差异需要指出的是当前仓库版本的绑定 API 与 README 示例存在细微差异在 module_definition.cc 中StreamClient通过engine.stream_client(listener_name)获取listener_name默认为空字符串流回调通过StreamPrototype.start(on_headers..., on_data..., on_complete..., ...)关键字参数一次性注册Stream提供send_headers(headers, end_stream, idempotentFalse)、send_data(bytes)、close(bytes|dict)、cancel()、read_data(n)方法。集成测试 fetch_test.py 展示了当前仓库中直接流式编程的标准写法stream ( engine.stream_client() .new_stream_prototype() .start( on_headerson_headers, on_dataon_data, on_completeon_complete, on_erroron_error, on_cancelon_cancel, ) ) headers { :method: GET, :scheme: http, :authority: self._echo_server_url, :path: /, } stream.send_headers(headers, end_streamTrue)因此在实际编码时应以本仓库 module_definition.cc 中 pybind11 暴露的签名为准。请求头必须使用 HTTP/2 伪头形式:method、:scheme、:authority、:pathend_streamTrue表示请求侧立即结束。回调与流内省数据流回调携带两类内省对象module_definition.ccStreamIntel流级信息含stream_id、connection_id、attempt_count重试次数、consumed_bytes_from_response已消费的响应字节数FinalStreamIntel流结束时的最终统计含 DNS 解析、TCP 连接、TLS 握手、发送/接收各阶段的毫秒时间戳未发生阶段为 -1、socket_reused、sent_byte_count、received_byte_count、upstream_protocol等。on_error收到的EnvoyError带有error_code与message错误码枚举ErrorCode包括UndefinedError、StreamReset、ConnectionFailure、BufferLimitExceeded、RequestTimeoutmodule_definition.cc。测试 fetch_test.py 验证了取消流的语义取消后stream_start_ms/stream_end_ms有值而sending_end_ms、response_start_ms、upstream_protocol为 -1。EngineBuilder 高级配置EngineBuilder是引擎配置的入口链式调用、逐项返回自身。pybind11 绑定module_definition.cc暴露了完整的配置面核心项分类如下类别方法说明日志set_log_level(level)、enable_logger(bool)日志级别枚举LogLeveltrace/debug/info/warn/error/critical/off生命周期set_on_engine_running(closure)、set_on_engine_exit(closure)引擎就绪/退出回调网络超时add_connect_timeout_seconds(n)、set_stream_idle_timeout_seconds(n)、set_per_try_idle_timeout_seconds(n)连接超时、流空闲超时、单次尝试空闲超时DNSadd_dns_refresh_seconds(n)、add_dns_failure_refresh_seconds(base, max)、add_dns_query_timeout_seconds(n)、add_dns_min_refresh_seconds(n)、enable_dns_cache(on, save_interval_seconds1)DNS 刷新/失败回退/查询超时/最小刷新间隔/持久缓存协议enable_http3(bool)、add_quic_hint(host, port)、add_quic_canonical_suffix(suffix)、add_h2_connection_keepalive_idle_interval_milliseconds(ms)、add_h2_connection_keepalive_timeout_seconds(n)HTTP/3 开关、QUIC 预连接提示、H2 keepalive连接池add_max_connections_per_host(n)单主机最大连接数传输优化enable_gzip_decompression(bool)、enable_brotli_decompression(bool)、enable_interface_binding(bool)、enable_socket_tagging(bool)、enable_worker_thread(bool)解压、网卡绑定、socket tag、worker 线程安全enforce_trust_chain_verification(bool)、enable_platform_certificates_validation(bool)、set_upstream_tls_sni(sni)信任链校验、平台证书校验、上游 SNI标识set_app_version(v)、set_app_id(id)、set_device_os(os)、set_node_id(id)应用/设备/节点标识用于统计上报运行时add_runtime_guard(guard, value)、enable_stats_collection(bool)、set_network_thread_priority(n)运行时开关、统计采集、线程优先级收尾build()构建并启动引擎GIL 释放集成测试中常见的组合是开启 worker 线程、统计与 socket tagging并通过set_on_engine_runningthreading.Event等待引擎就绪见 httpx_transport_fetch_test.pyengine_running threading.Event() engine ( EngineBuilder() .set_log_level(LogLevel.info) .enable_stats_collection(True) .enable_socket_tagging(True) .set_on_engine_running(lambda: engine_running.set()) .enable_worker_thread(True) .build() ) engine_running.wait(timeout30)与 httpx 生态集成Transport 与共享引擎除自带的AsyncClient外该库还提供一套完整的 httpx transport 适配让httpx.Client/httpx.AsyncClient直接跑在 Envoy 引擎之上EnvoyClientTransport同步 transport用threading.Event和queue.Queue桥接 Envoy 回调与 httpx 同步迭代AsyncEnvoyClientTransport异步 transport用asyncio.Future、asyncio.Queue桥接回调EnvoyTransportFactory单例工厂保证进程内只创建一个 Envoy 引擎引擎初始化代价高应存活于整个进程生命周期通过get_shared_engine()、get_async_transport()、get_sync_transport()对外提供。from envoy_mobile import EnvoyTransportFactory import httpx transport EnvoyTransportFactory.get_async_transport() async with httpx.AsyncClient(transporttransport) as client: response await client.get(https://example.com)请求头映射与限制transport 层通过 get_envoy_headers 将httpx.Request映射为 Envoy 伪头格式并有几点工程化处理跳过连接相关头connection、keep-alive、proxy-connection、transfer-encoding、upgrade这些 HTTP/1.1 专有头在 HTTP/2 中是被禁止的直接丢弃交由 Envoy 自己设置正确的上游连接头避免覆盖伪头与 host:开头的头和host头会被跳过由 Envoy Mobile 自动设置权威主机socket tag每个 transport 实例通过itertools.count生成唯一的 32 位 tag并写入x-envoy-mobile-socket-tag头用于连接池隔离transport 关闭时调用engine.drain_connections_by_socket_tag(tag)精准排空该 transport 的连接而不影响其他 transport测试见 httpx_transport_fetch_test.py。错误映射Envoy 错误码会被映射为 httpx 异常httpx_utils.pyConnectionFailure(2)→httpx.ConnectErrorRequestTimeout(4)→httpx.ReadTimeoutStreamReset(1)→httpx.RemoteProtocolError其余归入httpx.RequestError。请求体流式发送同步/异步 transport 都采用“先发头end_streamFalse→ 逐块send_data→stream.close(b)标记结束”的三段式发送async_client_transport.py支持生成器作为请求体逐块发送、大文件上传不整体驻留内存响应侧则以 64KB 为单位read_data(65536)拉取数据async_client_transport.py。构建与打包Bazel该库使用 Bazel 构建mobile/library/python/BUILD 给出了完整定义envoy_enginepybind11 扩展pybind_extension入口 module_definition.cc依赖 C 层的//library/cc:engine_builder_lib、//library/cc:envoy_engine_cc_lib_no_stamp等产物为envoy_engine.soWindows 为 .pyd并通过 genrule 拷贝到envoy_mobile/包目录内以便随 wheel 分发envoy_mobile_lib纯 Python 包glob全部envoy_mobile/**/*.py依赖httpx来自mobile_pip3的 pip 依赖并打包py.typed标记提供类型提示envoy_mobile_wheelpy_wheel目标发行名为envoy-mobile-client版本 0.5.0可通过--//library/python:python_platform与--//library/python:python_version两个 flag 选择平台manylinux2014_x86_64、macosx_10_15_x86_64、macosx_11_0_arm64与 Python 版本3.12/3.13/3.14。README 注释中的示例构建命令为bazel build -c opt --stripalways \ //library/python:envoy_mobile_wheel \ --//library/python:python_platformmanylinux2014_x86_64小结Envoy Mobile Python 库把 Envoy 的进程内网络引擎能力带给了 Python日常开发优先使用基于 asyncio 的AsyncClient及其便捷的Response读取接口需要精细化控制时直接操作Engine/Stream底层 API借助StreamIntel/FinalStreamIntel获取完整的连接与传输时序若已在 httpx 生态中可选用同步/异步 transport 或共享引擎工厂获得请求体流式发送、连接池隔离与精准排空等高级能力。实践时请注意以本仓库 module_definition.cc 中 pybind11 暴露的实际签名为准并参考 mobile/test/python 下的集成测试验证各 API 的真实行为。【免费下载链接】envoyCloud-native high-performance edge/middle/service proxy项目地址: https://gitcode.com/GitHub_Trending/en/envoy创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考