湖北省建设工程质量协会网站,云主机推荐,北京营销网站建设公司,基础建设的网站有哪些内容随着互联网的迅猛发展#xff0c;Java已成为最广泛应用于后端开发的语言之一。而在处理异步操作和事件驱动编程方面#xff0c;传统的Java多线程并不总是最佳选择。这时候#xff0c;RxJava作为一个基于观察者模式、函数式编程和响应式编程理念的库#xff0c;为我们提供了…随着互联网的迅猛发展Java已成为最广泛应用于后端开发的语言之一。而在处理异步操作和事件驱动编程方面传统的Java多线程并不总是最佳选择。这时候RxJava作为一个基于观察者模式、函数式编程和响应式编程理念的库为我们提供了一种强大而灵活的解决方案。
简介
RxJava是 ReactiveX 家族的重要一员, ReactiveX 是 Reactive Extensions 的缩写一般简写为 Rx。ReactiveX官方给Rx的定义是Rx是一个使用可观察数据流进行异步编程的编程接口。 ReactiveX 不仅仅是一个编程接口它是一种编程思想的突破它影响了许多其它的程序库和框架以及编程语言。它拓展了观察者模式使你能够自由组合多个异步事件而不需要去关心线程同步线程安全并发数据以及I/O阻塞。 RxJava在Java环境下使用它通过Observable可观测对象和Subscriber订阅者来实现异步编程模型。Observable可以发射出一系列的数据流而Subscriber则负责处理这些数据流。利用各种操作符我们可以对数据流进行变换、过滤、合并等操作从而完成复杂的异步任务。
GitHub - ReactiveX/RxJava: RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM. 结论 RxJava是 ReactiveX 在JVM上的一个实现ReactiveX使用Observable序列组合异步和基于事件的程序的库是一个 基于事件流、实现异步操作的库。
Observables · ReactiveX文档中文翻译
RxJava 是轻量级的
RxJava尽力做到非常轻巧。它仅关注Observable的抽象和与之相关的高层函数实现为一个单独的JAR文件。
RxJava 是一个多语言实现
RxJava 支持Java 6或者更新的版本以及其它的JVM语言如 Groovy, Clojure, JRuby, Kotlin 和 Scala。RxJava 可用于更多的语言环境而不仅仅是Java和Scala而且它致力于尊重每一种JVM语言的习惯。
RxJava 第三方库
下面是可与RxJava协作的第三方库 Hystrix - 用于分布式系统的一个延时和容错处理框架 Camel RX - 一个用于Apache Camel 的 RxJava 兼容层 rxjava-http-tail - 让你可以跟踪HTTP日志就像使用 tail -f 一样 mod-rxvertx - Extension for VertX - 使用 RxJava 封装的VertX库 rxjava-jdbc - 使用RxJava流式处理JDBC连接还支持语句的函数式组合 rtree - 使用RxJava实现的一个纯内存的可变的R-tree和R*-tree
使用指南
你可以在Maven Central http://search.maven.org 找到用于Maven, Ivy, Gradle, SBT和其它构建工具需要的二进制文件和依赖信息.
Maven示例:
dependencygroupIdio.reactivex.rxjava3/groupIdartifactIdrxjava/artifactIdversion3.1.7/version
/dependencyRxJava使用三步曲
RxJava的使用可以概括为三个步骤创建 Observable定义 Observer 处理数据流最后订阅SubscribeObservable。
创建 Observable 可以直接使用 Observable.just() 方法来创建一个发射固定数据项的 Observable 也可以通过 Observable.fromIterable() 方法来创建包含多个数据项的 Observable。
ObservableString observable Observable.just(Hello, World);定义 Observer
创建一个 Observer 对象并实现它的各个方法。在这些方法中你可以处理每个发射的数据项、对错误进行处理或者在数据全部发射完毕时执行一些操作。
ObserverString observer new ObserverString() {Overridepublic void onSubscribe(Disposable d) {// 在此方法中进行一些初始化操作或资源管理}Overridepublic void onNext(String s) {// 处理每个发射的数据项System.out.println(s);}Overridepublic void onError(Throwable e) {// 处理发生的异常情况}Overridepublic void onComplete() {// 完成所有的数据发射操作}
};订阅 Observable
observable.subscribe(observer);RxJava基本原理
生活例子引入
用一个生活例子引入来介绍 RxJava的基本原理 顾客到饭店吃饭
RxJava原理介绍 RxJava原理 基于 一种扩展的观察者模式 RxJava的扩展观察者模式中有4个角色
角色作用类比被观察者Observable产生事件顾客观察者Observer接收事件并给出响应动作厨房订阅Subscribe连接 被观察者 观察者服务员事件Event被观察者 观察者 沟通的载体菜
请结合上述 顾客到饭店吃饭 的生活例子理解 即RxJava原理可总结为被观察者 Observable 通过 订阅Subscribe 按顺序发送事件 给观察者 Observer 观察者Observer 按顺序接收事件 作出对应的响应动作。具体如下图
代码实现
步骤1创建被观察者 **Observable**** 生产事件**
即 顾客入饭店 - 坐下餐桌 - 点菜
// 步骤1创建被观察者 Observable 生产事件
// 即 顾客入饭店 - 坐下餐桌 - 点菜
// 1. 创建被观察者 Observable 对象
ObservableString observable Observable.create(new ObservableOnSubscribeString() {
// create() 是 RxJava 最基本的创造事件序列的方法
// 此处传入了一个 OnSubscribe 对象参数
// 当 Observable 被订阅时OnSubscribe 的 call() 方法会自动被调用即事件序列就会依照设定依次被触发
// 即观察者会依次调用对应事件的复写方法从而响应事件
// 从而实现被观察者调用了观察者的回调方法 由被观察者向观察者的事件传递即观察者模式// 2. 在复写的subscribe里定义需要发送的事件Overridepublic void subscribe(ObservableEmitterString emitter) throws Exception {// 通过 ObservableEmitter类对象产生事件并通知观察者// ObservableEmitter类介绍// 2.1 定义事件发射器// 2.2 作用定义需要发送的事件 向观察者发送事件emitter.onNext(event01);emitter.onNext(event02);emitter.onNext(event03);emitter.onComplete();}
});步骤2创建观察者 **Observer**** 并 定义响应事件的行为** 即 开厨房 - 确定对应菜式 发生的事件类型包括Next事件、Complete事件 Error事件。具体如下
事件类型定义作用使用规则使用方法Next普通事件向观察者发送需要响应事件的信号被观聚者可发送无限个Next事件观察者可接受无限个Next事件onNext()Complete表示所有的事件都已经成功完成RxJava把所有时间当作队列处理标志 被观察者 不再发送普通事件Next当被观察者发送了一个Complete事件后被观察者在Complete事件后的事件将会继续发送但观察者收到Complete事件后将不再继续接收任何事件被观察者可以不发送Complete事件。onComplete()Error事件队列异常事件标志 事件处理过程中出现异常此时队列自动终止不允许再有事件发出当被观察者发送了一个Error事件后被观察者在Error事件后的事件将会继续发送但观察者收到Error事件后将不再继续接收任何事件被观察者可以不发送Error事件。onError()
// 1. 创建观察者 Observer 对象
ObserverString observer new ObserverString() {// 2. 创建对象时通过对应复写对应事件方法 从而 响应对应事件// 观察者接收事件前默认最先调用复写 onSubscribeOverridepublic void onSubscribe(Disposable d) {}// 当被观察者生产Next事件 观察者接收到时会调用该复写方法 进行响应Overridepublic void onNext(String value) {System.out.println(对Next事件作出响应 value);}// 当被观察者生产Error事件 观察者接收到时会调用该复写方法 进行响应Overridepublic void onError(Throwable e) {}// 当被观察者生产Complete事件 观察者接收到时会调用该复写方法 进行响应Overridepublic void onComplete() {}
};步骤3通过订阅**Subscribe**连接观察者和被观察者
即 顾客找到服务员 - 点菜 - 服务员下单到厨房 - 厨房烹调
observable.subscribe(observer);
// 或者 observable.subscribe(subscriber)Subject
来看⼀个⾮常特殊的类型- Subject 为什么说它特殊呢原因很简单它同时充当了Observer和Observable的角色。因为它是一个Observer它可以订阅一个或多个Observable又因为它是一个Observable它可以转发它收到(Observe)的数据也可以发射新的数据。
由于一个Subject订阅一个Observable它可以触发这个Observable开始发射数据如果那个Observable是冷的–就是说它等待有订阅才开始发射数据。因此有这样的效果Subject可以把原来那个冷的Observable变成热的。
Subject的种类
针对不同的场景一共有四种类型的Subject。他们并不是在所有的实现中全部都存在而且一些实现使用其它的命名约定例如在RxScala中Subject被称作PublishSubject。
AsyncSubject
一个AsyncSubject只在原始Observable完成后发射来自原始Observable的最后一个值。如果原始Observable没有发射任何值AsyncObject也不发射任何值它会把这最后一个值发射给任何后续的观察者。
AsyncSubject asyncSubject AsyncSubject.create();
// 发送事件
asyncSubject.onNext(1);
// 订阅
asyncSubject.subscribe(event - {System.out.println(event);
});
asyncSubject.onNext(3);
// 再次发送事件
asyncSubject.onNext(4);
asyncSubject.onComplete();
// 只会监听到 事件4PublishSubject
可以不需要初始来进行初始化也就是可以为空并且它只会向订阅者发送在订阅之后才接收到的元素。
// 初始化⼀个PublishSubject
PublishSubject publishSubject PublishSubject.create();
// 发送事件
publishSubject.onNext(1);
// 订阅
publishSubject.subscribe(event - {System.out.println(event);
});
// 再次发送事件
publishSubject.onNext(2);
publishSubject.onNext(3);事件1是无法被订阅的只接受订阅之后的响应 BehaviorSubject
当观察者订阅BehaviorSubject时它开始发射原始Observable最近发射的数据如果此时还没有收到任何数据它会发射一个默认值然后继续发射其它任何来自原始Observable的数据。
