创建一个以有限速率发出项的 Flowable，以避免需要缓冲事件

Question

我有一个数据访问对象，它将数据源中的每个项目传递给使用者：

public interface Dao<T> {
void forEachItem(Consumer<T> item);
}

这总是以单线程方式生成项目 - 我目前无法更改此接口。

我想从这个界面创建一个Flowable：

private static Flowable<String> flowable(final Dao dao) {
return Flowable.create(emitter -> {
dao.forEachItem(item ->
emitter.onNext(item));
emitter.onComplete();
}, ERROR);
}

如果我在处理时间长于项目发射速率的情况下使用此Flowable，那么当我使用ERROR模式时，我可以理解地得到一个缺失的背压异常：

Dao<String> exampleDao =
itemConsumer ->
IntStream.range(0, 1_000).forEach(i ->
itemConsumer.accept(String.valueOf(i)));
flowable(exampleDao)
.map(v -> {
Thread.sleep(100);
return "id:" + v;
})
.blockingSubscribe(System.out::println);

我不希望缓冲项目 - 似乎这可能会导致非常大的数据集上的内存耗尽 - 如果操作明显比生产者慢。

我希望有一种背压模式，当发射器检测到背压时，允许发射器在通过下一个/完成事件时阻塞，但事实似乎并非如此？

就我而言，因为我知道 dao 以单线程方式生成项目，我认为我可以做这样的事情：

dao.forEachItem(item -> {
while (emitter.requested() == 0) {
waitABit();
}         
emitter.onNext(item)
});

但这似乎永远悬而未决。

我的方法有多错误？ ：-)鉴于我(相对限制性的)情况，有没有一种方法可以以尊重下游背压的方式生产物品？

我知道我可以通过一个单独的进程写入队列来做到这一点，然后根据该队列中的消耗编写一个 Flowable - 这是首选方法吗？

Answer 1

检查 Flowable 的部分，尤其是带有Supscription.request(long)的部分。我希望这能让你走上正确的道路。

---

此示例中的TestProducer在给定范围内生成Integer对象，并将它们推送到其Subscriber。它扩展了Flowable<Integer>类。对于新订阅服务器，它会创建一个Subscription对象，其request(long)方法用于创建和发布 Integer 值。

对于传递给subscriber的Subscription，可以从此onNext()调用中递归调用在订阅服务器上调用onNext()的request()方法非常重要。为了防止堆栈溢出，所示的实现使用outStandingRequests计数器和isProducing标志。

class TestProducer extends Flowable<Integer> {
static final Logger logger = LoggerFactory.getLogger(TestProducer.class);
final int from, to;
public TestProducer(int from, int to) {
this.from = from;
this.to = to;
}
@Override
protected void subscribeActual(Subscriber<? super Integer> subscriber) {
subscriber.onSubscribe(new Subscription() {
/** the next value. */
public int next = from;
/** cancellation flag. */
private volatile boolean cancelled = false;
private volatile boolean isProducing = false;
private AtomicLong outStandingRequests = new AtomicLong(0);
@Override
public void request(long n) {
if (!cancelled) {
outStandingRequests.addAndGet(n);
// check if already fulfilling request to prevent call  between request() an subscriber .onNext()
if (isProducing) {
return;
}
// start producing
isProducing = true;
while (outStandingRequests.get() > 0) {
if (next > to) {
logger.info("producer finished");
subscriber.onComplete();
break;
}
subscriber.onNext(next++);
outStandingRequests.decrementAndGet();
}
isProducing = false;
}
}
@Override
public void cancel() {
cancelled = true;
}
});
}
}

此示例中的使用者扩展DefaultSubscriber<Integer>，并且在启动和使用整数请求后请求下一个整数。在使用 Integer 值时，会有一点延迟，因此将为生产者建立背压。

class TestConsumer extends DefaultSubscriber<Integer> {
private static final Logger logger = LoggerFactory.getLogger(TestConsumer.class);
@Override
protected void onStart() {
request(1);
}
@Override
public void onNext(Integer i) {
logger.info("consuming {}", i);
if (0 == (i % 5)) {
try {
Thread.sleep(500);
} catch (InterruptedException ignored) {
// can be ignored, just used for pausing
}
}
request(1);
}
@Override
public void onError(Throwable throwable) {
logger.error("error received", throwable);
}
@Override
public void onComplete() {
logger.info("consumer finished");
}
}

在测试类的以下主要方法中，创建生产者和使用者并连接起来：

public static void main(String[] args) {
try {
final TestProducer testProducer = new TestProducer(1, 1_000);
final TestConsumer testConsumer = new TestConsumer();
testProducer
.subscribeOn(Schedulers.computation())
.observeOn(Schedulers.single())
.blockingSubscribe(testConsumer);
} catch (Throwable t) {
t.printStackTrace();
}
}

运行示例时，日志文件显示使用者连续运行，而生产者仅在需要重新填充 rxjava2 的内部 Flowable 缓冲区时处于活动状态。