3、rxjava基本原理研究

模拟rxjava基本工作原理

public abstract  class Observable {

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


    public final  void subscribe(Observer observer)
    {
        subscribeActual(observer);
    }

    protected abstract void subscribeActual(Observer observer);
}

/**
 * Observalbe的实现类
 */
public class ObservableCreate extends Observable {
    private  ObservableOnSubscribe observableOnSubscribe;
    public ObservableCreate(ObservableOnSubscribe observableOnSubscribe)
    {
        this.observableOnSubscribe=observableOnSubscribe;
    }

    @Override
    protected void subscribeActual(Observer observer) {

        observableOnSubscribe.subscribe(observer);

    }
}

public abstract class Observer {


    public abstract void onNext(String str);

}

public interface ObservableOnSubscribe {
    void subscribe(Observer observer);
}

public class Main {

    public  static  void main(String[] args)
    {
        Observable.create(new ObservableOnSubscribe(){
            @Override
            public void subscribe(Observer observer) {

                observer.onNext("hello world");
            }
        }).subscribe(new Observer(){
            @Override
            public void onNext(String str) {
                System.out.println(str);
            }
        });
    }
}

输出：

image.png