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

区块链Web3后端【免费下载链接】web3jLightweight Java and Android library for integration with Ethereum clients项目地址https://gitcode.com/gh_mirrors/we/web3j点击查看免费下载导读本文围绕 web3j 的事件监听体系展开覆盖四条主流路径基于生成 wrapper 的...EventFlowable(...)合约事件订阅、无 wrapper 时用EthFilterethLogFlowable的通用日志过滤、基于WebSocketService的节点级推送订阅如新区块头 newHeads、以及历史区块/交易回放replay。读完本文你将能根据实际场景正确选择事件监听路径写出可运行的事件监听器并理解其底层 Flowable 实现与资源释放dispose规范。一、先决策四条事件监听路径如何选web3j 的事件监听没有万能写法正确姿势是先根据使用场景做决策。官方决策规则将需求划分为四类场景推荐路径适用前提1. 已部署合约 已生成 wrapper调用 wrapper 生成的...EventFlowable(startBlock, endBlock)或...EventFlowable(filter)已通过 codegen 生成合约 Java 类2. 通用合约日志、无 wrapperEthFilterweb3j.ethLogFlowable(filter)手头有合约地址与事件签名/Topic3. 节点级事件新区块头、pending 交易WebSocketServicenewHeadsNotifications()等推送 Flowable节点开放 WebSocket 端口4. 回放历史区块/交易replayPastBlocksFlowable、replayPastAndFutureBlocksFlowable等需要补全历史数据或冷启动追块如果用户需要可立即运行的起始工程优先从仓库中的.agents/skills/web3j/assets/event-subscriptions/模板就近复制改造需要把事件监听嵌入完整合约工作流部署→调用→监听时则参考.agents/skills/web3j/assets/erc20-example/basic-erc20-app/。通用铁律无论走哪条路径订阅不再需要时必须dispose()释放订阅并调用web3j.shutdown()关闭底层服务。二、路径一wrapper 生成的事件 Flowable这是最推荐的合约事件订阅方式前提是合约已有生成的 wrapper。2.1 标准四步流程加载合约 wrapperERC20.load(contractAddress, web3j, credentials, gasProvider)调用事件辅助方法transferEventFlowable(DefaultBlockParameter.EARLIEST, DefaultBlockParameter.LATEST)或transferEventFlowable(EthFilter filter)订阅返回的 RxJavaFlowable.subscribe(onNext, onError)关闭时 disposesubscription.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 的extractEventParametersWithLogindexedValues对应事件中的 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/流程为Solidity 源码 ExampleToken.sol含Transfer/Approval事件Gradle 插件生成 wrapper 到example.contracts包主程序 Erc20ExampleApp.java 完成部署、读取状态、转账、监听Transfer事件、从回执解析事件五件事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(); ListExampleToken.TransferEventResponse receiptEvents token.getTransferEvents(receipt); // ... finally 中 transferSubscription.dispose() 且 web3j.shutdown()运行命令WEB3J_PRIVATE_KEY0xyourprivatekey ./gradlew run # 仅跑测试路径不需要私钥 ./gradlew test环境变量一览变量默认值说明WEB3J_HTTP_URLhttp://localhost:8545JSON-RPC 端点WEB3J_PRIVATE_KEY无必填run路径要求WEB3J_TOKEN_NAMESkill Token代币名WEB3J_TOKEN_SYMBOLSKILL代币符号WEB3J_INITIAL_SUPPLY1000000初始供应量WEB3J_TRANSFER_TO部署者地址无第二账户也能跑通转账WEB3J_TRANSFER_AMOUNT1转账数量测试中验证了getTransferEvents(receipt)与transferEventFlowable两条路径得到的_to/_value完全一致并用CountDownLatch断言流式事件确实被观察到——这是集成测试里最常用的验证模式。三、路径二无 wrapper 的通用日志监听EthFilter ethLogFlowable手头没有 wrapper、但有合约地址和事件 topic 时走这条路径。官方推荐顺序构造EthFilter指定区块范围与合约地址需要时用addSingleTopic(...)收紧到具体事件web3j.ethLogFlowable(filter)获得日志流直接打印原始Log或手工解码。3.1 EthFilter 的结构core/src/main/java/org/web3j/protocol/core/methods/request/EthFilter.java 提供以下构造方式new EthFilter(fromBlock, toBlock, address)——地址可为单个String或ListString对应 JSON-RPCeth_newFilter的fromBlock/toBlock/address参数new EthFilter()——全空过滤器接收所有日志new EthFilter(blockHash)——按区块哈希过滤不能与 fromBlock/toBlock 混用源码注释明确blockHashoptional, cannot be used together with fromBlock/toBlocknew 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 - { ListString 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(); });环境变量变量必填说明WEB3J_HTTP_URL否默认http://localhost:8545WEB3J_CONTRACT_ADDRESS是监听目标合约地址WEB3J_EVENT_TOPIC0否事件签名 topic0缺省则接收该合约全部日志3.3 手工解码与 EventFilterIT 佐证原始Log对象Log.java的关键字段getAddress()、getTopics()ListString、getData()、getBlockNumber()、getTransactionHash()。手工解码的标准模式可参考集成测试 integration-tests/src/test/java/org/web3j/protocol/scenarios/EventFilterIT.javaEvent event new Event(Notify, Arrays.asList(new TypeReferenceUint256() {}, new TypeReferenceUint256() {})); String encodedEventSignature EventEncoder.encode(event); // 1) 校验topics[0] 必须等于事件签名 assertEquals(topics.get(0), encodedEventSignature); // 2) 解码 data 中的非 indexed 参数 ListType results FunctionReturnDecoder.decode(log.getData(), event.getNonIndexedParameters()); // 3) 用 ethGetLogs 验证过滤器能捞到该事件 EthFilter ethFilter new EthFilter(EARLIEST, LATEST, fib.getContractAddress()); ethFilter.addSingleTopic(encodedEventSignature); ListEthLog.LogResult? filterLogs web3j.ethGetLogs(ethFilter).send().getLogs();该测试验证了三件事无 indexed 参数时topics只有一个元素事件签名EventEncoder.encode的结果与链上topics[0]完全一致构造的过滤器能查回同一事件。这套编码 → 比对 → 解码 → 过滤回查的流程就是手工日志解析的完整闭环。四、路径三WebSocket 推送订阅newHeads 等需要推送式订阅而非轮询时使用 WebSocket。官方流程new WebSocketService(wsUrl, includeRawResponses)webSocketService.connect()Web3j.build(webSocketService)基于该服务构建客户端订阅newHeadsNotifications()或其他推送 Flowable。4.1 底层原理从 eth_subscribe 到 Flowable看 core/src/main/java/org/web3j/protocol/core/JsonRpc2_0Web3j.java 的实现newHeadsNotifications()本质是对 JSON-RPC 订阅的封装public FlowableNewHeadsNotification 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:8546HTTP 的 8545 对应 WebSocket 通常为 8546通知对象NewHeadsNotification extends NotificationNewHead通过notification.getParams().getResult()拿到NewHead见 Notification.java 与 NewHeadsNotification.javaNewHead 字段NewHead.javanumber、hash、parentHash、timestamp、miner、gasLimit、gasUsed、stateRoot、transactionRoot等模板打印number/hash/parentHash资源释放三件套subscription.dispose()→web3j.shutdown()→webSocketService.close()模板用 shutdown hook 保证 CtrlC 时执行。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.javaAPI行为replayPastBlocksFlowable(startBlock, fullTransactionObjects)从 startBlock 逐块回放到最新replayPastAndFutureBlocksFlowable(startBlock, fullTransactionObjects, pollingInterval)先回放历史再接实时blockFlowable续流replayPastTransactionsFlowable(startBlock)历史区块内的交易流replayPastAndFutureTransactionsFlowable(startBlock, pollingInterval)历史 实时交易流replayTransactionsFlowable(start, end)指定区间交易流实现要点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.javaEVMTest(type NodeType.BESU)中有直接覆盖testBlockFlowable、testPendingTransactionFlowable、testTransactionFlowable、testLogFlowable、testReplayFlowable、testReplayPastAndFutureBlocksFlowable六个用例统一用CountDownLatch计数 超时断言 subscription.dispose()校验。六、关键概念速查概念说明相关源码FlowableRxJava 背压流类型web3j 所有事件订阅的载体JsonRpc2_0Rx.javaEthFilter日志过滤条件区块范围 地址 topicsEthFilter.javaethLogFlowable轮询式日志流HTTP 可用wrapper 事件方法的底层JsonRpc2_0Web3j.javaEventEncoder.encode事件签名 → topic0sha3EventEncoder.javanewHeadsNotificationsWebSocket 推送新区块头eth_subscribeJsonRpc2_0Web3j.javaWebSocketServiceWebSocket 传输层构造后需 connect()WebSocketService.javaindexed vs non-indexedindexed 参数进 topics最多 3 个其余进 dataContract.javadispose订阅释放配合web3j.shutdown()使用各模板的 shutdown hook七、仓库内进一步研读指南可运行模板总览.agents/skills/web3j/assets/event-subscriptions/README.mdERC20 端到端示例说明.agents/skills/web3j/assets/erc20-example/basic-erc20-app/README.md真实生成 wrappercontracts/src/main/java/org/web3j/contracts/eip20/generated/ERC20.java、contracts/src/main/java/org/web3j/contracts/eip721/generated/ERC721.javawrapper 生成逻辑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后端【免费下载链接】web3jLightweight Java and Android library for integration with Ethereum clients项目地址https://gitcode.com/gh_mirrors/we/web3j点击查看免费下载相关推荐Tendermint Core 事件订阅指南通过 WebSocket 订阅链上事件与交易Tendermint Core 事件订阅指南通过 WebSocket 订阅链上事件与交易 Tendermint Core 在区块提交、交易执行、验证人集合变化区块链共识算法Prisma 实时订阅Subscriptions完全指南从 WebSocket 协议到事件过滤Prisma 实时订阅Subscriptions完全指南从 WebSocket 协议到事件过滤 导读 GraphQL 订阅Subscriptions是后端数据库GraphQLPrisma Subscriptions 实时订阅 API 完整指南WebSocket 协议、类型订阅与过滤机制Prisma Subscriptions 实时订阅 API 完整指南WebSocket 协议、类型订阅与过滤机制 导读 在 Prisma 服务中GraphQ后端数据库GraphQL上一篇json-diff vs 其他JSON对比工具为什么它是最受欢迎的选择下一篇Remeda v1 迁移指南reject 函数已被移除如何用 filter isNot 完成等价替换创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考