Flowable 无法通过 onExceptionResumeNext 从异常中恢复

【问题标题】：Flowable does not recover from exception with onExceptionResumeNextFlowable 无法通过 onExceptionResumeNext 从异常中恢复
【发布时间】：2023-01-30 20:58:50
【问题描述】：

我正在使用 web3j 来收听来自区块链的实时事件（基于 web3j 附带的 rx-java2 ver 2.2.21）

它工作正常，但我有一个问题，

即使有例外，我也想保留实时订阅，我是这样实现的：

  web3j.ethLogFlowable(ethFilter)
                .onExceptionResumeNext(Flowable.empty())
                .subscribe(this::produceMessage);

我使用以下测试在 unitest 中对其进行了测试（Log 是应从 ethLogFlowable 返回的类）：

 Log log = mock(Log.class);
 doReturn(Flowable.just(log, "test", log), Flowable.just(log, log)).when(web3j).ethLogFlowable(any(EthFilter.class));

如您所见，"test" 字符串会导致异常，它确实会导致异常，问题是它正在停止流，而不是像我预期的那样继续下一步：

io.reactivex.exceptions.OnErrorNotImplementedException: The exception was not handled due to missing onError handler in the subscribe() method call. Further reading: https://github.com/ReactiveX/RxJava/wiki/Error-Handling | java.lang.ClassCastException: class java.lang.String cannot be cast to class org.web3j.protocol.core.methods.response.Log (java.lang.String is in module java.base of loader 'bootstrap'; org.web3j.protocol.core.methods.response.Log is in unnamed module of loader 'app')
    at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:704)
    at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:701)
    at io.reactivex.internal.subscribers.LambdaSubscriber.onError(LambdaSubscriber.java:79)
    at io.reactivex.internal.subscribers.LambdaSubscriber.onNext(LambdaSubscriber.java:69)
    at io.reactivex.internal.operators.flowable.FlowableOnErrorNext$OnErrorNextSubscriber.onNext(FlowableOnErrorNext.java:80)
    at io.reactivex.internal.operators.flowable.FlowableFromArray$ArraySubscription.fastPath(FlowableFromArray.java:132)
    at io.reactivex.internal.operators.flowable.FlowableFromArray$BaseArraySubscription.request(FlowableFromArray.java:88)
    at io.reactivex.internal.subscriptions.SubscriptionArbiter.setSubscription(SubscriptionArbiter.java:99)
    at io.reactivex.internal.operators.flowable.FlowableOnErrorNext$OnErrorNextSubscriber.onSubscribe(FlowableOnErrorNext.java:69)
    at io.reactivex.internal.operators.flowable.FlowableFromArray.subscribeActual(FlowableFromArray.java:38)
    at io.reactivex.Flowable.subscribe(Flowable.java:14935)
    at io.reactivex.internal.operators.flowable.FlowableOnErrorNext.subscribeActual(FlowableOnErrorNext.java:40)
    at io.reactivex.Flowable.subscribe(Flowable.java:14935)
    at io.reactivex.Flowable.subscribe(Flowable.java:14872)
    at io.reactivex.Flowable.subscribe(Flowable.java:14760)
    at com.citi.nexus.ethereumadaptor.events.EthEventListener.subscribe(EthEventListener.java:120)
    at com.citi.nexus.ethereumadaptor.events.EthEventListener.startLiveSubscription(EthEventListener.java:90)
    at com.citi.nexus.ethereumadaptor.events.EthEventListener.startSubscription(EthEventListener.java:79)
    at com.citi.nexus.ethereumadaptor.events.EthEventListener.executeOnApplicationReady(EthEventListener.java:62)
    at com.citi.nexus.ethereumadaptor.events.EthEventListenerTest.executeOnApplicationReadySyncLiveWithError(EthEventListenerTest.java:106)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:566)
    at org.junit.platform.commons.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:688)
    at org.junit.jupiter.engine.execution.MethodInvocation.proceed(MethodInvocation.java:60)
    at org.junit.jupiter.engine.execution.InvocationInterceptorChain$ValidatingInvocation.proceed(InvocationInterceptorChain.java:131)
    at org.junit.jupiter.engine.extension.TimeoutExtension.intercept(TimeoutExtension.java:149)
    at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestableMethod(TimeoutExtension.java:140)
    at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestMethod(TimeoutExtension.java:84)
    at org.junit.jupiter.engine.execution.ExecutableInvoker$ReflectiveInterceptorCall.lambda$ofVoidMethod$0(ExecutableInvoker.java:115)
    at org.junit.jupiter.engine.execution.ExecutableInvoker.lambda$invoke$0(ExecutableInvoker.java:105)
    at org.junit.jupiter.engine.execution.InvocationInterceptorChain$InterceptedInvocation.proceed(InvocationInterceptorChain.java:106)
    at org.junit.jupiter.engine.execution.InvocationInterceptorChain.proceed(InvocationInterceptorChain.java:64)
    at org.junit.jupiter.engine.execution.InvocationInterceptorChain.chainAndInvoke(InvocationInterceptorChain.java:45)
    at org.junit.jupiter.engine.execution.InvocationInterceptorChain.invoke(InvocationInterceptorChain.java:37)
    at org.junit.jupiter.engine.execution.ExecutableInvoker.invoke(ExecutableInvoker.java:104)
    at org.junit.jupiter.engine.execution.ExecutableInvoker.invoke(ExecutableInvoker.java:98)
    at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.lambda$invokeTestMethod$6(TestMethodTestDescriptor.java:210)
    at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
    at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.invokeTestMethod(TestMethodTestDescriptor.java:206)
    at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.execute(TestMethodTestDescriptor.java:131)
    at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.execute(TestMethodTestDescriptor.java:65)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$5(NodeTestTask.java:139)
    at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$7(NodeTestTask.java:129)
    at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:127)
    at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:126)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:84)
    at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)
    at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.invokeAll(SameThreadHierarchicalTestExecutorService.java:38)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$5(NodeTestTask.java:143)
    at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$7(NodeTestTask.java:129)
    at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:127)
    at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:126)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:84)
    at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)
    at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.invokeAll(SameThreadHierarchicalTestExecutorService.java:38)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$5(NodeTestTask.java:143)
    at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$7(NodeTestTask.java:129)
    at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:127)
    at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:126)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:84)
    at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.submit(SameThreadHierarchicalTestExecutorService.java:32)
    at org.junit.platform.engine.support.hierarchical.HierarchicalTestExecutor.execute(HierarchicalTestExecutor.java:57)
    at org.junit.platform.engine.support.hierarchical.HierarchicalTestEngine.execute(HierarchicalTestEngine.java:51)
    at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:108)
    at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:88)
    at org.junit.platform.launcher.core.EngineExecutionOrchestrator.lambda$execute$0(EngineExecutionOrchestrator.java:54)
    at org.junit.platform.launcher.core.EngineExecutionOrchestrator.withInterceptedStreams(EngineExecutionOrchestrator.java:67)
    at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:52)
    at org.junit.platform.launcher.core.DefaultLauncher.execute(DefaultLauncher.java:96)
    at org.junit.platform.launcher.core.DefaultLauncher.execute(DefaultLauncher.java:75)
    at com.intellij.junit5.JUnit5IdeaTestRunner.startRunnerWithArgs(JUnit5IdeaTestRunner.java:71)
    at com.intellij.rt.junit.IdeaTestRunner$Repeater$1.execute(IdeaTestRunner.java:38)
    at com.intellij.rt.execution.junit.TestsRepeater.repeat(TestsRepeater.java:11)
    at com.intellij.rt.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:35)
    at com.intellij.rt.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:235)
    at com.intellij.rt.junit.JUnitStarter.main(JUnitStarter.java:54)
Caused by: java.lang.ClassCastException: class java.lang.String cannot be cast to class org.web3j.protocol.core.methods.response.Log (java.lang.String is in module java.base of loader 'bootstrap'; org.web3j.protocol.core.methods.response.Log is in unnamed module of loader 'app')
    at io.reactivex.internal.subscribers.LambdaSubscriber.onNext(LambdaSubscriber.java:65)
    ... 83 more

我可以添加一个 onError 方法，但在我希望它继续后流仍然会停止

有人可以向我解释我做错了什么吗？

P.S 我的解决方案是在 onException 中重新创建 flowable 它确实有效但我认为它不太优雅

【问题讨论】：

  • 如果异常进入 RxJava，那将停止序列。 onErrorResumeNext 可以将该异常转换为其他序列。你必须抑制异常，这样 RxJava 就不知道它，或者你必须使用 retry 运算符之一来重新订阅。

标签： java rx-java2 web3-java


【解决方案1】：

OnErrorNotImplementedException 被抛出是因为您没有实现 onError 处理程序，如消息中所述。

基本上，而不是

  web3j.ethLogFlowable(ethFilter)
                .onExceptionResumeNext(Flowable.empty())
                .subscribe(this::produceMessage);

你可以有：

  web3j.ethLogFlowable(ethFilter)
                .onExceptionResumeNext(Flowable.empty())
                .subscribe(
                      this::produceMessage,
                      error -> { 
                            // log some message ...
                            error.printStackTrace();
                      });

看。 http://reactivex.io/RxJava/3.x/javadoc/io/reactivex/rxjava3/core/Flowable.html#subscribe-io.reactivex.rxjava3.functions.Consumer-io.reactivex.rxjava3.functions.Consumer-

【讨论】：

    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 2016-05-05
    • 2015-01-29
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    相关资源
    最近更新 更多
    热门标签
    Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode