建站
咨询
售后
建站咨询
一.Rxjava是什么
Rxjava在GitHub 主页上的自我介绍是 "a library for composing asynchronous and event-based programs using observable sequences for the Java VM"（一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库）。
成都创新互联服务项目包括滨城网站建设、滨城网站制作、滨城网页制作以及滨城网络营销策划等。多年来，我们专注于互联网行业，利用自身积累的技术优势、行业经验、深度合作伙伴关系等，向广大中小型企业、政府机构等提供互联网行业的解决方案，滨城网站推广取得了明显的社会效益与经济效益。目前，我们服务的客户以成都为中心已经辐射到滨城省份的部分城市，未来相信会继续扩大服务区域并继续获得客户的支持与信任！
通俗来说，Rxjava是一个采用了观察者模式设计处理异步的框架。链式调用设计让代码优雅易读。
举个例子：
Observable observable = Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(ObservableEmitter e) throws Exception {
e.onNext("a");
}
});
observable.subscribe(new Observer() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
}); 这是Rxjava2最简单的用法：
1.创建一个Observable，重写subscribe方法，这里主要处理被观察的事件。
2.订阅这个Observable，事件会回调observer的方法，我们可以对事件做响应的处理
二.Rxjava源码解析
2.1. 创建Observable：
创建Observable用的是Observable.create(ObservableOnSubscribe
public interface ObservableOnSubscribe {
/**
* Called for each Observer that subscribes.
* @param e the safe emitter instance, never null
* @throws Exception on error
*/
void subscribe(@NonNull ObservableEmitter e) throws Exception;
} ObservableOnSubscribe是一个接口，唯一的方法是subscribe，参数是ObservableEmitter
public interface Emitter {
/**
* Signal a normal value.
* @param value the value to signal, not null
*/
void onNext(@NonNull T value);
/**
* Signal a Throwable exception.
* @param error the Throwable to signal, not null
*/
void onError(@NonNull Throwable error);
/**
* Signal a completion.
*/
void onComplete();
} ObservableEmitter对接口Emitter进行扩展，增加了setDisposable、setCancellable等方法
基本参数了解了，现在看看create方法里面做了什么，代码如下：
public static Observable create(ObservableOnSubscribe source) {
return RxJavaPlugins.onAssembly(new ObservableCreate(source));
}
调用了RxJavaPlugins的onAssembly方法。又有一个新参数ObservableCreate(source)，我们看看它是什么：
final class ObservableCreate extends Observable {
public ObservableCreate(ObservableOnSubscribe source) {
this.source = source;
}
} 继承了Observable，所以也是个被观察对象，在构造函数中我们看到我们new的ObservableOnSubscribe对象，被存在了ObservableCreate的source里面
那我们继续看看onAssembly方法做什么：
public static Observable onAssembly(@NonNull Observable source) {
Function f = onObservableAssembly;
if (f != null) {
return apply(f, source);
}
return source;
} 一个Hook方法。onObservableAssembly是一个静态变量，我们没有设置，默认为空，所以直接返回source对象。也就是说，Observable的create方法其实就是把我们ObservableOnSubscribe对象，存储在ObservableCreate对象的source里面，然后返回ObservableCreate对象。
我们知道ObservableCreate是继承Observable的，所以创建了ObservableCreate对象，我们的Observable也就创建完了。
2.2 订阅事件（被观察者）
订阅被观察者的操作是observable.subscribe(new Observer
我们看看subscribe方法的源码：
public final void subscribe(Observer observer) {
try {
subscribeActual(observer);
} catch (NullPointerException e) { // NOPMD
throw e;
} catch (Throwable e) {
RxJavaPlugins.onError(e);
}
}看代码我们知道最主要调用的方法是：subscribeActual(observer);，这个方法是Observable里的抽象方法，而此时我们的Observable是一个ObservableCreate对象(前面create方法返回的对象)。所以我们去看一下ObservableCreate里面是如何重写这个方法的。代码如下：
protected void subscribeActual(Observer observer) {
CreateEmitter parent = new CreateEmitter(observer);
observer.onSubscribe(parent);
try {
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
我们一看到这个方法主要做了三件事：
①创建一个CreateEmitter对象parent。
②把parent传给source的subscribe方法。上面我们知道source就是刚才存的ObservableOnSubscribe对象，subscribe也就是我们重写的方法：
@Override
public void subscribe(ObservableEmitter e) throws Exception {
e.onNext("a");
} 所以我们在这个方法里就能收到一个CreateEmmiter，通过CreateEmitter可以回调相应的方法。CreateEmitter是实现ObservableEmitter接口，我们看看它内部实现，如：onNext源码如下：
@Override
public void onNext(T t) {
observer.onNext(t);
}也就是说，当我们在ObservableOnSubscribe的subscribe方法里调用ObservableEmitter的onNext方法的时候，它里面会调用observer的onNext。于是通过这样的传递，我们就能在observer里响应的回调方法里收到事件的相关状态。
至此一个简单Rxjava流式传递原理已经讲完了，总结流程如下：
使用Observbable.create方法，产生一个ObservableCreate对象，对象里存着ObservableOnSubscribe对象source。
调用ObservableCreate.subscribe方法，实际调用的是subscribeActual方法，传入一个Observer对象。
subscribeActual方法中创建一个CreateEmmiter对象，调用source.subscribe方法，传入CreateEmmiter对象。
于是我们在ObservableOnSubscribe中就接收到了一个CreateEmmiter，CreateEmmiter是ObservableEmmiter的子类。我们可以在这里调用CreateEmmiter的方法进行事件回调。
调用CreateEmmiter方法，实际上会调用Observer的响应的方法。也就是CreateEmmiter把事件状态传递给观察者。
Android架构师之路很漫长，所以我们从这份学习内容和计划开始执行吧！喜欢的话别忘记点击关注和赞哦
分享名称：还不会Rxjava响应式编程框架设计，先从这篇文章入手
网站地址：http://scyingshan.cn/article/gpcgpd.html
