深入RxJava之汇聚操作

[soapgu]

• 什么是汇聚

对于多项异步操作，我们怎么操作他全部完成以后？
相当于等价 await Task.WhenAll( Tasks[] tasks )

• Zip

Rx里面眼花缭乱的api总有一款适合你

• 怎么聚合可变集合的Rx对象那？

zip操作本身不复杂，但是实际项目中也不会是固定数量的组合操作，如果是可变集合的汇聚要怎么处理那

• 入参 Set

public interface IBootTask {
    Single<Boolean> run();
}

就是说我有一堆不定数量的启动任务集合

• 出参 Boolean
  就是所有的这些启动任务返回全部成功了没有
  如图所示
  

• 解决方案

zip就像电路世界的并联，正好和flatmap相反。完美把握所有并发的Rx，并在发送完所有数据后收集所有的Rx的result，合成一个新流。
这是一个非常漂亮的解耦模式，对内控制子Rx流程。对外部来看，封装成了一个新的Rx，无需关心子流的细节

 Stream<Single<Boolean>> stream = bootTasks.get().stream().map(IBootTask::run);
                                    return Single.zip( stream.collect(Collectors.toList()) ,
                                            args-> Arrays.stream(args).allMatch( r->(boolean)r )  );

是不是超级简洁

• 解析

从zip的两个参数入手

1. stream，JAVA中的Linq
   zip的第一个参数为Iterable<@nonnull ? extends SingleSource<? extends T>> sources
   就是Rx对象的集合枚举。要怎么转换成匹配的对象那，通过stream的api来转换，这东西基本对标Linq。
   比原版C#的Linq稍微差一点但是也凑合用
   map -> Select 投影
   collect -> ToList() 转换集合

2. Function<? super Object[], ? extends R> zipper)
   看到Object一开始有点懵逼，既然JAVA都已经到了泛型时代（吐槽下虽然还是伪泛型），为啥还有这种粗糙对象那。后来想到是为了兼容性考虑，可能第一个sources对象里面的类型不是完全的一致。可能需要处理不同类型返回的情况，或者根本就不关心类型。

3. zipArray的困惑
   还有疑惑的地方就是一开始我选择的api是zipArray，但是没有成功，还是用的zip完成了。反过头来还是用zipArray，需要写成这个样子，一定要把第二个参数转成Array才行。非常的别扭还会有类型警告。
   总结：zipArray完全没必要使用，zip完全替代了相关功能。wiki里面也没强调

Stream<Single<Boolean>> stream = bootTasks.get().stream().map(IBootTask::run);
                                    Single<Boolean>[] list = stream.toArray(Single[]::new);
                                    return Single.zipArray( args-> Arrays.stream(args).allMatch( r->(boolean)r ) , list );

• 参考

github wiki
java doc