RxJava:Observable和默认线程

Question

我有以下代码:

Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull final ObservableEmitter<String> s) throws Exception {
                Thread thread = new Thread(new Runnable() {
                    @Override
                    public void run() {
                        s.onNext("1");
                        s.onComplete();
                    }
                });
                thread.setName("background-thread-1");
                thread.start();
            }
        }).map(new Function<String, String>() {
            @Override
            public String apply(@NonNull String s) throws Exception {
                String threadName = Thread.currentThread().getName();
                logger.logDebug("map: thread=" + threadName);
                return "map-" + s;
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {}

            @Override
            public void onNext(String s) {
                String threadName = Thread.currentThread().getName();
                logger.logDebug("onNext: thread=" + threadName + ", value=" + s);
            }

            @Override
            public void onError(Throwable e) {}

            @Override
            public void onComplete() {
                String threadName = Thread.currentThread().getName();
                logger.logDebug("onComplete: thread=" + threadName);
            }
        });

这是输出:

map: thread=background-thread-1 
onNext: thread=background-thread-1, value=map-1 
onComplete: thread=background-thread-1

重要细节:我subscribe从另一个线程(mainAndroid中的线程)调用该方法.

所以看起来Observable类是同步的并且默认情况下它会map在发出事件(s.onNext)的同一个线程上执行所有操作(如+通知订阅者的操作符),对吗？我想知道......它是故意的行为还是我误解了什么？实际上我期待至少onNext和onComplete回调将在调用者的线程上调用,而不是在发出事件的那个上调用.我是否正确理解在这种特殊情况下,实际来电者的线程无关紧要？至少在异步生成事件时.

另一个问题 - 如果我从一些外部源接收一些Observable作为参数(即我不自己生成它)...如果我的用户无法检查它是同步还是异步,我只需要明确指定我想通过subscribeOn和observeOn方法接收回调的位置,对吧？

谢谢!

Answer 1

RxJava对并发性没有任何意见.如果您不使用observeOn/subscribeOn之类的任何其他机制,它将在订阅线程上生成值.请不要在运算符中使用Thread之类的低级构造,否则可能会违反合同.

由于使用了Thread,onNext将从调用Thread('background-thread-1')调用.订阅发生在调用(UI-Thread)上.链中的每个运算符都将从'background-thread-1'-calling-Thread调用.订阅onNext也将从'background-thread-1'调用.

如果要生成不在调用线程上的值,请使用:subscribeOn.如果要将线程切换回主要,请在链中的某处使用observeOn.最有可能在订阅它之前.

例:

Observable.just(1,2,3) // creation of observable happens on Computational-Threads
            .subscribeOn(Schedulers.computation()) // subscribeOn happens only once in chain. Nearest to source wins
            .map(integer -> integer) // map happens on Computational-Threads
            .observeOn(AndroidSchedulers.mainThread()) // Will switch every onNext to Main-Thread
            .subscribe(integer -> {
                // called from mainThread
            });

这是一个很好的解释. http://tomstechnicalblog.blogspot.de/2016/02/rxjava-understanding-observeon-and.html