RxJava Flowable.Interval背压时flatmap with single

Question

我有一个场景,我需要定期调用API来检查结果.我正在Flowable.interval创建一个调用API的区间函数.

但是,我在背压方面遇到了麻烦.在下面的示例中,在区间中的每个刻度上创建一个新单.如果呼叫尚未进行,则所需的效果是仅调用API

Flowable.interval(1, 1, TimeUnit.SECONDS).flatMap {
        System.out.println("Delay $it")

        //simulates API call
        Single.just(1L).doAfterSuccess {
            System.out.println("NEW SINGLE!!!")
        }.delay(4, TimeUnit.SECONDS).doAfterSuccess {
            System.out.println("SINGLE SUCCESS!!!")
        }.toFlowable()
    }.subscribeOn(Schedulers.io()).observeOn(Schedulers.computation()).blockingFirst()

我可以使用像这样的过滤变量来解决这个问题:

var filter = true

Flowable.interval(1, 1, TimeUnit.SECONDS).filter {
    filter
}.flatMap {

    System.out.println("Delay $it")

    Single.just(1L).doOnSubscribe {
        filter = true
    }.doAfterSuccess {
        System.out.println("NEW SINGLE!!!")
    }.delay(4, TimeUnit.SECONDS).doAfterSuccess {
        System.out.println("SINGLE!!!")
        filter = true
    }.toFlowable()
}.subscribeOn(Schedulers.io()).observeOn(Schedulers.computation()).blockingFirst()

但它似乎是一个hacky解决方案.我已经厌倦了onBackPressureDrop在interval功能之后应用,但它没有效果.

有什么建议？

Answer 1

你也必须约束flatMap:

Flowable.interval(1, 1, TimeUnit.SECONDS)
.onBackpressureDrop()
.flatMapSingle({
    System.out.println("Delay $it")

    //simulates API call
    Single.just(1L).doAfterSuccess {
        System.out.println("NEW SINGLE!!!")
    }.delay(4, TimeUnit.SECONDS).doAfterSuccess {
        System.out.println("SINGLE SUCCESS!!!")
    }
}, false, 1)  // <----------------------------------------------------------
.subscribeOn(Schedulers.io())
.observeOn(Schedulers.computation())
.subscribe()