web3j v4.8.7实现区块事件监听时出现The filter has not found错误咨询

我尝试使用web3j（v4.8.7）实现区块链事件读取功能，编写的代码如下：

@Slf4j
@Component
public class ContractEventSubscriber {

    @Autowired
    Web3j web3j;


    @PostConstruct
    public void init() {
        log.info("Initializing...");
        web3j.blockFlowable(false).subscribe(
                block -> {
                    log.info("New block {}", block);
                }, error -> {
                    log.error("Event error: {}", error, error);
                });

        log.info("Initialized.");
    }

}

初始化方法执行即失败，抛出The filter has not been found警告，完整报错日志如下：

2021-08-26 14:40:23,723 [main] INFO  com.xcompany.web3j.ContractEventSubscriber.init(43) -Initializing...
2021-08-26 14:40:25,134 [pool-2-thread-1] WARN  org.web3j.protocol.core.filters.Filter.reinstallFilter(153) -The filter has not been found. Filter id: 330683721788227458774458119455366581270
Logging initialized using 'class org.apache.ibatis.logging.stdout.StdOutImpl' adapter.
2021-08-26 14:40:25,420 [pool-2-thread-1] ERROR org.web3j.protocol.core.filters.Filter.lambda$run$0(97) -Error sending request
org.web3j.protocol.core.filters.FilterException: Error sending request
    at org.web3j.protocol.core.filters.Filter.throwException(Filter.java:194)
    at org.web3j.protocol.core.filters.Filter.run(Filter.java:104)
    at org.web3j.protocol.core.filters.Filter.reinstallFilter(Filter.java:155)
    at org.web3j.protocol.core.filters.Filter.pollFilter(Filter.java:137)
    at org.web3j.protocol.core.filters.Filter.lambda$run$0(Filter.java:92)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.io.InterruptedIOException: interrupted
    at okio.Timeout.throwIfReached(Timeout.kt:98)
    at okio.OutputStreamSink.write(Okio.kt:53)
    at okio.AsyncTimeout$sink$1.write(AsyncTimeout.kt:103)
    at okio.RealBufferedSink.flush(RealBufferedSink.kt:247)
    at okhttp3.internal.http1.Http1ExchangeCodec.finishRequest(Http1ExchangeCodec.kt:158)
    at okhttp3.internal.connection.Exchange.finishRequest(Exchange.kt:90)
    at okhttp3.internal.http.CallServerInterceptor.intercept(CallServerInterceptor.kt:76)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.kt:112)
    at okhttp3.internal.connection.ConnectInterceptor.intercept(ConnectInterceptor.kt:37)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.kt:112)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.kt:87)
    at okhttp3.internal.cache.CacheInterceptor.intercept(CacheInterceptor.kt:82)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.kt:112)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.kt:87)
    at okhttp3.internal.http.BridgeInterceptor.intercept(BridgeInterceptor.kt:84)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.kt:112)
    at okhttp3.internal.http.RetryAndFollowUpInterceptor.intercept(RetryAndFollowUpInterceptor.kt:71)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.kt:112)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.kt:87)
    at com.huobi.pool.hedging.strategy.contract.config.RetryInterceptor.intercept(RetryInterceptor.java:25)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.kt:112)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.kt:87)
    at okhttp3.RealCall.getResponseWithInterceptorChain(RealCall.kt:194)
    at okhttp3.RealCall.execute(RealCall.kt:67)
    at org.web3j.protocol.http.HttpService.performIO(HttpService.java:165)
    at org.web3j.protocol.Service.send(Service.java:48)
    at org.web3j.protocol.core.Request.send(Request.java:87)
    at org.web3j.protocol.core.filters.BlockFilter.sendRequest(BlockFilter.java:34)
    at org.web3j.protocol.core.filters.Filter.run(Filter.java:59)
    ... 10 common frames omitted

1. 底层HTTP连接被主动中断：报错栈底部的java.io.InterruptedIOException: interrupted说明web3j和链节点的HTTP请求被打断，HTTP模式下的filter依赖轮询维持状态，一旦请求超时、节点断开或者自定义拦截器主动断连，节点侧已经创建的filter就会失效，直接触发filter未找到的警告。
2. blockFlowable默认基于eth_newFilter接口实现，大部分公链节点对长期无轮询的filter会自动超时回收，第三方节点服务很多会默认限制filter接口权限或者缩短filter存活时间，也会导致该问题。
3. 日志中的自定义RetryInterceptor拦截器可能存在超时断连逻辑，提前终止了web3j的轮询请求。

方案1：改用WebSocket协议连接节点 #

HTTP协议本身不适合长订阅场景，切换成WebSocket连接后，filter状态会随连接保持，不会被节点频繁回收，修改Web3j初始化逻辑即可：

// 替换原有HttpService初始化方式，使用WebSocketService
Web3j web3j = Web3j.build(new WebSocketService("wss://你的节点WebSocket地址", false));

如果是Spring注入的Web3j实例，直接修改对应Bean的初始化配置。

方案2：改用手动轮询拉块逻辑，完全规避filter依赖 #

如果节点不支持WebSocket或者filter接口被限制，直接手动轮询拉取区块，无需依赖节点的filter能力，稳定性更高：

@Slf4j
@Component
public class ContractEventSubscriber {
    @Autowired
    Web3j web3j;
    // 记录已处理的最新区块高度
    private volatile Long lastProcessedBlock = 0L;
    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

    @PostConstruct
    public void init() {
        log.info("Initializing block subscriber...");
        // 轮询间隔可根据对应链的出块速度调整，此处示例为3秒
        scheduler.scheduleAtFixedRate(() -> {
            try {
                Long latestBlock = web3j.ethBlockNumber().send().getBlockNumber().longValue();
                if (latestBlock > lastProcessedBlock) {
                    // 遍历处理所有遗漏区块
                    for (long blockNum = lastProcessedBlock + 1; blockNum <= latestBlock; blockNum++) {
                        EthBlock block = web3j.ethGetBlockByNumber(
                                DefaultBlockParameter.valueOf(BigInteger.valueOf(blockNum)), 
                                true
                        ).send();
                        log.info("New block received: {}", block);
                        // 此处可添加自定义事件解析逻辑
                    }
                    lastProcessedBlock = latestBlock;
                }
            } catch (Exception e) {
                log.error("Pull block failed: {}", e.getMessage(), e);
            }
        }, 0, 3, TimeUnit.SECONDS);
        log.info("Block subscriber initialized.");
    }

    @PreDestroy
    public void destroy() {
        scheduler.shutdown();
    }
}

方案3：调整HTTP模式配置适配filter规则 #

如果必须使用HTTP模式的Flowable，调整web3j的轮询间隔和拦截器规则：

1. 增大web3j的轮询间隔，匹配节点的filter超时时间，默认轮询间隔为15秒，可按需调整：

// 初始化Web3j时指定轮询间隔，单位为毫秒，此处示例为5秒
Web3j web3j = Web3j.build(new HttpService("你的节点HTTP地址"), 5000, Async.defaultExecutorService());

2. 检查自定义RetryInterceptor的逻辑，不要主动打断web3j的长轮询请求，适当调整拦截器的超时阈值。

内容的提问来源于stack exchange，提问作者Paul Verest on LinkedIn