关于RxJava，其实大家都应该很熟悉了。最早了解RxJava的时候，是看的抛物线大神的关于RxJava的讲解，不过那个时候是基于RxJava 1.0的版本做的讲解，如今，RxJava已经更新到3.0了，所以自己再回头学习学习。
定义：
是一个在Java虚拟机上，使用可观察的序列构成基于事件的，异步的程序库。
优点：
（1）简洁，通过链式的方式去处理请求，代码易懂而且非常有条理性。
（2）是一种响应式编程，可以很好的处理各种多变需求且不会影响到代码的扩展性和维护性。
响应式编程：
响应式编程是一种基于异步数据流概念的编程模式。数据流就像一条河：它可以被观测，被过滤，被操作，或者为新的消费者与另外一条流合并为一条新的流。响应式编程的一个关键概念是事件。事件可以被等待，可以触发过程，也可以触发其它事件。事件是唯一的以合适的方式将我们的现实世界映射到我们的软件中。
RxJava的基本使用：
** 第一步：初始化 Observable **
** 第二步：初始化 Observer **
** 第三步：建立订阅关系 **
// 1.创建观察者（Observer）
val observer = object : Observer<String> {
// 当观察者被订阅时回调
override fun onSubscribe(d: Disposable?) {
Log.i(TAG, "onSubscribe")
}
// 当观察者收到onNext事件时回调
override fun onNext(t: String?) {
Log.i(TAG, "onNext-->$t")
}
// 当观察者收到onError事件时回调
override fun onError(e: Throwable?) {
Log.i(TAG, "onError")
}
// 当观察者收到onComplete事件时回调
override fun onComplete() {
Log.i(TAG, "onComplete")
}
}
// 2.创建被观察者（Observable）
val observable = Observable.create(ObservableOnSubscribe<String> { emitter ->
emitter?.onNext("1")
emitter?.onNext("2")
emitter?.onNext("3")
emitter?.onNext("4")
emitter?.onComplete()
})
// 3.订阅
observable.subscribe(observer)
当然我们也可以使用链式完成上面的功能
// 使用链式调用完成上面的功能
Observable.create(ObservableOnSubscribe<String> { emitter ->
emitter?.onNext("1")
emitter?.onNext("2")
emitter?.onNext("3")
emitter?.onNext("4")
emitter?.onComplete()
}).subscribe(object : Observer<String> {
override fun onSubscribe(d: Disposable?) {
Log.i(TAG, "onSubscribe")
}
override fun onNext(t: String?) {
Log.i(TAG, "onNext-->$t")
}
override fun onError(e: Throwable?) {
Log.i(TAG, "onError")
}
override fun onComplete() {
Log.i(TAG, "onComplete")
}
})
需要说明的是：
被观察者中有个Emitter（发射器），发射器的位置位于subscribe回调参数ObservableEmitter e中，RxJava的事件通知就是通过它来进行发送的，发射器能发送的事件有onNext()，onComplete()，onError()，在观察者的回调中，分别对应着相同方法名进行回调。
onSubscribe()：当观察者被订阅时回调
onNext()：当观察者收到onNext事件时回调
onComplete()：当观察者收到onComplete事件时回调
onError()：当观察者收到onError事件时回调
RxJava的线程调度
在RxJava中的线程调度是RxJava的一大核心亮点之一，因为RxJava遵循的是事件在哪个线程产生，就在哪个线程消费，然而，在实际的开发过程中，很多的时候我们必须要进行切换线程操作，比如，我们通常将耗时的操作放置在子线程，然后将View的显示放置在UI线程，所以当数据处理完毕后（读取文件、请求网络等等）我们需要切换线程，以往我们更多使用Handler来进行处理，而如今我们可以使用RxJava来进行处理，究其原因，在于RxJava进行线程调度的时候非常方便，看下面这个例子：
//创建被观察者
Observable.create(new ObservableOnSubscribe<String>() {
@Override
//默认在主线程里执行该方法
public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
e.onNext("1");
e.onNext("2");
e.onNext("3");
e.onNext("4");
e.onComplete();
}
})
//将被观察者切换到子线程
.subscribeOn(Schedulers.io())
//将观察者切换到主线程 需要在Android环境下运行
.observeOn(AndroidSchedulers.mainThread())
//创建观察者并订阅
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
System.out.println("onNext=" + s);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
简单的两句
//将被观察者切换到子线程
subscribeOn(Schedulers.io())
//将观察者切换到主线程 需要在Android环境下运行
observeOn(AndroidSchedulers.mainThread())
就可以完美切换线程，使用它不香吗？当然，除了这两种还有其他的方式，如下：
线程调度器
在RxJava中也内置了多种线程选线供我们使用。这些内置的Scheduler已经足够满足我们开发的需求， 因此我们应该使用内置的这些选项,，在RxJava内部使用的是线程池来维护这些线程, 所以效率也比较高。
（1）Schedulers.io()：代表io操作的线程, 通常用于网络,读写文件等io密集型的操作。
（2）Schedulers.computation()： 代表CPU计算密集型的操作, 例如需要大量计算的操作。
（3）Schedulers.newThread()：代表一个常规的新线程。
（4）AndroidSchedulers.mainThread()：代表Android的主线程。
RxJava常用操作符
在RxJava中，操作符也是非常重要的一个环节，熟练掌握并使用好操作符可以达到事半功倍的效果，而关于操作符的使用，可以看看这篇博客，博主写的非常完善，为这么厉害的博主点个赞，具体的链接为：
<u>https://blog.csdn.net/weixin_42046829/article/details/104836592</u>
结语：现在在项目中RxJava已经成为了标配，而在此基础上还产生了诸如RxAndroid,RxBus等等优秀的框架，作为响应式编程的组件，用好它可以给我们的开发带来非常棒的体验。
