使用 RxJava2 创建一个带有 generate 函数的 flowable答案

【问题标题】：Create a flowable with generate function using RxJava2使用 RxJava2 创建一个带有 generate 函数的 flowable
【发布时间】：2017-11-09 22:50:10
【问题描述】：

我需要创建一个自定义 Flowable 并实现背压。我正在尝试实现某种分页。这意味着当下游请求 5 个项目时，我将“询问数据源”以获取项目 0 - 5。然后当下游需要另外 5 个项目时，我将获取项目 5 - 10 并发回。

到目前为止我发现的最好的方法是使用Flowable.generate 方法，但我真的不明白为什么没有办法（据我所知）如何获取下游的requested 项目数正在请求。我可以使用生成器的state 属性来保存最后请求的项目的索引，所以我只需要新请求的项目的数量。我在 BiFunction apply 中得到的 emmiter 实例是 GeneratorSubscription，它是从 AtomicLong 扩展而来的。因此，将 emmiter 转换为 AtomicLong 可以获得我请求的号码。但我知道这不是“推荐”的方式。

另一方面，当您使用 Flowable.create 时，您会得到具有 long requested() 方法的 FlowableEmitter。使用generate 更适合我的用例，但现在我也很好奇使用Flowable.generate 的“正确”方式是什么。

也许我想太多了，所以请指出我正确的方向。 :) 谢谢。

这是实际代码的样子（在 Kotlin 中）：

Flowable.generate(Callable { 0 }, BiFunction { start /*state*/, emitter ->
        val requested = (emitter as AtomicLong).get().toInt()  //this is bull*hit
        val end = start + requested
        //get items [start to end] -> items
        emmiter.onNext(items)
        end /*return the new state*/
    })

【问题讨论】：

标签： java kotlin observable rx-java2 backpressure

【解决方案1】：

好的，我发现 BiFunction 的 apply 函数被调用了很多次，请求量 (n) 也是如此。因此，没有理由为此使用吸气剂。这不是我所希望的，但显然generate 是如何工作的。 :)

【讨论】：