基本的，我们将有

Observable.subscribe(Observer);// => Observer observe Observable and Observable subscribe Observer

示例

requestInterface.callApi().subscribe(new Observer...); // requestInterface.callApi() <=> Observable

从http://reactivex.io/documentation/operators/subscribeon.html

• • 订阅开启**
• SubscribeOn运算符指定Observable将在哪个线程上开始操作，无论在运算符链的哪个点调用该运算符
• • ObserveOn**（影响2项）
• 指示Observable向指定Scheduler上的Observers发送通知。
• ObserveOn会影响Observable将在该运算符出现的下方使用的线程

示例

registerUserReturnedObserverble()  // run on worker thread because subscribeOn(Schedulers.io()) (line 5)
.andThen(loginReturnObserverble()) // run on worker thread because subscribeOn(Schedulers.io()) (line 5)
.observeOn(AndroidSchedulers.mainThread())
.andThen(getUserDataReturnObserverble()) // run on main thread because .observeOn(AndroidSchedulers.mainThread()) is above this operator (line 3)
.subscribeOn(Schedulers.io())
.subscribe(new Observer<Void>{
    // run on main thread because observeOn(AndroidSchedulers.mainThread()) 
});

• subscribeOn（Schedulers.io（））：这告诉Observable在后台线程上运行任务
• • • observeOn（AndroidSchedulers. mainThread（））**：这告诉Observer接收android UI线程上的数据，以便您可以采取任何UI相关的操作。

下面是一个示例：

getCardsObservable.subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new rx.Observer<List<Card>>() {
                @Override
                public void onCompleted() {
                }
                @Override
                public void onError(Throwable e) {
                    listener.onError(e.getMessage());
                }
                @Override
                public void onNext(List<Card> cards) {
                    listener.onSuccess(cards);
                }
            });

**subscribeOn -->**执行调用的线程，类似于调用asynctask
**observeOn -->**其中响应将在UI线程的进程中被观察
**subscribe -->**观察者回调

以下情况指定了使用observeOn（）和/或subscribeOn（）时可能出现的所有不同情况。

1. subscribeOn影响上游运算符（subscribeOn之上的运算符）
2. observeOn影响下游运算符（observeOn下面的运算符）
   1.如果你没有在RxJava中指定线程（如果你没有指定subscribeOn，observeOn或两者），数据将由当前的调度器/线程（通常是主线程）发出和处理。例如，下面链中的所有运算符都将由当前线程（Android的主线程）处理。

Observable
.just("big", "bigger", "biggest")    
.map(String::length)    
.filter { it > 6 }    
.subscribe { length -> println("item length $length") }

1.如果只指定了subscribeOn，则将在该线程上执行所有运算符

Observable
.just("big", "bigger", "biggest") 
.subscribeOn(Schedulers.io())    
.map(String::length)    
.filter { it > 6 }    
.subscribe { length -> println("item length $length") }

仅数据发射，Map和过滤操作符将在io调度器上执行，如上游操作符subscribeOn所指示的。
1.如果只指定了observeOn，则所有运算符都将在当前线程上执行，只有observeOn以下的运算符才会切换到observeOn指定的线程

Observable
.just("big", "bigger", "biggest")   
.map(String::length)
.observeOn(Schedulers.computation())     
.filter { it > 6 }    
.subscribe { length -> println("item length $length") }

数据发射，Map将在currentThread调度器上执行。
filter将在计算调度器上执行，如下游运算符observeOn所指示的。
1.如果同时指定了subscribeOn和observeOn，则observeOn以下的所有运算符将切换到observeOn指定的线程，其余observeOn以上的运算符将切换到subscribeOn指定的线程。无论指定subscribeOn和observeOn的顺序如何，这都是正确的

Observable
.just("big", "bigger", "biggest")   
.subscribeOn(Schedulers.io()) 
.map(String::length)
.observeOn(Schedulers.computation())     
.filter { it > 6 }    
.subscribe { length -> println("item length $length") }

数据发射和Map操作符将在io调度器上执行，如上游操作符subscribeOn所指示的。
filter将在计算调度器上执行，如下游运算符observeOn所指示的。
即使subscribeOn在observeOn之后被调用，线程的使用也是相同的。

Observable
.just("big", "bigger", "biggest")  
.map(String::length)
.observeOn(Schedulers.computation()) 
.filter { it > 6 }   
.subscribeOn(Schedulers.io()) 
.subscribe { length -> println("item length $length") }