生產者消費者的例子，在生產者中有背壓支援

此示例中的 TestProducer 在給定範圍內生成 Integerobjects 並將它們推送到 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;
            }
        });
    }
}

此示例中的 Consumer 擴充套件了 DefaultSubscriber<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 緩衝區時才會生效。