web3j 事件订阅实战指南：EventFlowable、EthFilter 过滤器与 WebSocket 推送订阅

• 区块链
• Web3
• 后端

导读

本文围绕 web3j 的事件监听体系展开，覆盖四条主流路径：基于生成 wrapper 的...EventFlowable(...)合约事件订阅、无 wrapper 时用EthFilter+ethLogFlowable的通用日志过滤、基于WebSocketService的节点级推送订阅（如新区块头 newHeads）、以及历史区块/交易回放（replay）。读完本文，你将能根据实际场景正确选择事件监听路径，写出可运行的事件监听器，并理解其底层 Flowable 实现与资源释放（dispose）规范。

一、先决策：四条事件监听路径如何选

web3j 的事件监听没有"万能写法"，正确姿势是先根据使用场景做决策。官方决策规则将需求划分为四类：

如果用户需要可立即运行的起始工程，优先从仓库中的.agents/skills/web3j/assets/event-subscriptions/模板就近复制改造；需要把事件监听嵌入完整合约工作流（部署→调用→监听）时，则参考.agents/skills/web3j/assets/erc20-example/basic-erc20-app/。

通用铁律：无论走哪条路径，订阅不再需要时必须dispose()释放订阅，并调用web3j.shutdown()关闭底层服务。

二、路径一：wrapper 生成的事件 Flowable

这是最推荐的合约事件订阅方式，前提是合约已有生成的 wrapper。

2.1 标准四步流程

1. 加载合约 wrapper：ERC20.load(contractAddress, web3j, credentials, gasProvider)；
2. 调用事件辅助方法：transferEventFlowable(DefaultBlockParameter.EARLIEST, DefaultBlockParameter.LATEST)或transferEventFlowable(EthFilter filter)；
3. 订阅返回的 RxJavaFlowable：.subscribe(onNext, onError)；
4. 关闭时 dispose：subscription.dispose()。

2.2 wrapper 到底生成了什么

以仓库中真实生成的 ERC20 wrapper contracts/src/main/java/org/web3j/contracts/eip20/generated/ERC20.java 为例，每个事件会生成四类成员（ERC721 的Transfer/Approval/ApprovalForAll同理，见 ERC721.java）：

• 类型化响应类TransferEventResponse：含_from、_to、_value及log字段；
• getTransferEvents(TransactionReceipt)：从交易回执中静态解析事件（同步）；
• transferEventFlowable(EthFilter filter)：接受自定义过滤器的实时流；
• transferEventFlowable(startBlock, endBlock)：便捷重载——内部自动构造EthFilter并调用filter.addSingleTopic(EventEncoder.encode(TRANSFER_EVENT))。

看其实现，transferEventFlowable(filter)的核心只有一句话（第 170-183 行）：

return web3j.ethLogFlowable(filter).map(log -> { Contract.EventValuesWithLog eventValues = extractEventParametersWithLog(TRANSFER_EVENT, log); TransferEventResponse typedResponse = new TransferEventResponse(); typedResponse.log = log; typedResponse._from = (String) eventValues.getIndexedValues().get(0).getValue(); typedResponse._to = (String) eventValues.getIndexedValues().get(1).getValue(); typedResponse._value = (BigInteger) eventValues.getNonIndexedValues().get(0).getValue(); return typedResponse; });

即 wrapper 只是把"原始Log→ 解码 → 类型化响应对象"这一过程封装成Flowable的map变换。底层解码逻辑位于 core/src/main/java/org/web3j/tx/Contract.java 的extractEventParametersWithLog：indexedValues对应事件中的 indexed 参数（从 topics 解码），nonIndexedValues对应非 indexed 参数（从 data 解码），最终经FunctionReturnDecoder完成 ABI 解码。

2.3 源码佐证：codegen 如何生成这些方法

wrapper 代码不是手写的，而是由 codegen/src/main/java/org/web3j/codegen/SolidityFunctionWrapper.java 的buildEventFlowableFunction/buildDefaultEventFlowableFunction生成：

• 方法名 = 事件名首字母小写 +EventFlowable（如Transfer→transferEventFlowable）；
• 带 filter 版本直接return web3j.ethLogFlowable(filter).map(log -> ...getXxxFromLog(log))；
• 带区块范围版本则构造new EthFilter(startBlock, endBlock, getContractAddress())并addSingleTopic(EventEncoder.encode(事件定义))；
• 回执解析方法名 =get+ 事件名首字母大写 +Events（如getTransferEvents）。

Topic 的编码由 abi/src/main/java/org/web3j/abi/EventEncoder.java 完成：拼接EventName(type1,type2,...)签名 →Hash.sha3→ 十六进制，即以太坊标准的事件签名 topic0 计算方式。

2.4 端到端实战：basic-erc20-app

仓库提供了一套完整的可运行示例 .agents/skills/web3j/assets/erc20-example/basic-erc20-app/，流程为：

1. Solidity 源码 ExampleToken.sol（含Transfer/Approval事件）；
2. Gradle 插件生成 wrapper 到example.contracts包；
3. 主程序 Erc20ExampleApp.java 完成部署、读取状态、转账、监听Transfer事件、从回执解析事件五件事；
4. Erc20EvmTest.java 在@EVMTest(type = NodeType.BESU)内嵌节点下验证同一套事件订阅逻辑。

核心监听代码（节选）：

CountDownLatch eventSeen = new CountDownLatch(1); transferSubscription = token.transferEventFlowable( DefaultBlockParameterName.EARLIEST, DefaultBlockParameterName.LATEST) .subscribe( event -> { System.out.printf("Transfer event: from=%s to=%s value=%s%n", event._from, event._to, event._value); eventSeen.countDown(); }, error -> { error.printStackTrace(System.err); eventSeen.countDown(); }); TransactionReceipt receipt = token.transfer(recipient, transferAmount).send(); List<ExampleToken.TransferEventResponse> receiptEvents = token.getTransferEvents(receipt); // ... finally 中 transferSubscription.dispose() 且 web3j.shutdown()

运行命令：

WEB3J_PRIVATE_KEY=0xyourprivatekey ./gradlew run # 仅跑测试路径（不需要私钥） ./gradlew test

环境变量一览：

测试中验证了getTransferEvents(receipt)与transferEventFlowable两条路径得到的_to/_value完全一致，并用CountDownLatch断言流式事件确实被观察到——这是集成测试里最常用的验证模式。

三、路径二：无 wrapper 的通用日志监听（EthFilter + ethLogFlowable）

手头没有 wrapper、但有合约地址和事件 topic 时，走这条路径。官方推荐顺序：

1. 构造EthFilter，指定区块范围与合约地址；
2. 需要时用addSingleTopic(...)收紧到具体事件；
3. web3j.ethLogFlowable(filter)获得日志流；
4. 直接打印原始Log或手工解码。

3.1 EthFilter 的结构

core/src/main/java/org/web3j/protocol/core/methods/request/EthFilter.java 提供以下构造方式：

• new EthFilter(fromBlock, toBlock, address)——地址可为单个String或List<String>，对应 JSON-RPCeth_newFilter的fromBlock/toBlock/address参数；
• new EthFilter()——全空过滤器（接收所有日志）；
• new EthFilter(blockHash)——按区块哈希过滤，不能与 fromBlock/toBlock 混用（源码注释明确：blockHashoptional, cannot be used together with fromBlock/toBlock）；
• new EthFilter(blockHash, address)。

注意三个过滤维度：区块范围、地址、topics。文档特别强调：过滤器作用于indexed topics；省略 topics 会扩大订阅范围（收到该合约的所有日志，需自行按 event signature 过滤）；此外文档明确提醒Infura 等部分托管节点不支持过滤器，使用前需确认节点能力。

3.2 可运行模板：contract-logs

仓库提供了最小 Gradle 模板 ContractLogSubscriber.java，核心逻辑：

Web3j web3j = Web3j.build(new HttpService(httpUrl)); // 默认 http://localhost:8545 EthFilter filter = new EthFilter( DefaultBlockParameterName.EARLIEST, DefaultBlockParameterName.LATEST, contractAddress); if (eventTopic0 != null && !eventTopic0.isBlank()) { filter.addSingleTopic(eventTopic0); // 可选：精确到某一事件 } Disposable subscription = web3j.ethLogFlowable(filter) .subscribe( log -> { List<String> topics = log.getTopics(); System.out.printf("log block=%s tx=%s address=%s topics=%s data=%s%n", log.getBlockNumber(), log.getTransactionHash(), log.getAddress(), topics, log.getData()); }, error -> { error.printStackTrace(); keepAlive.countDown(); });

环境变量：

3.3 手工解码与 EventFilterIT 佐证

原始Log对象（Log.java）的关键字段：getAddress()、getTopics()（List<String>）、getData()、getBlockNumber()、getTransactionHash()。

手工解码的标准模式可参考集成测试 integration-tests/src/test/java/org/web3j/protocol/scenarios/EventFilterIT.java：

Event event = new Event("Notify", Arrays.asList(new TypeReference<Uint256>() {}, new TypeReference<Uint256>() {})); String encodedEventSignature = EventEncoder.encode(event); // 1) 校验：topics[0] 必须等于事件签名 assertEquals(topics.get(0), encodedEventSignature); // 2) 解码 data 中的非 indexed 参数 List<Type> results = FunctionReturnDecoder.decode(log.getData(), event.getNonIndexedParameters()); // 3) 用 ethGetLogs 验证过滤器能捞到该事件 EthFilter ethFilter = new EthFilter(EARLIEST, LATEST, fib.getContractAddress()); ethFilter.addSingleTopic(encodedEventSignature); List<EthLog.LogResult<?>> filterLogs = web3j.ethGetLogs(ethFilter).send().getLogs();

该测试验证了三件事：无 indexed 参数时topics只有一个元素（事件签名）；EventEncoder.encode的结果与链上topics[0]完全一致；构造的过滤器能查回同一事件。这套"编码 → 比对 → 解码 → 过滤回查"的流程就是手工日志解析的完整闭环。

四、路径三：WebSocket 推送订阅（newHeads 等）

需要推送式订阅（而非轮询）时使用 WebSocket。官方流程：

1. new WebSocketService(wsUrl, includeRawResponses)；
2. webSocketService.connect()；
3. Web3j.build(webSocketService)基于该服务构建客户端；
4. 订阅newHeadsNotifications()或其他推送 Flowable。

4.1 底层原理：从 eth_subscribe 到 Flowable

看 core/src/main/java/org/web3j/protocol/core/JsonRpc2_0Web3j.java 的实现，newHeadsNotifications()本质是对 JSON-RPC 订阅的封装：

public Flowable<NewHeadsNotification> newHeadsNotifications() { return web3jService.subscribe( new Request<>("eth_subscribe", Collections.singletonList("newHeads"), web3jService, EthSubscribe.class), "eth_unsubscribe", NewHeadsNotification.class); }

即发起eth_subscribe订阅"newHeads"主题，取消时发eth_unsubscribe。同类推送 API 还有logsNotifications(addresses, topics)（对应eth_subscribe的"logs"主题，参数为{address, topics}，见 JsonRpc2_0Web3j.java）。WebSocketService的构造签名是WebSocketService(String serverUrl, boolean includeRawResponses)，第二个参数控制是否保留原始响应（WebSocketService.java）。

4.2 可运行模板：websocket-newheads

仓库提供最小 Gradle 模板 NewHeadsSubscriber.java，是"监听新区块头"的首选可运行示例：

String websocketUrl = envOrDefault("WEB3J_WS_URL", "ws://localhost:8546"); WebSocketService webSocketService = new WebSocketService(websocketUrl, false); webSocketService.connect(); Web3j web3j = Web3j.build(webSocketService); CountDownLatch keepAlive = new CountDownLatch(1); Disposable subscription = web3j.newHeadsNotifications() .subscribe( notification -> printHead(notification.getParams().getResult()), error -> { error.printStackTrace(); keepAlive.countDown(); }); Runtime.getRuntime().addShutdownHook(new Thread(() -> { subscription.dispose(); web3j.shutdown(); webSocketService.close(); })); keepAlive.await();

要点：

• 环境变量：仅WEB3J_WS_URL，默认ws://localhost:8546（HTTP 的 8545 对应 WebSocket 通常为 8546）；
• 通知对象：NewHeadsNotification extends Notification<NewHead>，通过notification.getParams().getResult()拿到NewHead（见 Notification.java 与 NewHeadsNotification.java）；
• NewHead 字段（NewHead.java）：number、hash、parentHash、timestamp、miner、gasLimit、gasUsed、stateRoot、transactionRoot等，模板打印number/hash/parentHash；
• 资源释放三件套：subscription.dispose()→web3j.shutdown()→webSocketService.close()，模板用 shutdown hook 保证 Ctrl+C 时执行。

4.3 运行方式

两类模板均为最小 Gradle Java 工程（settings.gradle分别声明web3j-websocket-newheads与web3j-contract-log-subscriber），运行命令：

gradle run # 或项目带 wrapper 时 ./gradlew run

五、路径四：历史区块与交易回放（Replay Flowable）

回放类 Flowable 用于冷启动追块、补历史数据或"过去+未来"连续监听，实现在 core/src/main/java/org/web3j/protocol/rx/JsonRpc2_0Rx.java：

实现要点（JsonRpc2_0Rx.java）：

• 用getBlockNumber(startBlock)与getLatestBlockNumber()解析起止块号；
• 若 startBlock 已是最新，则返回onCompleteFlowable（即后续实时流）；否则Flowable.concat(历史段, Flowable.defer(递归续追))实现"追平后无缝切换实时"；
• 通过Flowables.range逐块ethGetBlockByNumber拉取，且整个链subscribeOn(scheduler)异步执行。

这些回放 API 连同轮询式事件流，都在集成测试 integration-tests/src/test/java/org/web3j/protocol/core/FlowableIT.java（@EVMTest(type = NodeType.BESU)）中有直接覆盖：testBlockFlowable、testPendingTransactionFlowable、testTransactionFlowable、testLogFlowable、testReplayFlowable、testReplayPastAndFutureBlocksFlowable六个用例，统一用CountDownLatch计数 + 超时断言 +subscription.dispose()校验。

六、关键概念速查

七、仓库内进一步研读指南

• 可运行模板总览：.agents/skills/web3j/assets/event-subscriptions/README.md
• ERC20 端到端示例说明：.agents/skills/web3j/assets/erc20-example/basic-erc20-app/README.md
• 真实生成 wrapper：contracts/src/main/java/org/web3j/contracts/eip20/generated/ERC20.java、contracts/src/main/java/org/web3j/contracts/eip721/generated/ERC721.java
• wrapper 生成逻辑：codegen/src/main/java/org/web3j/codegen/SolidityFunctionWrapper.java
• 轮询/回放 Flowable 实现：core/src/main/java/org/web3j/protocol/rx/JsonRpc2_0Rx.java
• 事件过滤集成测试：integration-tests/src/test/java/org/web3j/protocol/scenarios/EventFilterIT.java
• 各类 Flowable 集成测试：integration-tests/src/test/java/org/web3j/protocol/core/FlowableIT.java

关于 wrapper 加载/部署、以及事务与过滤器更底层的细节，可继续查阅仓库中对应的 smart-contracts 与 transactions 参考资料。最后再次强调：无论选择哪条路径，订阅结束后请务必 dispose 订阅、关闭 Web3j 实例，避免资源泄漏。

• 区块链
• Web3
• 后端