将 Observable 发出的值传递给不同的对象，同时组合它

Question

我想在 3 个对象中传递每个发出的值，对于一个对象，我们基本上可以照常进行：

Observable.just(KV("1", "Alex"), KV("2", "LeBron"), KV("3", "Kobe"), KV("1", "Kevin"))
.flatMapCompletable(keyValue -> doSth(keyValue))

但对于其他人，我想 groupBy(KV::getKey) 然后收集（例如，连接值）。

Observable.just(KV("1", "Alex"), KV("2", "LeBron"), KV("3", "Kobe"), KV("1", "Kevin"))
.groupBy(KV::getKey)
.collect(Result::new, (acc, keyVal) -> acc.add(keyVal.getVal()))
.flatMapCompletable(result -> writeResultSomeWhere());

我怎样才能在 1 个链中做到这一点？

谢谢！

Answer 1

你可以publish主要来源：

Observable.just(KV("1", "Alex"), KV("2", "LeBron"), KV("3", "Kobe"), KV("1", "Kevin"))
.publish(o -> Completable.merge(
    o.flatMapCompletable(keyValue -> doSth(keyValue)),
    o.groupBy(KV::getKey)
       .collect(Result::new, (acc, keyVal) -> acc.add(keyVal.getVal()))
       .flatMapCompletable(result -> writeResultSomeWhere())
));