使用RxJava2正确处理背压和并发

Question

我有一个注册了回拨的服务，现在我想将其公开为可流动的，具有某些要求/限制：

1. 接收回调的线程不应该被阻塞（工作应该交给观察者指定的不同线程/调度程序）
2. 不应该有任何异常抛出由于消费者是慢下来流
3. 多个消费者可以相互独立订阅
4. 消费者可以选择缓冲所有的物品，这样它们就不会丢失，但是它们不应该在生产者类中被缓冲

以下是我目前的情况

class MyBroadcaster {
    private PublishProcessor<Packet> packets = PublishProcessor.create();
    private Flowable<Packet> backpressuredPackets = packets.onBackpressureLatest();

    public MyBroadcaster() {
       //this is actually different to my exact use but same conceptually
       registerCallback(packets::onNext);
    }

    public Flowable<Packet> observeAllPacketsOn(Scheduler scheduler) {
        return backpressuredPackets.observeOn(scheduler);
    }
}

我不确定这是否符合我的要求。在onBackpressureLatestjavadoc上有一条关于observeOn的注释，我不明白：

请注意，由于背压请求是如何通过subscribeOn/observeOn传播的，因此从下游请求超过1个并不保证onNext事件的连续传递

我还有其他问题：

• onBackpressureLatest调用是否使项目不再进行多播

Answer 1

我不确定这是否符合我的要求。

事实并非如此。分别应用onBackpressureLatest或onBackpressureBuffer和observealpacketson中的observeOn。

OnBackPressureRelatest调用是否使项目不再进行多播？

多播是由PublishProcencer完成的，不同的订阅者将独立建立一个通道，其中onBackpressureXXX和观察运算符在单个订阅者的基础上生效。

如何测试我的需求？

使用TestSubscriber（Flowable.test（））通过有损或无损Flowable订阅，将一组已知的数据包馈送到数据包中，并查看它们是否都通过TestSubscriber到达。assertValueCount（）或TestSubscriber。values（）。有损耗的应该是1。。N，在一个宽限期之后，无损的应该有N个值。

奖励：如果我有多个这样的发布者（在同一个类或其他地方），那么让同一模式可重用的最佳方法是什么。使用委派/额外方法创建我自己的可流动性？

您可以将观察AllPacketsOn转换为FlowableTransov，而不是MyBroadcaster上的方法调用，而是使用comush，例如：

class MyTransformers {
    public static FlowableTransformer<T, T> lossyObserveOn(Scheduler s) {
        return f -> f.onBackpressureLatest().observeOn(s);
    }
}

new MyBroadcaster().getPacketFlow()
.compose(MyTransformers.lossyObserveOn(scheduler))
.subscribe(/* ... */);