观察者模式，只听别人讲不太容易理解，要自己深入，慢慢学习它的精妙之处！

调用

public void test(){
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(Emitter<String> emitter) {
                emitter.onNext("ddddd");
            }
        }).subscribeObserver(new Observer<String>() {
            @Override
            public void onNext(String o) {
                Log.e("chy", o);
            }

            @Override
            public void onSubscribe() {
                Log.e("chy", "onSubscribe");
            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

ObservableOnSubscribe.java接口类

public interface ObservableOnSubscribe<T> {
    void subscribe(Emitter<T> emitter);
}

被观察者ObservableSource.java

public interface ObservableSource {
    void subscribeObserver(Observer observer);
}

被观察者实现类Observable.java

public abstract class Observable implements ObservableSource {
    @Override
    public void subscribeObserver(Observer observer) {
        subscribeActual(observer);
    }

    protected abstract void subscribeActual(Observer observer);


    public static<T> Observable create(ObservableOnSubscribe<T> source){
        return new ObservableCreate<T>(source);
    }
}

被观察者子类ObservableCreate.java

public class ObservableCreate<T> extends Observable {
    public ObservableOnSubscribe source;

    public ObservableCreate(ObservableOnSubscribe<T> observableOnSubscribe){
        this.source = observableOnSubscribe;
    }

    @Override
    protected void subscribeActual(Observer observer) {
        observer.onSubscribe();

        CreateEmitter createEmitter = new CreateEmitter(observer);
        source.subscribe(createEmitter);
    }

    static final class CreateEmitter<T> implements Emitter<T>{

        Observer<T> observer;

        public CreateEmitter(Observer<T> observer){
            this.observer = observer;

        }
        @Override
        public void onNext(T t) {
            observer.onNext(t);
        }

        @Override
        public void onError(Throwable throwable) {
            observer.onError(throwable);
        }

        @Override
        public void onComplete() {
            observer.onComplete();
        }
    }
}

观察者Observer.java

public interface Observer<T> {
    void onNext(T t);
    void onSubscribe();
    void onError(Throwable throwable);
    void onComplete();
}

发射者Emitter.java

public interface Emitter<T> {
    void onNext(T t);
    void onError(Throwable throwable);
    void onComplete();
}