RxJava3。为什么 FlowableSubscriber onNext 没有被调用？

Question

我需要获取订阅对象才能有机会取消订阅侦听器。为此，我想提供一个 FlowableSubscriber 来运行。

代码：

FlowableSubscriber fs = new FlowableSubscriber() {
        @Override
        public void onSubscribe(@NonNull Subscription s) {
            System.out.println("Flowable onSubs");
        }

        @Override
        public void onNext(Object o) {
            System.out.println("Flowable onNext");
        }

        @Override
        public void onError(Throwable t) {
            System.out.println("Flowable onErr");
        }

        @Override
        public void onComplete() {
            System.out.println("Flowable onComlet");
        }
    };

日志是：

Running...
Flowable onSubs

如果我使用 lambda，它可以工作，但没有 onSubscribe 回调。

我怎样才能获得订阅以及为什么这些方法没有被调用？

Answer 1

由于支持反压，您必须通过调用方法Flowable来实际控制可以消耗多少项目，以便可以通过以下方式发出它们：requestSubscriptionFlowable

FlowableSubscriber fs = new FlowableSubscriber() {

        private Subscription subscription;

        @Override
        public void onSubscribe(Subscription s) {
            System.out.println("Flowable onSubs");
            subscription=s;
            subscription.request(Integer.MAX_VALUE);
        }

        @Override
        public void onNext(Object o) {
           System.out.println("Flowable onNext");
           //subscription.request(1); you can also request more items from onNext method - it is up to you
        }

        @Override
        public void onError(Throwable t) {
            System.out.println("Flowable onErr");
        }

        @Override
        public void onComplete() {
            System.out.println("Flowable onComlet");
        }
};

在示例中Integer.MAX_VALUE请求订阅，但这可能不是最好的主意。问题是您应该调用Subscription::requestfromonSubscribe来请求初始项目，然后调用 fromonNext并决定您实际可以处理多少项目。