        Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> observableEmitter) throws Exception {
                observableEmitter.onNext("emit A");
                observableEmitter.onNext("emit B");
                observableEmitter.onNext("emit C");
                observableEmitter.onComplete();
            }
        });

        Observer<String> observer = new Observer<String>() {
            @Override
            public void onSubscribe(Disposable disposable) {
                System.out.println("onSubscribe");
            }

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

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

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

        observable.subscribe(observer);

执行结果：

????????onSubscribe? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?
????????onNext? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?
????????emit A? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?
????????onNext? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?
????????emit B????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????
????????onNext? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?
????????emit C????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????
????????onComplete? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?

---->??observer 的 onSubscribe（Disposable disposable）方法在执行订阅动作后立即执行回调，在这里可以用??disposable 取消订阅或者将该对象保存起来以后再取消订阅。

---->??oberver 的 onNext（）在emitter.onNext（）执行后回调执行，表示响应被观察者事件，onNext（）方法会执行多次，表示有多个事件，具体执行次数由被观察者决定。

---->??observer 的 onComplete（）在被观察者的 onComplete（）执行后回调执行，表示事件全部完成，通常在这里做一些收尾工作。

---->??observer 的 onError（）在被观察者执行过程中的任何位置出现异常时回调执行，表示事件执行异常。同 onComplete（）互斥。

? ? ? ? 通过以上说明可以发现：在 observer 中，除了 onSubscribe（Disposable d）以外，其他方法都和创建?observable 时发射器?observableEmitter 的方法同名，且都在同名方法执行后回调。比如，observableEmitter 执行一次 onNext（），则 observer 中的 onNext（）方法执行一次。

????????onComplete（）和onError（）互斥，两者智能触发其一（且其中一个触发后，onNext（）变不会再执行）：

a.?onComplete（）触发后，后续 observableEmitter 调用任何方法都不在生效；

b. onError（）触发后，如果 observableEmitter 再调用?onComplete（）或 onError（），RxJava会抛出异常，开发者需要自行保证唯一性。

? ? ? ? 特别需要注意的是：ObservableOnSubscribe 的?subscribe（）方法在每次有新的observer加入时，都会在 observer 的 onSubscribe（）回调后触发，这就保证了所有的观察者都可以接收到事件。

2.3 关于subscribe（）

????????即订阅动作，他将观察者和被观察者建立了订阅关系，通过该方法，observable 回调 observer 的对应方法，从而达到观察者相应被观察者发出的事件（实际上被观察者只负责产生事件，他是事件源，真正发送事件的是他在订阅时，即subscribe（）被调用时）。

? ? ? ? 有多个重载的方法：

    // 1.观察者对被观察者发送的任何事件都响应
    @SchedulerSupport("none")
    public final void subscribe(Observer<? super T> observer) {
        ObjectHelper.requireNonNull(observer, "observer is null");

        try {
            observer = RxJavaPlugins.onSubscribe(this, observer);
            ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");
            this.subscribeActual(observer);
        } catch (NullPointerException var4) {
            throw var4;
        } catch (Throwable var5) {
            Exceptions.throwIfFatal(var5);
            RxJavaPlugins.onError(var5);
            NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
            npe.initCause(var5);
            throw npe;
        }
    }

    // 2.观察者对被观察者发出的事件作出响应（被观察者还可以继续发送事件）
    @SchedulerSupport("none")
    public final Disposable subscribe() {
        return this.subscribe(Functions.emptyConsumer(), Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION, Functions.emptyConsumer());
    }

    // 3.观察者只对被观察者发出的onNext（）事件作出响应    
    @SchedulerSupport("none")
    public final Disposable subscribe(Consumer<? super T> onNext) {
        return this.subscribe(onNext, Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION, Functions.emptyConsumer());
    }
    
    // 4.观察者只对被观察者发出的onNext（）、onError（）事件作出响应  
    @SchedulerSupport("none")
    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {
        return this.subscribe(onNext, onError, Functions.EMPTY_ACTION, Functions.emptyConsumer());
    }

    // 5.观察者只对被观察者发出的onNext（）、onError（）、onComplete（）事件作出响应  
    @SchedulerSupport("none")
    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {
        return this.subscribe(onNext, onError, onComplete, Functions.emptyConsumer());
    }

    // 6.观察者只对被观察者发出的onNext（）、onError（）、onComplete（）、onSubscribe（）事件作出响应
    @SchedulerSupport("none")
    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {
        ObjectHelper.requireNonNull(onNext, "onNext is null");
        ObjectHelper.requireNonNull(onError, "onError is null");
        ObjectHelper.requireNonNull(onComplete, "onComplete is null");
        ObjectHelper.requireNonNull(onSubscribe, "onSubscribe is null");
        LambdaObserver<T> ls = new LambdaObserver(onNext, onError, onComplete, onSubscribe);
        this.subscribe((Observer)ls);
        return ls;
    }

? ? ? ? 其中，2、3、4、5都this到6，而6最后又调用了1。另外，上面增加了两个新类：Consumer 和 Action?：observer 中的 onNext（）、onComplete（）、onError（）被 Consumer.accept（）代替；observer中的 onComplete（）方法被 Action.run（）方法代替。因此，如果只关心observer中的一个或几个回调方法，则可以通过使用?Consumer 或?Action 来替换observer注册到被观察者中。比如：上面的示例代码，如果只关心onNext（），则可以写为：

        Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> observableEmitter) throws Exception {
                observableEmitter.onNext("emit A");
                observableEmitter.onNext("emit B");
                observableEmitter.onNext("emit C");
                observableEmitter.onComplete();
            }
        });

        Consumer<String> consumer = new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                System.out.println(s);
            }
        };

        observable.subscribe(consumer);