RxJava2 将两个 Flowable 压缩为一个

Question

我正在努力寻找将两个 Flowable 压缩为一个的任何 RxJava2 示例。

我正在尝试修改 this test 以包含类似于

的内容

    Integer[] ints = new Integer[count];
    Integer[] moreints = new Integer[count];
    Arrays.fill(ints, 777);
    Arrays.fill(moreints, 777);

    Flowable<Integer> source = Flowable.fromArray(ints);
    Flowable<Integer> anothersource = Flowable.fromArray(moreints);

    Flowable<Integer> zippedsources = Flowable.zip(source, anothersource,
            new BiFunction<Flowable<Integer>, Flowable<Integer>, Flowable<Integer>>() {

                @Override
                public void apply(Flowable<Integer> arg0, Flowable<Integer> arg1) throws Exception {
                    return arg0.blockingFirst() + arg1.blockingLast();
                }

    }).runOn(Schedulers.computation()).map(this).sequential();

编辑：我试图从源和另一个源中获取一个整数并将它们相加，但它似乎与 RxJava1 这样做的方式根本不同......我尝试了一堆返回 Integer、Publisher、Flowable 和 void 的变体但在 Eclipse 中，zip 运算符本身不断出现错误。

我无法弄清楚.zip(Iterable<? extends Publisher<? extends T>>, Function<? super Object[], ? extends R>). 中的内容是什么

Answer 1

由于你只需要压缩两个flowable，你可以使用Flowable.zipWith Operator。

使用方法如下：

source.zipWith(anotherSource, new BiFunction<Integer, Integer, Integer>() {
    @Override public Integer apply(Integer a, Integer b) {
        return a + b;
    } 
};