RxJava：如何从可观察到的

Question

我正在从事一个涉及Hystrix的项目，我决定使用RxJava。现在，忘记Hystrix的其余部分，因为我相信主要问题是我完全搞砸了正确编写可观察代码。

需要：我需要一种方法来返回一个代表多个可观察对象的可观察对象，每个可观察对象都运行一个用户任务。我希望该可观察对象能够返回任务的所有结果，甚至错误。

问题：可观测流会因错误而消亡。如果我有三个任务，而第二个任务引发了一个异常，那么即使第三个任务成功了，我也不会收到它。

我的代码：

public <T> Observable<T> observeManagedAsync(String groupName,List<EspTask<T>> tasks) {
    return Observable
            .from(tasks)
            .flatMap(task -> {
                try {
                    return new MyCommand(task.getTaskId(),groupName,task).toObservable().subscribeOn(this.schedulerFactory.get(groupName));
                } catch(Exception ex) {
                    return Observable.error(ex);
                }
            });
}

考虑到MyCommand是一个扩展HystrixObservableCommand的类，它返回一个可观察的，因此不应该涉及我所看到的问题。

尝试1：

使用Observable.flatMap如上

• 好：每个命令都调度在自己的线程上，任务异步运行。
• 坏：在第一个命令异常时，可观察完成发出先前的成功结果并发出异常。任何飞行中的命令都将被忽略。

尝试2：

使用<代码>可观察。concatMapDelayError，而不是flatMap

• 坏：由于某种原因，任务同步运行。为什么？？
• 好：我得到了所有成功的结果。
• ~好：OnError获取一个复合异常，其中包含抛出的异常列表。

任何帮助都会非常感激，而且可能会因为我自己没有想到而感到非常尴尬。

附加代码

此测试成功，可观察到<代码>。flatMap，但在使用Observable时失败。concatMapDelayError因为任务不是异步运行的：

java.lang.AssertionError：执行时间超过350ms限制：608

@Test
public void shouldRunManagedAsyncTasksConcurrently() throws Exception {
    Observable<String> testObserver = executor.observeManagedAsync("asyncThreadPool",getTimedTasks()); 
    TestSubscriber<String> testSubscriber = new TestSubscriber<>();
    long startTime = System.currentTimeMillis();
    testObserver.doOnError(throwable -> {
        System.out.println("error: " + throwable.getMessage());
    }).subscribe(testSubscriber);
    System.out.println("Test execution time: "+(System.currentTimeMillis()-startTime));
    testSubscriber.awaitTerminalEvent();
    long execTime = (System.currentTimeMillis()-startTime);
    System.out.println("Test execution time: "+execTime);
    testSubscriber.assertCompleted();
    System.out.println("Errors: "+testSubscriber.getOnErrorEvents());
    System.out.println("Results: "+testSubscriber.getOnNextEvents());
    testSubscriber.assertNoErrors();
    assertTrue("Execution time ran under the 300ms limit: "+execTime,execTime>=300);
    assertTrue("Execution time ran over the 350ms limit: "+execTime,execTime<=350);
    testSubscriber.assertValueCount(3);
    assertThat(testSubscriber.getOnNextEvents(),containsInAnyOrder("hello","wait","world"));
    verify(this.mockSchedulerFactory, times(3)).get("asyncThreadPool");
}

上述单元测试的任务：

protected List<EspTask<String>> getTimedTasks() {
    EspTask longTask = new EspTask("helloTask") {
        @Override
        public Object doCall() throws Exception {
            Thread.currentThread().sleep(100);
            return "hello";
        }
    };
    EspTask longerTask = new EspTask("waitTask") {
        @Override
        public Object doCall() throws Exception {
            Thread.currentThread().sleep(150);
            return "wait";
        }

    };
    EspTask longestTask = new EspTask("worldTask") {
        @Override
        public Object doCall() throws Exception {
            Thread.currentThread().sleep(300);
            return "world";
        }
    };
    return Arrays.asList(longTask, longerTask, longestTask);
}

Answer 1

使用.料化（）允许所有排放和错误作为包装通知通过，然后按照您的意愿处理它们：

     .flatMap(task -> {
            try {
                return new MyCommand(task.getTaskId(),groupName,task)
                    .toObservable()
                    .subscribeOn(this.schedulerFactory.get(groupName))
                    .materialize();
            } catch(Exception ex) {
                return Observable.error(ex).materialize();
            }
        });

Answer 2

您想使用mergeDelayError：

public <T> Observable<T> observeManagedAsync(String groupName,List<EspTask<T>> tasks) {
    return Observable.mergeDelayError(Observable
        .from(tasks)
        .map(task -> {
            try {
                return new MyCommand(task.getTaskId(),groupName,task).toObservable().subscribeOn(this.schedulerFactory.get(groupName));
            } catch(Exception ex) {
                return Observable.error(ex);
            }
        }));
}

请注意，MyCommand构造函数不应引发任何异常；这样可以更简洁地编写代码：

public <T> Observable<T> observeManagedAsync(String groupName,List<EspTask<T>> tasks) {
    return from(tasks)
           .map(task -> new MyCommand(task.getTaskId(), groupName, task)
                        .toObservable()
                        .subscribeOn(this.schedulerFactory.get(groupName)))
           .compose(Observable::mergeDelayError);

}

请记住，这最多只能调用一次onError；如果需要显式处理所有错误，请使用类似于

Answer 3

您可以使用Observable.on错误返回（），并返回特殊值（例如null），然后过滤下游的非特殊值。请记住，源可观察将在错误时完成。也取决于用例Observable.onErrorResumeNext（）方法也很有用。如果您对错误通知感兴趣，请使用Observable.materialize（），这将将项目和onError（）、on完整（）转换为通知，然后可以由Notification.getKind（）过滤

编辑应在<代码>之后立即添加上述所有运算符。toObservable（）。subscribeOn（this.schedulerFactory.get（groupName）） 假设没有尝试/捕获。