RxJava:如何.zip两个Observable,然后.merge和最终.reduce汇总所有结果

Question

我有以下代码:

public void foo() {
    Long[] gData = new Long[] { 1L, 2L };

    rx.Observable.from(gData)
    .concatMap(data -> {

        rx.Observable<GmObject> depositObs1 = depositToUserBalance(data, 1);
        rx.Observable<GmObject> depositObs2 = depositToUserBalance(data, 2);


        return rx.Observable.zip(depositObs1, depositObs2, (depositObj1, depositObj2) -> {

            depositObj1.putNumber("seat_index", data);
            depositObj2.putNumber("seat_index", data);

            return rx.Observable.merge(
                    rx.Observable.just(depositObj1),
                    rx.Observable.just(depositObj2));
        })
    })
    .reduce(new ArrayList<Long>(), (payoutArr, payoutObj) -> {

        int seatIndex = ((GmObject) payoutObj).getNumber("seat_index").intValue();
        long payout = ((GmObject) payoutObj).getNumber("payout").longValue();
        payoutArr.add(seatIndex, payout);
        return payoutArr;
    })
    .subscribe(results -> {
        System.out.println(results);
    });
}

此代码使用.zip发送到observable,然后添加'seat_index'属性并调用.merge以使用.reduce,因此最终所有结果将聚合到ArrayList中.

这段代码有一个问题:当.reduce处理它的输入时,它将它作为Observable而不是GmObject ...什么函数可以从它的Observable包中"提取"GmObject？

以这种方式使用rxJava是否有意义？还是有更好的技术？

谢谢!

Answer 1

该zip运营商采取的第三个参数一个lambda.这个lambda是一个2 args函数,它返回一个args组成的对象.而不是Observable组合的结果(但是,当然,对象可以是一个Observable,但它不是你想要的).

所以在你zip打电话之后,你会有一个,Observable<Observable<GmObject>>但你期待一个Observable<GmObject>.

我不认为zip运营商是您正在寻找的运营商.

public void foo() {
    Long[] gData = new Long[] { 1L, 2L };

    rx.Observable.from(gData)
    .concatMap(data -> {

        rx.Observable<GmObject> depositObs1 = depositToUserBalance(data, 1).doOnNext(obj -> obj.putNumber("seat_index", data));
        rx.Observable<GmObject> depositObs2 = depositToUserBalance(data, 2).doOnNext(obj -> obj.putNumber("seat_index", data));


        return rx.Observable.merge(depositObs1, depositObs2);
    })
   .reduce(new ArrayList<Long>(), (payoutArr, payoutObj) -> {

        int seatIndex = ((GmObject) payoutObj).getNumber("seat_index").intValue();
        long payout = ((GmObject) payoutObj).getNumber("payout").longValue();
        payoutArr.add(seatIndex, payout);
        return payoutArr;
   })
   .subscribe(results -> System.out.println(results));
}