rxjava2 Backpressure

gistfile1.txt

package com.coffee.rx.rxjava2;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.subscribers.ResourceSubscriber;

import java.io.IOException;

/**
 * Created by zhaocongying on 17/4/7.
 */
public class BackpressureTest {
    public static void main(String[] args) {

        question();

        System.out.println("start...");
        try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 求问一下，为什么rxjava2上述代码中的Subscriber的onNext只能收到128次调用；
    // 但是当去掉observeOn()这行代码就能收到全部的数据了？
    private static void question() {
        Flowable.<String>create(emitter -> {
            for (int i = 0; ; i++) {
                System.out.println("\nsend: " + i);
                emitter.onNext(String.valueOf(i));
                try {
                    Thread.sleep(100);
                } catch (Throwable e) {
                    e.printStackTrace();
                }
            }

        }, BackpressureStrategy.BUFFER)
                .map(s -> {
                    System.out.println("map: " + s);
                    return s;
                })
                .subscribeOn(Schedulers.io())
                .observeOn(Schedulers.newThread())
                .subscribe(new ResourceSubscriber<String>() {

                    @Override
                    public void onNext(String s) {
                        System.out.println("ResourceSubscriber.onNext: " + s);
                    }

                    @Override
                    public void onError(Throwable t) {
                        t.printStackTrace();
                    }

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

                    }
                });
    }

}