有没有更好的方法可以使用RxJava获得两个观测值的交点

Question

使用RxJava，我有一个源Observable，它发出许多项目，我希望这些项目与另一个发出相同类型的Observable相交。在研究了许多选项之后，似乎构建事物的最连贯的方式是：

Observable<String> source = ...emits 20 items

Observable.create(subscriber -> {
    source
        .buffer(5)
        .subscribe(things -> {
            tocheck.getMatches(things) //emits 3 matches
                .subscribe(subscriber::onNext, subscriber::onError, () -> {});
        }, subscriber::onError, subscriber::onCompleted));

这里的预期输出是，当我订阅结果的可观察时，我会发出12个项目。由于getMatches的契约，我需要缓冲结果。

从表面上看，这似乎是可行的，但它似乎不是最干净的方式。过滤器似乎不适用于此，因为出于性能原因，我无法对每个项目运行相交检查。我尝试使用平面图，但getMatches可观察完成流，而不是来自源可观察的完成通知。

有没有更好的方法来构建这个？

编辑：要澄清这种风格的代码发生了什么，请执行以下操作：

Observable<String> source = ...emits 20 items

source
    .buffer(5)
    .flatMap(this::getMatches);  //final observable would emit a total of 12 items

这显然更简洁，但当我添加一些日志记录时（假设数据大小与原始片段相同）：

source
    .doOnEach(notification -> {
        log.trace("Processing {}", notification.getValue());
    })
    .buffer(5)
    .flatMap(this::getMatches)
    .doOnEach(notification -> {
        log.trace("Processing after match {}", notification.getValue());
    });

我得到了20个“Processing”日志实例，奇怪的是，“Processing after”日志中只有几行日志（我预计只有12行）。它似乎比它应该完成的时间更早。也许我的结构有问题？

Answer 1

所以看起来Androidx是正确的。我使用的是Redis莴苣反应API，它的行为看起来不太正常。上面添加的代码片段是构造两个可观察对象相交的正确方法。