groovy - 如何在 RxJava 中延迟地从集合中发出项目？

Question

我有以下代码从集合中发出项目。

// fills the given coll with some items and
// returns the observable emitting the filled collection
// does ASYNC work
Observable<Collection> generate(Collection coll){ ... }

def coll = []
generate(coll).flatMap({ filledColl ->
    rx.Observable.from(filledColl)
}).subscribe({ ... })

问题是这个集合可以包含数千个项目，并且由于 generate 异步工作，因此此代码会导致 subscribe 方法几乎立即被调用数千次(这对于我的工作来说是不希望的)正在做内部观察者)。

如何修改此代码以延迟从集合中发出项目？ 例如:发出 100 个项目，然后等待 100 毫秒，然后发出另外 100 个项目，或者等待 10 毫秒，然后发出下一个项目？

Answer 1

在 flatMap 内部，您需要将 filledColl 拆分为更小的部分，延迟每个部分，然后将它们全部合并为一个可观察对象，您将在 FlatMap 内部返回该可观察对象。

generate(coll).flatMap({ filledColl ->
    def chunkSize = 100
    resultStream = rx.Observable.never()
    for (i in 0 ..< filledCol.size()/chunkSize) {
        def chunk = filledCol[i*chunkSize .. (i+1)*chunkSize]
        resultStream = resultStream.mergeWith(
            rx.Observable.from(chunk).delay(100*i, TimeUnit.MILLISECONDS)
        )
    }
    resultStream
}).subscribe({ ... })

这只是粗略的想法，您可能仍然想根据您的需要进行测试、调整和纠正。另外，将其移至生成函数中可能更有意义，但这取决于您，因为我不知道生成()中的内容。