RXJava 造轮子

什么是rxjava?

rxjava 是一个响应式编程，为了更好更优雅的处理异步时间，内部实现原理是观察者模式

自己造轮子实现Rxjava

观察者模式核心有两个，一个观察者一个被观察者

先创建一个被观察者observer(基层抽象类)

//观察者的基层抽象类

public interface Observer {

void onNext(T value);

void onError(Throwable throwable);

void onComplete();

}

观察者内部实现了三个方法，完成，错误，下一步

创建一个observable(观察者)

//被观察者

public class Observable{

OnSubscrib onSubscrib;

//私有构造

private Observable(OnSubscribonSubscrib){

this.onSubscrib=onSubscrib;

}

//创造被观察者对象

public static Observable create(OnSubscribonSubscrib){

return new Observable(onSubscrib);

}

//开始观察，观察者和被观察者建议联系

public void subscrib(Subscriber subscriber){

subscriber.start();

//被观察者拿到观察者对象

onSubscrib.call(subscriber);

}

}

观察者的构造函数中传入了一个onSubscrib对象

onsubscrib实际就是观察者发生(发起通知)实现的对象

onsubscrib 中的call()方法,就是建立观察者和被观察者之间关系的桥梁

onSubscrib类如下

public interface OnSubscrib{

void call(Subscriber subscriber);

}

例子实现

总体流程是这样的:

1. 创建一个被观察者对象observable，传入了一个onsubscrib对象，用来设置通知发生的时机

2. 创建一个观察者subscriber,依次实现抽象方法

3. 通过被观察者observable调用subscrib(),传入了subscriber

- 其实是observable中的onsubscrib拿到了subscriber

- 拿到subscriber之后,根据出发条件通知观察者

4. subscriber被调用抽象方法，观察者被通知了

添加map操作符

添加map操作符原理其实是这样的:

观察者模式先创建了observable对象，传入了onsubscrib对象

调用map操作符，生成了一对新的观察者和被观察者的observable 和subscriber

生成的subscriber 直接观察上层的observable

这样通知流程就完成了

如下图：

map操作符

线程切换实现

1.Worker是实现代码的runnable,会在指定的线程中工作

例如，代码中是一个单线程工作

public class Worker {

Executor executor;

public Worker(Executor executor){

this.executor=executor;

}

//执行操作

public void schedule(Runnable runnable){

executor.execute(runnable);

}

}

Scheduler 提供了生成Worker的方法

定义一个Schedule作为scheduler的工厂类,内部提供了各种线程，以供在worker中使用

具体实现

定义一个subscribOn()指定被观察者的工作线程

public Observable subscribOn(final Scheduler scheduler){

return Observable.create(new OnSubscrib() {

@Override

public void call(final Subscriber subscriber) {

//将线程发送的操作用在具体的线程中

scheduler.createWorker().schedule(new Runnable() {

@Override

public void run() {

//订阅上层观察者

Observable.this.subscrib(subscriber);

}

});

}

});

}

内部实现和map操作符类似

生成一个新的observeble和subscriber

监听上层observable,并且将onsubscrib.call()方法出发通知放在设置的线程中发送通知

定义一个observeOn()指定观察者的线程

public Observable observeOn(final Scheduler scheduler){

//需要将subscriber中的方法放置在对应线程中处理

return Observable.create(new OnSubscrib() {

@Override

public void call(final Subscriber subscriber) {

final Worker worker=scheduler.createWorker();

Observable.this.subscrib(new Subscriber() {

@Override

public void onNext(final T value) {

//下一步执行在指定线程中

worker.schedule(new Runnable() {

@Override

public void run() {

subscriber.onNext(value);

}

});

}

@Override

public void onError(Throwable throwable) {

subscriber.onError(throwable);

}

@Override

public void onComplete() {

subscriber.onComplete();

}

});

}

});

}

原理和上面一样

生成一个新的observeble和subscriber

监听上层observable,生成新的subscriber

在subscriber中，将next()方法在指定的线程中运行

测试例子

Observable.create(new OnSubscrib() {

@Override

public void call(Subscriber subscriber) {

Log.e(TAGObservable, Thread.currentThread().getName());

subscriber.onNext("1");

}

})

.subscribOn(Schedulers.io())

.observeOn(Schedulers.io())

.subscrib(new Subscriber() {

@Override

public void onNext(String value) {

Log.e(TAGSubscriber, value);

Log.e(TAGSubscriber, Thread.currentThread().getName());

}

@Override

public void onError(Throwable throwable) {

}

@Override

public void onComplete() {

}

});

线程调度流程如下图：

线程调度.jpeg

rxJava简单的轮子实现就先告一段落了，下面开始各种操作符的使用体验了啊

github传送门