使用 RxJava 组合具有依赖关系的异步 Observable

Question

我是响应式编程的新手，对编写具有依赖关系的可观察对象感到困惑。这是场景： 有两个可观察对象 A、B。可观察的 A 取决于 B 发出的值。 （因此A需要观察B）。有没有办法创建一个由 A 和 B 组成并发出 V 的 Observable C？我只是在 RxJava documentation 中寻找指针。

Answer 1

我也是响应式编程的新手，只是整理了一些可能对您的案例感兴趣的代码

A需要观察B

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.annotations.Test;
import rx.Observable;
import rx.Subscriber;
import rx.functions.Func1;

import java.util.concurrent.atomic.AtomicBoolean;

import static org.testng.Assert.assertTrue;

public class Q22284380TestCase {

    private static final Logger LOGGER = LoggerFactory.getLogger(
            Q22284380TestCase.class);

    private AtomicBoolean completed = new AtomicBoolean(false);

    @Test
    public void testName() throws Exception {

        final Observable.OnSubscribe<Integer> onSubProduceTwoValues = new Observable.OnSubscribe<Integer>() {

            @Override
            public void call(final Subscriber<? super Integer> subscriber) {

                final Thread thread = new Thread(new Runnable() {

                    public Integer i = 0;

                    @Override
                    public void run() {

                        final Integer max = 2;
                        while (i < max) {
                            subscriber.onNext(i);
                            i++;
                        }

                        subscriber.onCompleted();
                    }
                });

                thread.start();
            }
        };

        final Observable<Integer> values = Observable.create(onSubProduceTwoValues);

        final Observable<Integer> byTwoMultiplier = values
                .flatMap(new Func1<Integer, Observable<Integer>>() {

                    @Override
                    public Observable<Integer> call(Integer aValue) {

                        return doubleIt(aValue);

                    }
                });

        byTwoMultiplier.subscribe(new Subscriber<Integer>() {

            @Override
            public void onNext(Integer a) {

                LOGGER.info("" + a);

            }

            @Override
            public void onCompleted() {

                completed.set(true);

            }

            @Override
            public void onError(Throwable e) {

                LOGGER.error(e.getMessage());
            }
        });

        Thread.sleep(1000L);
        assertTrue(completed.get());

    }

    private Observable<Integer> doubleIt(final Integer value) {

        return Observable.create(new Observable.OnSubscribe<Integer>() {

            @Override
            public void call(final Subscriber<? super Integer> subscriber) {

                final Thread thread = new Thread(new Runnable() {

                    @Override
                    public void run() {

                        try {
                            subscriber.onNext(value * 2);
                            subscriber.onCompleted();
                        } catch (Throwable e) {
                            subscriber.onError(e);
                        }

                    }
                });

                thread.start();

            }
        });
    }
}

拥有一个值的生产者，它只使用flatMap 将一个doubleIt 函数应用于输出。 要做一些不同的事情，如果你想要一个由 A 和 B 组合而成的 V，你可以阅读 zip。

Answer 2

我认为这取决于你需要做的 A 和 B 之间的组合类型以及 A 如何依赖 B。

C 是否将 A 和 B 组合成对（A1 与 B1 组合，A2 与 B2 组合等） - 那么zip 将是您想要的功能。但是，在那种情况下，我想知道当你首先将 B 转换为 A 时，你是否不能只做这项工作 - 毕竟我假设你将 B 逐个元素转换为 A （在这种情况下， map 将是要走的路)。

如果您想为 B 发出的每个值创建一个新的 A（但想将所有这些 As 合并到一个 Observable 中），那么您需要 flatMap。

如果您确实首先需要 B 来创建 A，然后再次需要它来组合 A 和 B，那么您可能需要 cacheB 来为您保存再次计算一切的麻烦。

这里还有其他有趣的功能（例如reduce 或combineLatest）。或许你可以提供更多关于你想要做什么的细节？

Answer 3

你的问题对于 A 如何依赖于 B 有点含糊，所以我将尝试举几个例子来说明如何组合 observables。

示例 - 没有 B 就无法创建 A - 使用 map()

public class B {
    public final int value;
    public B(int value) {
        this.value = value;
    }
}

public class A {
    public final B b;
    public A(B b) {
        this.b = b;
    }
}

public Observable<B> createObservableB() {
    return Observable.from(new B(0), new B(1), new B(2), new B(3));
}

public Observable<A> createObservableA() {
    return createObservableB()
            .map(new Func1<B, A>() {
                @Override
                public A call(B b) {
                    return new A(b);
                }
            });
}

示例 - B 的每次出现都可以创建零个或多个 A - 使用 flatMap()

public class B {
    public final int value;
    public B(int value) {
        this.value = value;
    }
}

public class A {
    public final int value;
    public A(int value) {
        this.value = value;
    }
}

public Observable<B> createObservableB() {
    return Observable.from(new B(0), new B(1), new B(2), new B(3));
}

public Observable<A> createObservableA() {
    return createObservableB()
            .flatMap(new Func1<B, Observable<? extends A>>() {
                @Override
                public Observable<? extends A> call(final B b) {
                    return Observable.create(new Observable.OnSubscribe<A>() {
                        @Override
                        public void call(Subscriber<? super A> subscriber) {
                            for (int i = 0; i < b.value; i++) {
                                subscriber.onNext(new A(i));
                            }
                            subscriber.onCompleted();
                        }
                    });
                }
            });
}

我不太清楚你对 Observables C 和 V 的要求是什么，所以让我们看看更多组合 observables 的方法。

示例 - 组合两个可观察对象发出的每对项目 - 使用 zip()

public class A {
    public final int value;
    public A(int value) {
        this.value = value;
    }
}

public class B {
    public final int value;
    public B(int value) {
        this.value = value;
    }
}

public class C {
    private final A a;
    private final B b;
    public C(A a, B b) {
        this.a = a;
        this.b = b;
    }
}

public Observable<B> createObservableB() {
    return Observable.from(new B(0), new B(1), new B(2), new B(3));
}

public Observable<A> createObservableA() {
    return Observable.from(new A(0), new A(1), new A(2), new A(3));
}

public Observable<C> createObservableC() {
    return Observable.zip(createObservableA(), createObservableB(),
            new Func2<A, B, C>() {
                @Override
                public C call(A a, B b) {
                    return new C(a, b);
                }
            }
    );
}

示例 - 合并两个 Observable 的最后一项 - 使用 combineLatest()

// Use the same class definitions from previous example.
public Observable<C> createObservableC1() {
    return Observable.combineLatest(createObservableA(), createObservableB(),
            new Func2<A, B, C>() {
                @Override
                public C call(A a, B b) {
                    return new C(a, b);
                }
            }
    );
}

Answer 4

如果您正在寻找使异步可观察对象工作，我建议您快速查看这个问题RxJava Fetching Observables In Parallel。

Ben（RxJava 的作者）帮助澄清了我对该主题的疑虑。

希望对你有帮助

和