RxJava - 缓存 Observable 更新并发出最大值答案

【问题标题】：RxJava - Cache Observable Updates and Emit Largest ValuesRxJava - 缓存 Observable 更新并发出最大值
【发布时间】：2018-09-09 18:29:55
【问题描述】：

我目前有一个 Observable<ProductIDUpdate> 发出一个表示产品 ID 更新的对象。更新可以是 ID 是新的ADDITION，也可以是已过期并需要DELETION。

public class ProductIDUpdate {

    enum UpdateType {
        ADDITION, DELETEION;
    }

    private int id;
    private UpdateType type;

    public ProductIDUpdate(int id) {
        this(id, UpdateType.ADDITION);
    }

    public ProductIDUpdate(int id, UpdateType type) {
        this.id = id;
        this.type = type;
    }
}

我想跟踪具有最大 ID 值的更新，因此我想修改流以便发出当前最高 ID。我将如何缓存流中的更新项，以便如果当前最高 ID 被删除，则发出下一个最高可用 ID？

【问题讨论】：

哪个流，哪个缓存？这些部分对我来说不够清楚。
没有缓存，流是我收到的更新的 Observable。我不确定如何处理数据以仅发出我需要的值 - 我假设预期行为需要缓存。
那你怎么称呼流？它是字面上的 Java Stream 还是您的可观察对象？很抱歉仍在尝试解决问题。
好吧，抱歉，该流是一个 RxJava Observable 流——即它发出我从流式源接收的 ProductIdUpdate 对象。

标签： java caching java-8 stream rx-java

【解决方案1】：

我对 Rx 一无所知，但这是我的理解：

您有一堆产品 ID。我不清楚您是否会随着时间的推移收到它们作为向您的班级发送的某些消息的一部分，或者您是否从一开始就知道所有 ID
您想在您的产品 ID 源之上创建一个流，以在任何时间点发出最高可用 ID

如果我的理解是正确的，那么使用PriorityQueue 怎么样？您使用反向比较器将 id 缓存在队列中（默认情况下，它将最小的元素保留在堆的顶部），当您想要发出一个新值时，您只需弹出顶部的值。

【讨论】：

【解决方案2】：

这样的东西能满足你的要求吗？

public static void main(String[] args) {
    Observable<ProductIDUpdate> products =
            Observable.just(new ProductIDUpdate(1, ADDITION),
                            new ProductIDUpdate(4, ADDITION),
                            new ProductIDUpdate(2, ADDITION),
                            new ProductIDUpdate(5, ADDITION),
                            new ProductIDUpdate(1, DELETION),
                            new ProductIDUpdate(5, DELETION),
                            new ProductIDUpdate(3, ADDITION),
                            new ProductIDUpdate(6, ADDITION));

    products.distinctUntilChanged((prev, current) -> prev.getId() > current.getId())
            .filter(p -> p.getType().equals(ADDITION))
            .subscribe(System.out::println,
                       Throwable::printStackTrace);

    Observable.timer(1, MINUTES) // just for blocking the main thread
              .toBlocking()
              .subscribe();
}

打印出来：

ProductIDUpdate{id=1, type=ADDITION}
ProductIDUpdate{id=4, type=ADDITION}
ProductIDUpdate{id=5, type=ADDITION}
ProductIDUpdate{id=6, type=ADDITION}

如果您删除 filter()，则会打印：

ProductIDUpdate{id=1, type=ADDITION}
ProductIDUpdate{id=4, type=ADDITION}
ProductIDUpdate{id=5, type=ADDITION}
ProductIDUpdate{id=5, type=DELETION}
ProductIDUpdate{id=6, type=ADDITION}

【讨论】：