java - 具有太多 Observable 的 RxJava1 StackOverflow 异常

coder 2024-03-13 原文

我正在使用 rxjava1 开发一个项目，我有一个 Observable 链，偶尔会包含数千个合并或连接在一起的 observable。当发生这种情况时，将发生 StackOverflow 异常，我们将得到如下信息:

java.lang.StackOverflowError
    at java.util.HashMap.putVal(HashMap.java:631)
    at java.util.HashMap.put(HashMap.java:612)
    at rx.internal.operators.OnSubscribeToMap$ToMapSubscriber.onNext(OnSubscribeToMap.java:127)
    at rx.internal.operators.OnSubscribeFilter$FilterSubscriber.onNext(OnSubscribeFilter.java:76)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.emitScalar(OperatorMerge.java:395)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.tryEmit(OperatorMerge.java:355)
    at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:846)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.emitScalar(OperatorMerge.java:395)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.tryEmit(OperatorMerge.java:355)
    at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:846)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.emitScalar(OperatorMerge.java:395)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.tryEmit(OperatorMerge.java:355)
    at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:846)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.emitScalar(OperatorMerge.java:395)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.tryEmit(OperatorMerge.java:355)
    at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:846)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.emitScalar(OperatorMerge.java:395)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.tryEmit(OperatorMerge.java:355)

堆栈跟踪将持续数百行。我看到的唯一相关帖子是 github 中的这个问题:https://github.com/ReactiveX/RxJava/issues/3035 .但是，将可观察对象添加到列表中的建议解决方案是我们已经使用过的，但行不通。

我能做些什么来防止这些 StackOverflow 异常？我是否需要进行某种节流或背压？

这是当前代码的样子并导致计算器溢出的示例:

public Observable<Map<String, JsonObject>> extractTopLevelSummariesFromForms(JsonArray summaries, Func2<String, String, Observable<JsonObject>> summaryGatherer) {
    List<Observable<JsonObject>> summaryObservables = new LinkedList<>();
    summaries.stream()
             .map(JsonUtil::safeJsonObject)
             .filter(summary -> StringUtils.isNotEmpty(summary.getString(NAME))|| StringUtils.isNotEmpty(summary.getString(Form.TITLE)))
             .forEach(summary -> {
                 if (StringUtils.isNotEmpty(summary.getString(TEXT)))
                     summaryObservables.add(gatherSummariesFromElement(summary.getString(Summary.SHORT_NAME), Summary.SummaryValues.FORM,
                                                                       summary.getString(Summary.SHORT_NAME) + ".hidden",
                                                                       summary.getString(VALUE), summaryGatherer));
                 if (StringUtils.isNotEmpty(summary.getString(Form.TEXT)))
                     summaryObservables.add(gatherSummariesFromElement(summary.getString(Summary.SHORT_NAME), Summary.SummaryValues.FORM,
                                                                       summary.getString(Summary.SHORT_NAME) + ".title",
                                                                       summary.getString(Summary.VALUE), summaryGatherer, true));
             });
    return Observable.merge(Observable.from(summaryObservables))
                     .filter(summaryResult -> summaryResult != null)
                     .toMap(summaryResult -> summaryResult.getString(KEY), summaryResult -> summaryResult.getJsonObject(TEXT));
}

 private Observable<JsonObject> gatherSummariesFromElement(String parentName, String parentType, String elementName, String summaryValue, Func2<String, String, Observable<JsonObject>> summaryGatherer, Set<String> visited, boolean isList) {
    if (visited.contains(elementName))
        return Observable.just(null);
    visited.add(elementName);

    Map<String, JsonObject> summariesMap = new HashMap<>();

    summariesMap.put(elementName, new JsonObject().put(Summary.SummaryValues.FORM, form).put(SUMMARY_TYPE, parentType));
    Set<String> variables = TextEngine.getVariables(summariesMap);

    Observable<JsonObject> elementSummaryObservable = Observable.just(getSummaryEntry(elementName, form, parentType, isList));

    if (variables != null && !variables.isEmpty()) {
        elementSummaryObservable = elementSummaryObservable.mergeWith(Observable.from(variables).flatMap(variable -> {
            if (StringUtils.contains(variable, ".") && StringUtils.equals(parentName, StringUtils.split(variable, ".")[0]))
                return Observable.just(null);
            else
                return summaryGatherer.call(parentName, variable).flatMap(variableEntry -> {
                    if (variableEntry == null)
                        return Observable.just(null);
                    else
                        return gatherSummariesFromElement(parentName, variableEntry.getString(SOURCE_TYPE), variable, variableEntry.getString(FORM), summaryGatherer, visited, variableEntry.getBoolean(Summary.SummaryValues.IS_LIST, false));
                });
        }));
    }
    return elementSummaryObservable;
}

我已经尝试在 Schedulers.computation() 调度程序中运行除网络请求之外的所有内容，那些在 Schedulers.io() 调度程序中运行，我是仍然得到计算器溢出:

Exception in thread "pool-26-thread-2" java.lang.IllegalStateException: Fatal Exception thrown on Scheduler.Worker thread.
at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:59)
at rx.internal.schedulers.ExecutorScheduler$ExecutorSchedulerWorker.run(ExecutorScheduler.java:107)
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.lang.StackOverflowError
at rx.internal.operators.OperatorMerge$MergeSubscriber.tryEmit(OperatorMerge.java:355)
at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:846)
at rx.internal.operators.OperatorMerge$MergeSubscriber.emitScalar(OperatorMerge.java:395)
at rx.internal.operators.OperatorMerge$MergeSubscriber.tryEmit(OperatorMerge.java:355)
at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:846)
at rx.internal.operators.OperatorMerge$MergeSubscriber.emitScalar(OperatorMerge.java:395)

最佳答案

您对 gatherSummariesFromElement 进行了递归调用，这可能会进行得非常深入
Observable.merge(Observable.from(summaryObservables)) 调用看起来很奇怪，你应该只使用 Observable.merge(summaryObservables)
您可以使用 Observable.empty()

Observable.just(null)

summaryObservables 的构建看起来有点矫枉过正。您可以构建有效摘要列表而不是然后在 flatMap 中处理它们
将 gatherSummariesFromElement 中的递归 Observable 创建替换为元素创建的递归列表，然后从该列表创建 observable

public Observable<Map<String, JsonObject>> extractTopLevelSummariesFromForms(JsonArray summaries, Func2<String, String, JsonObject> summaryGatherer) {
    List<JsonObject> validSummaries = new LinkedList<>();
    summaries.stream()
            .map(JsonUtil::safeJsonObject)
            .filter(summary -> StringUtils.isNotEmpty(summary.getString(NAME)) || StringUtils.isNotEmpty(summary.getString(Form.TITLE)))
            .forEach(validSummaries::add);
    Set<String> visited = new HashSet<>();
    return Observable.from(validSummaries)
            .flatMap(summary -> {
                if (StringUtils.isNotEmpty(summary.getString(TEXT)))
                    Observable.from(gatherSummariesFromElement(summary.getString(Summary.SHORT_NAME), Summary.SummaryValues.FORM,
                            summary.getString(Summary.SHORT_NAME) + ".hidden",
                            summary.getString(VALUE), visited, summaryGatherer)));
                if (StringUtils.isNotEmpty(summary.getString(Form.TEXT)))
                    Observable.from(gatherSummariesFromElement(summary.getString(Summary.SHORT_NAME), Summary.SummaryValues.FORM,
                            summary.getString(Summary.SHORT_NAME) + ".title",
                            summary.getString(Summary.VALUE), summaryGatherer, visited,true)));
            })
            .toMap(summaryResult -> summaryResult.getString(KEY), summaryResult -> summaryResult.getJsonObject(TEXT));
}

private List<JsonObject> gatherSummariesFromElement(String parentName, String parentType, String elementName, String summaryValue, Func2<String, String, JsonObject> summaryGatherer, Set<String> visited, boolean isList) {
    if (visited.contains(elementName))
        return Collections.emptyList();
    visited.add(elementName);
    List<JsonObject> result = new ArrayList<>()

    Map<String, JsonObject> summariesMap = new HashMap<>();

    summariesMap.put(elementName, new JsonObject().put(Summary.SummaryValues.FORM, form).put(SUMMARY_TYPE, parentType));
    Set<String> variables = TextEngine.getVariables(summariesMap);

    result.add(getSummaryEntry(elementName, form, parentType, isList));

    if (variables != null && !variables.isEmpty()) {
        for (String variable : variables) {
            if (StringUtils.contains(variable, ".") && StringUtils.equals(parentName, StringUtils.split(variable, ".")[0])) {
                // do nothing
            } else {
                JsonObject variableEntry = summaryGatherer.call(parentName, variable)
                if (variableEntry != null) {
                    result.addAll(gatherSummariesFromElement(parentName, variableEntry.getString(SOURCE_TYPE), variable, variableEntry.getString(FORM), summaryGatherer, visited, variableEntry.getBoolean(Summary.SummaryValues.IS_LIST, false));
                }
            }
        }
    }
    return result;
}

关于java - 具有太多 Observable 的 RxJava1 StackOverflow 异常，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/50753349/

有关java - 具有太多 Observable 的 RxJava1 StackOverflow 异常的更多相关文章

ruby - 具有身份验证的私有(private) Ruby Gem 服务器 - 2
我想安装一个带有一些身份验证的私有(private)Rubygem服务器。我希望能够使用公共(public)Ubuntu服务器托管内部gem。我读到了http://docs.rubygems.org/read/chapter/18.但是那个没有身份验证-如我所见。然后我读到了https://github.com/cwninja/geminabox.但是当我使用基本身份验证(他们在他们的Wiki中有)时，它会提示从我的服务器获取源。所以。如何制作带有身份验证的私有(private)Rubygem服务器？这是不可能的吗？谢谢。编辑:Geminabox问题。我尝试“捆绑”以安装新的gem..
java - 等价于 Java 中的 Ruby Hash - 2
我真的很习惯使用Ruby编写以下代码:my_hash={}my_hash['test']=1Java中对应的数据结构是什么？最佳答案 HashMapmap=newHashMap();map.put("test",1);我假设？关于java-等价于Java中的RubyHash，我们在StackOverflow上找到一个类似的问题： https://stackoverflow.com/questions/22737685/
ruby-on-rails - Rails - 乐观锁定总是触发 StaleObjectError 异常 - 2
我正在学习Rails，并阅读了关于乐观锁的内容。我已将类型为integer的lock_version列添加到我的articles表中。但现在每当我第一次尝试更新记录时，我都会收到StaleObjectError异常。这是我的迁移:classAddLockVersionToArticle当我尝试通过Rails控制台更新文章时:article=Article.first=>#我这样做:article.title="newtitle"article.save我明白了:(0.3ms)begintransaction(0.3ms)UPDATE"articles"SET"title"='dwdwd
ruby - ＃之间？ Cooper 的 *Beginning Ruby* 中的错误或异常 - 2
在Cooper的书BeginningRuby中，第166页有一个我无法重现的示例。classSongincludeComparableattr_accessor:lengthdef(other)@lengthother.lengthenddefinitialize(song_name,length)@song_name=song_name@length=lengthendenda=Song.new('Rockaroundtheclock',143)b=Song.new('BohemianRhapsody',544)c=Song.new('MinuteWaltz',60)a.betwee
ruby - 在 Ruby 中重新分配常量时抛出异常？ - 2
我早就知道Ruby中的“常量”(即大写的变量名)不是真正常量。与其他编程语言一样，对对象的引用是唯一存储在变量/常量中的东西。(侧边栏:Ruby确实具有“卡住”引用对象不被修改的功能，据我所知，许多其他语言都没有提供这种功能。)所以这是我的问题:当您将一个值重新分配给常量时，您会收到如下警告:>>FOO='bar'=>"bar">>FOO='baz'(irb):2:warning:alreadyinitializedconstantFOO=>"baz"有没有办法强制Ruby抛出异常而不是打印警告？很难弄清楚为什么有时会发生重新分配。最佳答案
java - 从 JRuby 调用 Java 类的问题 - 2
我正在尝试使用boilerpipe来自JRuby。我看过guide从JRuby调用Java，并成功地将它与另一个Java包一起使用，但无法弄清楚为什么同样的东西不能用于boilerpipe。我正在尝试基本上从JRuby中执行与此Java等效的操作:URLurl=newURL("http://www.example.com/some-location/index.html");Stringtext=ArticleExtractor.INSTANCE.getText(url);在JRuby中试过这个:require'java'url=java.net.URL.new("http://www
ruby-on-rails - Rails 3.1 中具有相同形式的多个模型？ - 2
我正在使用Rails3.1并在一个论坛上工作。我有一个名为Topic的模型，每个模型都有许多Post。当用户创建新主题时，他们也应该创建第一个Post。但是，我不确定如何以相同的形式执行此操作。这是我的代码:classTopic:destroyaccepts_nested_attributes_for:postsvalidates_presence_of:titleendclassPost...但这似乎不起作用。有什么想法吗？谢谢! 最佳答案 @Pablo的回答似乎有你需要的一切。但更具体地说...首先改变你View中的这一行对此#
java - 我的模型类或其他类中应该有逻辑吗 - 2
我只想对我一直在思考的这个问题有其他意见，例如我有classuser_controller和classuserclassUserattr_accessor:name,:usernameendclassUserController//dosomethingaboutanythingaboutusersend问题是我的User类中是否应该有逻辑user=User.newuser.do_something(user1)oritshouldbeuser_controller=UserController.newuser_controller.do_something(user1,user2)我
java - 什么相当于 ruby 的 rack 或 python 的 Java wsgi？ - 2
什么是ruby的rack或python的Java的wsgi？还有一个路由库。最佳答案来自Python标准PEP333:Bycontrast,althoughJavahasjustasmanywebapplicationframeworksavailable,Java's"servlet"APImakesitpossibleforapplicationswrittenwithanyJavawebapplicationframeworktoruninanywebserverthatsupportstheservletAPI.ht
Observability：从零开始创建 Java 微服务并监控它（二） - 2
这篇文章是继上一篇文章“Observability：从零开始创建Java微服务并监控它（一）”的续篇。在上一篇文章中，我们讲述了如何创建一个Javaweb应用，并使用Filebeat来收集应用所生成的日志。在今天的文章中，我来详述如何收集应用的指标，使用APM来监控应用并监督web服务的在线情况。源码可以在地址 https://github.com/liu-xiao-guo/java_observability 进行下载。摄入指标指标被视为可以随时更改的时间点值。当前请求的数量可以改变任何毫秒。你可能有1000个请求的峰值，然后一切都回到一个请求。这也意味着这些指标可能不准确，你还想提取最小/

java - 具有太多 Observable 的 RxJava1 StackOverflow 异常

有关java - 具有太多 Observable 的 RxJava1 StackOverflow 异常的更多相关文章

随机推荐