保持主线程等待直到 RxJava Observable 完成

Question

我想保持我的主线程直到 RxJava 2 Observable 完成。我的编码在这里。情况是当我请求方法包含以下代码时，方法只是在执行而不等待可观察

 Observable
                        .fromArray(observableList)
                        .subscribeOn(Schedulers.io())
                        .subscribe(new Consumer<List<Observable<List<String>>>>() {
                        @Override
                        public void accept(List<Observable<List<String>>> list) throws Exception {

                            for (Observable<List<String>> observable : list) {
                                System.out.println("Thread name " + Thread.currentThread().getName());

                                observable.subscribe(new Consumer<List<String>>() {
                                    @Override
                                    public void accept(List<String> t) throws Exception {

                                        Gson jsonBuilder = new Gson();

                                        Object obj = new Object();

                                        JsonElement element = jsonBuilder.toJsonTree(obj);

                                        element.getAsJsonObject().addProperty(t.get(0), t.get(1));

                                        Gson g = new Gson();
                                        Object out = g.fromJson(t.get(1), Object.class);

                                        microResponses.put(t.get(0), out);
                                    }
                                });

                            }

                        }


         });

Answer 1

使用以下任意一项：- 1.删除订阅 2. 使用 Thread.sleep 足够长的时间来完成您的任务 3. Google toBlocking 在 rxjava 中的用法

Answer 2

你不需要多级订阅，你不应该需要嵌套你的 Observables:

Oservable
.fromArray(observableList)
.flatMapIterale(i -> i)
.flatMap(i -> i)
.doOnNext(t -> {
   Gson jsonBuilder = new Gson();
   Object obj = new Object();
   JsonElement element = jsonBuilder.toJsonTree(obj);
   element.getAsJsonObject().addProperty(t.get(0), t.get(1));
   Gson g = new Gson();
   Object out = g.fromJson(t.get(1), Object.class);
   microResponses.put(t.get(0), out);
})
// .subscribe(...) or .doOnComplete(...).subscribe();
;

但是，您编写的代码存在几个问题：

前四行是空操作，因为您没有使用 JsonElement。
为什么不直接使用new JsonObject()？
Object out = g.fromJson(t.get(1), Object.class); 除了一个空的 JsonObjecy 之外不会产生任何东西，因为 Object 没有任何属性。
为什么要使用 List，而不是真正有意义的东西，例如 Map.Entry<String,Object>，甚至 Pair<String, Object>？

保持主线程等待直到 RxJava Observable 完成

Keep main thread wait until RxJava Observable complete

java

multithreading

rx-java

rx-java2