web3.py 内部机制完全指南：Provider、Middleware、Manager 三层架构与持久连接请求处理

• Web3
• 区块链

web3.py 在公开 API（Web3/AsyncWeb3对象）与所连接的节点之间，存在多层抽象：Provider 负责与区块链的底层通信，Middleware 提供请求/响应的拦截钩子，Manager 则承担线程安全与异步原语。本文以官方文档 docs/internals.rst 为核心骨架，结合仓库源码与测试，逐层拆解请求生命周期、请求缓存（Request Caching）、HTTP 异常重试，以及WebSocketProvider/AsyncIPCProvider持久连接下的请求-响应匹配与订阅处理，帮助高级用户理解并定制这些底层 API。

⚠️ 适用提示：本文涉及的是面向高级用户的内层 API。如果你不清楚自己在做什么，建议停留在公开 API（如w3.eth、w3.contract）层面。

请求生命周期：一次 RPC 调用的完整旅程

每一个 web3 RPC 调用都会依次穿过以下三层：

*********** ************ | Request | | Response | *********** ************ | ^ v | +-----------------------------+ | Manager | +-----------------------------+ | ^ v | +-----------------------------+ | Middleware | +-----------------------------+ | ^ v | +-----------------------------+ | Provider | +-----------------------------+

可以用“洋葱”来理解这个关系：Provider 在最中心，请求从最外层的 Manager 发起，逐层穿过每一层 Middleware，最终抵达中心的 Provider；Provider 处理完后，响应再从洋葱中心逐层向外传出，最终由 Manager 返回给调用者。

从源码看，这一流程的实现位于 RequestManager：_make_request()调用provider.request_func()，后者在 BaseProvider.request_func() 中把整个MiddlewareOnion与self.make_request组合（combine_middleware），再逐层包装出最终的请求函数；同步路径经request_blocking()返回，异步路径经coro_request()返回，最终都会调用 formatted_response() 做 JSON-RPC 结果校验与格式化。异步场景下请求与响应分离（send()/recv()），详见下文“持久连接”一节。

利用这套分层可以实现的常见场景包括：

• 把某些 RPC 请求重定向到不同的 Provider，例如所有读操作发给远程节点，所有写操作发给本地自控节点；
• 透明拦截eth_sendTransaction发送的交易，在本地签名后改走eth_sendRawTransaction发送；
• 修改 RPC 响应格式，例如把响应中的整数值统一转换为十六进制；
• 校验 RPC 请求的输入参数。

Providers：直接与区块链交互的层

Provider 负责所有与区块链的直接交互。大多数情况下这意味着通过 HTTP 或 IPC socket 与以太坊节点的 JSON-RPC 服务通信；但 Provider 并不强制要求基于 RPC，例如测试场景可以基于内存 EVM（in-memory EVM）实现一个 Provider 来满足请求，仓库中的 eth-tester 相关实现即属此类。

编写自定义 Provider

自定义 Provider 需要实现两个必需方法，并设置该 Provider 使用的 Middleware：

• BaseProvider.make_request(method, params)：每个 Provider 类必须实现此方法。它应当返回一个 JSON 对象：成功时带'result'键，失败时带'error'键。
  • method：被调用的 JSON-RPC 方法名字符串，如'eth_sendTransaction'；
  • params：该 JSON-RPC 方法的参数列表或其他可迭代对象。
  • 基类中的默认实现直接抛出NotImplementedError（见 web3/providers/base.py）。
• BaseProvider.is_connected(show_traceback=False)：根据 Provider 是否应被视为“已连接”返回True/False。例如 IPC socket 类 Provider，socket 打开返回True，关闭返回False；若传入show_traceback=True，则在不应视为已连接时抛出ProviderConnectionError并给出原因。参考实现JSONBaseProvider.is_connected()（web3/providers/base.py）会发一次web3_clientVersion请求来探测连接。
• BaseProvider.middleware：应为可迭代的 Middleware 集合。可以通过给provider.middleware赋值来设置新的 Middleware 列表，列表开头的第一个 Middleware 最先处理请求。

此外，JSONBaseProvider还提供了encode_rpc_request/decode_rpc_response（见 web3/providers/base.py），负责把请求封装为 JSON-RPC 2.0 格式（自动生成递增的id）以及解析原始字节响应，自定义 HTTP/IPC 类 Provider 可复用这些工具。

Provider 配置：请求缓存（Request Caching）

⚠️ 重要：在启用请求缓存之前，请先熟悉其校验逻辑。该特性为了尽力保证数据有效性，常常需要在底层发起额外的请求，可能为你的场景引入不必要的开销。可以设置request_cache_validation_threshold为None来关闭校验（缓存所有允许缓存的请求），也可以按需调整以匹配你的性能诉求。

请求缓存通过 Provider 实例上的以下配置项启用与调整：

• cache_allowed_requests: bool = False：总开关，默认关闭；
• cacheable_requests: Optional[Set[RPCEndpoint]]：允许被缓存的 RPC 端点集合；
• request_cache_validation_threshold: Optional[Union[RequestCacheValidationThreshold, int]]：缓存校验阈值。

对不依赖区块数据的请求（如eth_chainId），把cache_allowed_requests设为True即可安全地缓存所有响应。但对依赖区块数据的请求（如eth_getBlockByNumber），不能总是缓存——区块数据会变化，例如链重组（reorg）期间或尚未达成最终性（finality）时。request_cache_validation_threshold即为这类请求配置一个安全的缓存阈值：

• 默认情况下，该选项被配置为对你所连接链的 chain id 而言“安全”的内部值：连接以太坊主网时，取finalized（已最终确定）区块号；
• 连接其他链时，取一个从当前时刻算起、对该链最终性机制“安全”的时间间隔（秒）。
• 对不在内部列表中的链（包括所有测试网），默认值为 1 小时。

内部配置的链及其默认阈值如下（源码见 CHAIN_VALIDATION_THRESHOLD_DEFAULTS）：

以以太坊主网为例：请求所依赖的区块号小于等于finalized区块号时，响应会被缓存；超过finalized区块号则不缓存。对其他链，当请求所依赖数据的区块时间戳早于等于该链配置的时间间隔时缓存响应。

可以通过三种方式修改该行为（见 BaseProvider 的构造参数）：

• 设为RequestCacheValidationThreshold.SAFE：以safe区块作为阈值（仅以太坊主网）；
• 设为自定义秒数（任何链，包括以太坊主网）；
• 设为None：关闭校验、缓存所有请求（对非测试网链不推荐）。

RequestCacheValidationThreshold枚举（主网finalized与safe两个值）从web3.utils模块导入，见 web3/utils/caching.py。另外注意cacheable_requests用于限定允许缓存的端点集合，默认是一个内部维护的“安全可缓存”列表，排除了像eth_call这种响应可变、不宜缓存的端点。默认的可缓存请求列表如下（加粗者为受request_cache_validation_threshold校验的请求）：

• eth_chainId
• web3_clientVersion
• net_version
• eth_getBlockByNumber
• eth_getRawTransactionByBlockNumberAndIndex
• eth_getBlockTransactionCountByNumber
• eth_getUncleByBlockNumberAndIndex
• eth_getUncleCountByBlockNumber
• eth_getBlockByHash
• eth_getTransactionByHash
• eth_getTransactionByBlockNumberAndIndex
• eth_getTransactionByBlockHashAndIndex
• eth_getBlockTransactionCountByHash
• eth_getRawTransactionByBlockHashAndIndex
• eth_getUncleByBlockHashAndIndex
• eth_getUncleCountByBlockHash

从实现看，该默认列表由 INTERNAL_VALIDATION_MAP 的键构成（CACHEABLE_REQUESTS），并按端点类型分为三类校验器：参数含区块号（validate_from_block_id_in_params）、结果含区块信息（validate_from_blocknum_in_result）、参数含区块哈希（validate_from_blockhash_in_params），异步场景有对应实现（request_caching_validation.py）。值得注意的是，这些校验器在必要时会额外发起 RPC 请求来获取区块信息（例如验证一笔交易是否已越过安全阈值），高频率请求场景下可能造成明显开销；此外校验期间会临时关闭缓存以防递归（见 is_beyond_validation_threshold 的cache_allowed_requests = False处理）。因此，理解何时开缓存、如何配置校验，是避免不必要开销的关键。

配置示例：

from web3 import Web3, HTTPProvider from web3.utils import RequestCacheValidationThreshold w3 = Web3(HTTPProvider( endpoint_uri="...", # 可选：开启请求缓存，默认 False cache_allowed_requests=True, # 可选：允许缓存的端点集合，默认是内部“安全可缓存”列表（见上） cacheable_requests={"eth_chainId", "eth_getBlockByNumber"}, # 可选：默认根据 chain id 取值（见上） request_cache_validation_threshold=60 * 60, # 1 小时 # request_cache_validation_threshold=RequestCacheValidationThreshold.SAFE, # 仅以太坊主网 ))

Provider 配置：HTTP Provider 请求重试

HTTPProvider与AsyncHTTPProvider默认会对某些请求在异常时进行重试。通过 Provider 实例上的exception_retry_configuration属性配置，其值为 ExceptionRetryConfiguration（一个 pydanticBaseModel）实例。重试机制采用指数退避策略：以backoff_factor决定的初始值为起点，每次重试间隔翻倍，最多尝试retries次。

配置项说明：

• errors：Provider 应重试的异常元组。默认值：HTTPProvider为(ConnectionError, requests.HTTPError, requests.Timeout)；AsyncHTTPProvider为(aiohttp.ClientError, asyncio.TimeoutError)。两种 Provider 的DEFAULT_EXCEPTIONS定义见下。
• retries：重试次数，默认 5。
• backoff_factor：初始延迟倍数，每次重试翻倍，默认 0.125。
• method_allowlist：允许重试的方法列表，默认是内部“安全可重试”方法清单REQUEST_RETRY_ALLOWLIST（见 web3/providers/rpc/utils.py，包含admin、net、txpool、testing、evm前缀方法以及eth_call、eth_sendRawTransaction等读/写方法）。

示例（展示默认选项及如何覆盖）：

from web3 import Web3, HTTPProvider from web3.providers.rpc.utils import ( REQUEST_RETRY_ALLOWLIST, ExceptionRetryConfiguration, ) w3 = Web3(HTTPProvider( endpoint_uri="...", exception_retry_configuration=ExceptionRetryConfiguration( errors=DEFAULT_EXCEPTIONS, # 重试次数 retries=5, # 初始延迟倍数，每次重试翻倍 backoff_factor=0.125, # 内部默认的可重试方法清单 method_allowlist=REQUEST_RETRY_ALLOWLIST, ), ))

不同 HTTP Provider 的DEFAULT_EXCEPTIONS定义为：

• HTTPProvider：(ConnectionError, requests.HTTPError, requests.Timeout)
• AsyncHTTPProvider：(ConnectionError, aiohttp.ClientError, asyncio.TimeoutError)

把retry_configuration（即exception_retry_configuration）设为None将关闭该 Provider 实例的异常重试：

from web3 import Web3, HTTPProvider w3 = Web3(HTTPProvider(endpoint_uri="...", retry_configuration=None)

从实现看，重试逻辑位于 HTTPProvider._make_request() 与AsyncHTTPProvider的对应方法：仅当配置非空且check_if_retry_on_failure()（web3/providers/rpc/utils.py）判定方法在允许列表中时执行重试，每次捕获errors中的异常后按backoff_factor * 2**i计算退避延迟（见 async_rpc.py）。仓库在 tests/core/providers/test_http_request_retry.py 中对重试次数、异常类型与配置等价性均有覆盖。

Managers：请求/响应生命周期的守门人

Manager 是请求/响应生命周期的守门人（gatekeeper）。大部分功能可以在 Middleware 层实现，因此一般无需更换 Manager。RequestManager在初始化时会装配默认 Middleware 洋葱（get_default_middleware：gas_price_strategy、ens_name_to_address、attrdict、validation、gas_estimate），并提供request_blocking/coro_request两个同步/异步请求入口。

持久连接 Provider 的请求处理（RequestProcessor）

RequestProcessor 负责为PersistentConnectionProvider存储并同步异步请求与其响应。WebSocketProvider 与AsyncIPCProvider是两个持久连接 Provider。由于要通过同一个 socket 发送请求并接收对应响应，PersistentConnectionProvider必须把请求的id与 socket 返回的响应的id匹配起来。任何不按 JSON-RPC 2.0 规范携带 id 的 Provider 都无法与PersistentConnectionProvider一起工作。

从实现看，RequestProcessor内部维护了三个独立数据结构（request_processor.py）：_request_information_cache（请求信息缓存，容量默认 500）、_request_response_cache（一对一响应缓存，容量 500）、_subscription_response_queue（一对多订阅队列，FIFO，容量默认 500，类型为TaskReliantQueue）。

监听响应

PersistentConnectionProvider的实现类在 socket 连接建立时会启动一个消息监听后台任务（_message_listener_task，见 persistent.py 的connect()）。该任务负责监听 socket 连接上到来的所有消息，并存入 Provider 实例内部的RequestProcessor。RequestProcessor根据消息是否带 JSON-RPCid值，把消息存到不同的缓存：一对一缓存，或一对多（订阅）队列。

一对一请求

一对一请求指期望只收到一个响应的请求，例如通过eth模块 API 获取最新区块号：

>>> async def ws_one_to_one_example(): ... async with AsyncWeb3(WebSocketProvider(f"ws://127.0.0.1:8546")) as w3: ... # 发起请求并在同一行内收到单个响应 ... latest_block_num = await w3.eth.block_number >>> asyncio.run(ws_one_to_one_example())

持久 socket 连接下，我们只能调用send()，再通过其他方式（一般是recv()或遍历 socket 消息）异步接收响应。正因收发是异步的，为了把响应匹配回原始请求，必须把请求信息存到某处（即与收到的响应id匹配的那个请求）。存储的请求信息随后用于处理响应：把它送入 web3.py 库内部的响应格式化器（response formatters）与中间件处理管线。

存储请求信息的载体是RequestProcessor内部的RequestInformation缓存类，其保存的字段如下（源码见 RequestInformation）：

• method：方法名，如"eth_subscribe"；
• params：发起调用时的参数，如("newPendingTransactions", True)；
• response_formatters：用于处理响应的格式化器；
• middleware_response_processors：发起请求时实例上存在的、处理响应的中间件处理器，按顺序追加，响应到来时依序经过这些逻辑；
• subscription_id：若请求是eth_subscribe，收到订阅调用的响应（即订阅id）时不是把这条信息从缓存弹出，而是把订阅 id 一并保存到请求信息中，以便正确后续处理所有携带该订阅id的订阅消息；一对一请求场景该值恒为None。

一对一响应（响应对象中含 JSON-RPCid）存放在内部SimpleCache实例中，与任何一对多响应隔离。PersistentConnectionProvider在内部查找响应时，期望监听任务把响应存入该缓存；由于缓存键的生成使用了请求id，它会查找与响应id匹配请求id的缓存键：找到则处理并返回给用户；找不到则操作超时并抛出TimeExhausted异常。该超时可通过实例化PersistentConnectionProvider时的response_timeout关键字参数配置（对应源码中的request_timeout，默认 30 秒，见 persistent.py）。相关缓存键生成逻辑见 generate_cache_key。

一对多请求（订阅）

一对多请求指一次初始请求后期望收到多条响应的请求，目前唯一的例子是eth_subscribe。初始的eth_subscribe请求本身只期待一个响应——订阅id值，但成功后还期待收到很多eth_subscription消息。因此原始请求被视为一对一请求，以便在同一行把订阅id返回给用户；该调用后续产生的多条响应可以按以下几种方式之一处理。

方式一：订阅管理器 API（推荐）

订阅管理器 API 是AsyncWeb3类上的公开 API，当连接的是PersistentConnectionProvider实例时可用，允许用户订阅一个订阅并异步处理其多条响应。只要每次订阅调用都传入 handler，subscription_manager实例就负责处理 socket 连接上到来的多条响应；用户用完订阅后，也可以用它退订。

>>> async def new_heads_handler( ... handler_context: NewHeadsSubscriptionContext, ... ) -> None: ... result = handler_context.result ... print(f"New block header: {result}\n") ... if result["number"] > 1234567: ... await handler_context.subscription.unsubscribe() >>> async def ws_subscription_example(): ... async with AsyncWeb3(WebSocketProvider(f"ws://127.0.0.1:8546")) as w3: ... # 订阅新区块头并拿到 subscription_id。 ... # 这是一次一对一调用，附带大量响应的触发器 ... subscription_id = await w3.eth.subscribe("newHeads", handler=new_heads_handler) ... ... # 用订阅管理器异步处理订阅消息。 ... # 会一直运行到订阅管理器中不再有订阅为止； ... # 若把 `run_forever` 置为 `True` 则无限运行。 ... await w3.subscription_manager.handle_subscriptions(run_forever=False) >>> asyncio.run(ws_subscription_example())

订阅管理器还可以一次性订阅多个订阅。EthSubscription系列类可从web3.utils.subscriptions导入，提供了友好的订阅管理 API。由于每个连接与 Provider 实例都有各自的监听任务与订阅管理器实例，你可以一次订阅多个订阅，并通过 handler 处理多条响应。handler 中包含：

• async_w3：发起该订阅的AsyncWeb3实例；
• subscription：handler 所依附的订阅实例；
• result：socket 连接上到来的、属于该订阅的响应。

订阅还接受handler_context参数，可在订阅时向 handler 传递额外信息，例如传入一个事件对象，用于在日志事件到来时解析它。

>>> from web3 import ( ... AsyncWeb3, ... WebSocketProvider, ... AsyncIPCProvider, ... ) >>> from web3.utils.subscriptions import ( ... EthSubscription, ... NewHeadsSubscription, ... NewHeadsSubscriptionContext, ... PendingTxSubscription, ... PendingTxSubscriptionContext, ... LogsSubscription, ... LogsSubscriptionContext, ... ) >>> async def new_heads_handler( ... handler_context: NewHeadsSubscriptionContext, ... ) -> None: ... header = handler_context.result ... print(f"New block header: {header}\n") ... if header["number"] > 1234567: ... await handler_context.subscription.unsubscribe() >>> async def pending_txs_handler( ... handler_context: PendingTxSubscriptionContext, ... ) -> None: ... ... >>> async def log_handler( ... handler_context: LogsSubscriptionContext, ... ) -> None: ... log_receipt = handler_context.result ... # 订阅日志时我们把事件传入了 handler_context，现在可直接取用 ... event_data = handler_context.transfer_event.process_log(log_receipt) ... print(f"Log event data: {event_data}\n") >>> async def sub_manager(): ... local_w3 = await AsyncWeb3(AsyncIPCProvider(LOCAL_IPC)) ... ... # 通过订阅管理器 + handler 一次订阅多个订阅 ... weth_contract = local_w3.eth.contract( ... address=local_w3.to_checksum_address("0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"), ... abi=WETH_ABI, ... ) ... transfer_event = weth_contract.events.Transfer() ... await local_w3.subscription_manager.subscribe( ... [ ... NewHeadsSubscription(label="new-heads-mainnet", handler=new_heads_handler), ... PendingTxSubscription( ... label="pending-tx-mainnet", # 可选 label ... full_transactions=True, ... handler=pending_tx_handler, ... ), ... LogsSubscription( ... label="WETH transfers", # 可选 label ... address=weth_contract.address, ... topics=[transfer_event.topic], ... handler=log_handler, ... # 可选的 handler_context 参数，帮助解析响应 ... handler_context={"transfer_event": transfer_event}, ... ), ... ] ... ) ... ... public_w3 = await AsyncWeb3(WebSocketProvider(PUBLIC_PROVIDER_WS)) ... # 通过 eth_subscribe 订阅，带 handler 与可选 label ... await public_w3.eth.subscribe("public_newHeads", handler=pending_tx_handler, label="new-heads-public-ws") >>> # 处理所有订阅，直到两个订阅管理器中都不再有订阅。 ... # 若任一管理器的 `run_forever` 为 `True`，则无限运行。 >>> await asyncio.gather( ... public_w3.subscription_manager.handle_subscriptions(), ... local_w3.subscription_manager.handle_subscriptions(), ... ) ... ... # 关闭连接 ... await local_w3.provider.disconnect() ... await public_w3.provider.disconnect() >>> asyncio.run(sub_manager())

说明：订阅系列类（如 NewHeadsSubscription、PendingTxSubscription、LogsSubscription、SyncingSubscription）的构造参数在源码中均有类型注解与默认值，例如PendingTxSubscription(full_transactions=False)、LogsSubscription(address=None, topics=None)，且各订阅都支持可选的label、handler_context与parallelize参数；w3.eth.subscribe()会按订阅类型自动创建对应的类型化订阅（见 async_eth.py 的 subscribe() 与 _create_type_aware_subscription）。

方式二：process_subscriptions()异步迭代器

PersistentConnection类（web3/providers/persistent/persistent_connection.py）是操作活动持久 socket 连接的公开 API，其process_subscriptions()方法按异步迭代器模式接收eth_subscription响应。你可以用它在消息到来时监听并处理原始消息。

>>> async def ws_subscription_example(): ... async with AsyncWeb3(WebSocketProvider(f"ws://127.0.0.1:8546")) as w3: ... # 订阅新区块头并拿到 subscription_id。 ... # 这是一次一对一调用，附带大量响应的触发器 ... subscription_id = await w3.eth.subscribe("newHeads") ... ... # 利用 `w3.socket`（`PersistentConnection` 公开 API）的 ... # `process_subscriptions()` 方法监听 socket 上的大量响应 ... async for response in w3.socket.process_subscriptions(): ... # 这里只接收一对多响应，避免把一对一请求的响应 ... # 意外地带进这个代码块 ... ... print(f"{response}\n") ... ... if some_condition: ... # 退订新区块头，这是另一次一对一请求 ... is_unsubscribed = await w3.eth.unsubscribe(subscription_id) ... if is_unsubscribed: ... break >>> asyncio.run(ws_subscription_example())

一对多响应（响应对象中不含 JSON-RPCid）存放在内部asyncio.Queue实例中，与一对一响应隔离。PersistentConnectionProvider在内部查找一对多响应时，期望监听任务把这些消息存入该队列。由于消息顺序很重要，该队列是 FIFO 队列；PersistentConnection的process_subscriptions()方法按 FIFO 从队列弹出消息，并暴露为异步迭代器模式。从源码看，RequestProcessor.cache_raw_response()（request_processor.py）会把带 handler 的订阅消息放入_handler_subscription_queue，无 handler 的放入_subscription_response_queue；_persistent_message_stream()（manager.py）则从队列弹出并经_process_response()格式化后 yield。

如果 socket 的消息流没有被其他任务中断，队列一般会与 socket 上到来的消息保持同步：监听者把消息放入队列，process_subscriptions()从队列弹出该消息并把循环控制权交回监听者，如此往复，直到 socket 连接关闭或用户退订。如果消息流稍有滞后，或 Provider 订阅了订阅却未消费消息，内部队列可能填满至最大容量，随后触发等待中的asyncio.Event（_listen_event），直到 Provider 重新开始消费队列消息。因此，一旦发起订阅，就应尽快通过process_subscriptions()开始消费队列消息。

小结

web3.py 的三层架构把“通信”（Provider）、“拦截/改造”（Middleware）与“调度/同步”（Manager）解耦：请求自 Manager 发起，逐层穿过 Middleware 洋葱到达 Provider，响应再原路返回。对高级使用者而言，最有价值的可定制点集中在 Provider 层——通过make_request/is_connected编写自定义 Provider、通过cache_allowed_requests/cacheable_requests/request_cache_validation_threshold精细控制请求缓存、通过ExceptionRetryConfiguration调整 HTTP 重试策略；而在持久连接（WebSocket/IPC）场景下，RequestProcessor以请求/响应id匹配为核心，配合RequestInformation缓存、FIFO 订阅队列与订阅管理器 API，实现了透明的一对一与一对多（订阅）消息处理。深入理解这些内部机制，能帮助你在不触碰公开 API 的前提下，安全地为特定场景定制 web3.py 的行为。

• Web3
• 区块链