java-8 - Observable 发射器对 HystrixCommands 产生太大的压力

Question

我有一个发出文件行的 Observable(从 GCS 读取的许多 GB)。

return StringObservable.byLine(
    Observable.using(
        () -> storage.get(blobId).reader(),
            reader -> Observable.create(
                    new OnSubscribeReadChannel(reader, 64 * 1024)
                ),
            ReadChannel::close
    )
)

每一行都会导致对各种 DB 的多次(在某些情况下很多)调用，所有这些调用都包含在 Hystrix 命令中。显然，线路最终压倒了 Hystrix 命令，线路开始开放，每个人都度过了糟糕的一天。

这大致是我在做什么:

readLinesFromCloudStorageFile.readLines(blobInfo.getBlobId()))
            .map(this::deserializeLine)
            .flatMap(this::addDataToObjectFromSomeDb)
            .flatMap(this::writeObj)
            .map(Set::size)
            .reduce(0, (a, b) -> a + b)
            .toBlocking().single()

有没有办法可以施加一些背压，或者限制一次处理的行数或其他什么？

Answer 1

您需要使用 Emitter.BackpressureMode.BUFFER

BUFFER
Buffers (unbounded) all onNext calls until the downstream can consume them.

http://reactivex.io/RxJava/1.x/javadoc/index.html?rx/Emitter.BackpressureMode.html