使用示例： Observable.create(new ObservableOnSubscribe<String>() { @Override public void subscribe(ObservableEmitter<String> emitter) throws Exception { //发射数据 emitter.onNext("你好"); } }).subscribe(new Observer<String>() { @Override public void onSubscribe(Disposable d) { } @Override public void onNext(String s) {
//接受数据 Log.i("======",s); } @Override public void onError(Throwable e) { } @Override public void onComplete() { } });
这个简单的使用：创建一个Observable 对象调用subscribe()方法去订阅一个Observer
先来分析Observable类
1.Observable源码
ObservableSource.java 是观察者的接口
从Observable实现了ObservableSource这个接口我们可以知道，它需要实现subscribe(Observer observer)这个方法。
可以看出需要在subscribe这个方法里面调用到subscribeActual这个抽象方法
总结下：我们的被观察者实现了ObservableSource这个接口，因此需要去实现subscribe这个方法，在这个方法里面调用了Observable的抽象方法subscribeActual
所以我们可以猜测到Observable的子类需要在subscribeActual这个方法里面去处理对应的逻辑。
下面看下Observable.create方法做了什么。
这个时候就会产生一个疑问？这个抽象方法是谁触发的呢？
在订阅方法的时候我们传入了一个观察者对象.subscribe(new Observer<String>())
看下这个订阅的方法：
总的来说就是observable.subscribe一个observable后就会触发observable里面的subscribeActual这个方法，在这个方法里面创建了一个发射器对象，当发射数据的时候onNext就会调用我们observer这个里面的onNext方法。
原文：https://www.cnblogs.com/lianzhen/p/12939898.html