java响应式编程框架_Android框架第（三）篇--响应式编程

清疚 2022-11-09 03:53 216阅读 0赞

![1e79ebbb88b7][]

RxJava响应式编程

上篇简单的阐述了响应式编程的基本理论。这篇主要对响应编程进行详细的介绍。以及主要介绍响应编程的主要实现-RxJava。其中RxJava的操作符和调度器(Scheduler)主要介绍对象。

一、什么是响应式编程？

响应式编程：是一种基于异步数据流概念的编程模式。关键概念：事件 把现实中的事件， 搬到软件中区去， 事件过程等待 触发到另一种事件。响应式编程最通用的一个场景是UI：我们的移动App必须做出对网络调用、用户触摸输入和系统弹框的响应。RxJava是响应式编程的具体实现，是异步数据处理库 扩展的观察者模式。

二、扩展观察者模式

扩展的观察者模式：有四个基本的要素：Observable (被观察者)、 Observer (观察者)、 subscribe (订阅)、event(事件)。Observable (被观察者)和 Observer (观察者)通过 subscribe() 方法实现订阅关系，Observable就可以在需要的时候通知Observer。其中onCompleted、onError、onNext是必须要实现的方法，他们的含义分别是：

1.onCompleted：事件队列完结，RxJava 不仅把每个事件单独处理，还会把它们看做一个队列。当不会再有新的 onNext发出时，需要触发 onCompleted() 方法作为完成标志。

2.onError：事件队列异常，在事件处理过程中出异常时，onError() 会被触发，同时队列自动终止，不允许再有事件发出。

3.onNext：普通的事件，将要处理的事件添加到事件队列中。

4.onStart：它会在事件还未发送之前被调用，可以用于做一些准备工作。例如数据的清零或重置，这是一个可选方法，默认情况下它的实现为空。

三、基本概念

RxJava的四个基本角色：Observable、Observer、Subscriber、Subject。Observable和Subject是两个“生产”实体，Observer和Subscriber是两个“消费”实体。RxJava本质上是Rx是一个函数库，让开发者可以利用可观察序列和LINQ风格查询操作符来编写异步和基于事件的程序，开发者可以用Observables表示异步数据流，用LINQ操作符查询异步数据流， 用Schedulers参数化异步数据流的并发处理，Rx可以这样定义：Rx = Observables + LINQ + Schedulers。

四、RxJava操作符的分类

操作符的分类

1.Creating ObServables(创建Observable)

2.Transforming Observables(转换Observabl)

3.Filtering Observables(过滤Observable)

4.Combing Observables(组合Observable)

5.Error Handling Operators(处理错误)

4.1Creating ObServables(创建Observable)

Create

Just

From

Defer

Empty/Never/Throw

Interval

Range

Repeat

Start

Timer

\*\*\*

1.原始的创建：Observable.create()

private static void create() \{ //一个原始的创建

Observable.create(new OnSubscribe()\{

@Override

public void call(Subscriber super String> subscriber) \{

subscriber.onNext("RxJava学习");

\}).subscribe(new Subscriber()\{

@Override

public void onCompleted() \{

System.out.println("onCompleted()");

@Override

public void onError(Throwable arg0) \{

System.out.println("onError()");

@Override

public void onNext(String arg0) \{

System.out.println("onNext():"+arg0);

\});

2.Observable.just() 方法可以传入一到九个参数，它们会按照传入的参数的顺序来发射它们。 just() 方法也可以接受列表或数组，

private static void just() \{//对create的简写 快捷键

Observable.just("sdgsagsag").subscribe(new Subscriber()\{

@Override

public void onCompleted() \{

System.out.println("onCompleted()");

@Override

public void onError(Throwable arg0) \{

// TODO Auto-generated method stub

@Override

public void onNext(String arg0) \{

System.out.println("onNext():"+arg0);

\});

3.Observable.from() 创建符可以从一个列表/数组来创建Observable,并一个接一个的从列表/数组中发射出来每一个对象，或者也可以从Java Future 类来创建Observable，并发射Future对象的 .get() 方法返回的结果值。传入 Future 作为参数时，我们可以指定一个超时的值。Observable将等待来自 Future 的结果；如果在超时之前仍然没有结果返回，Observable将会触发 onError() 方法通知观察者有错误发生了。

private static void from() \{ //创建一个数组和列表转换成其他数据类型的对象 和数据类型

ArrayList items = new ArrayList();

items.add(1);

items.add(2);

items.add(3);

items.add(5);

items.add(6);

items.add(7);

Observable.from(items).subscribe(new Subscriber()\{

@Override

public void onCompleted() \{

System.out.println("onCompleted()");

@Override

public void onError(Throwable arg0) \{

// TODO Auto-generated method stub

@Override

public void onNext(Integer arg0) \{

System.out.println("onNext():"+arg0);

\});

4.Observable.defer()对象后赋值 也能输出来

private static String valuestr;

private static void defer() \{//Observable对象后赋值 也能输出来

ArrayList items = new ArrayList();

items.add(1);

items.add(2);

items.add(3);

items.add(4);

items.add(5);

items.add(6);

items.add(7);

Observable observable = Observable.defer(new Func0>() \{

@Override

public Observable call() \{

return Observable.just(valuestr);

\});

valuestr = "速度更快sag凯撒";

observable.subscribe(new Subscriber()\{

@Override

public void onCompleted() \{

System.out.println("onCompleted()");

@Override

public void onError(Throwable arg0) \{

@Override

public void onNext(String arg0) \{

System.out.println("onNext():"+arg0);

\});

5.Obervable.range()对象后赋值 也能输出来

private static void range() \{//Observable对象后赋值 也能输出来

ArrayList items = new ArrayList();

items.add(1);

items.add(2);

items.add(3);

items.add(4);

items.add(5);

items.add(6);

items.add(7);

Observable observable = Observable.range(1, 5).repeat(4);

observable.subscribe(new Subscriber()\{

@Override

public void onCompleted() \{

System.out.println("onCompleted()");

@Override

public void onError(Throwable arg0) \{

@Override

public void onNext(Integer arg0) \{

System.out.println("onNext():"+arg0);

\});

4.2Transforming Observables(转换Observable)

1.Map:将一个对象转换为我们任何一个想要的对象

2.FlatMap:Observable.flatMap()接收一个Observable的输出作为输入，同时输出另外一个Observable 一个对象返回Observable对象 一对多

3.GroupBy:通过一定的规则对数据列表进行分组和归类。

4.Buffer:一次性把Observable数据对象发射到我们的观察者

5.Scan:把前面的数据进行累加 发射到观察者

private static void testStranfrom() \{ //整形转换成字符串

Observable.just(123).map(new Func1() \{

@Override

public String call(Integer arg0) \{

// TODO Auto-generated method stub

return arg0 +"";

\}).subscribe(new Subscriber()\{

@Override

public void onCompleted() \{

System.out.println("onCompleted()");

@Override

public void onError(Throwable arg0) \{

@Override

public void onNext(String arg0) \{

System.out.println("onNext():"+arg0);

\});

private static void textflatmap() \{

Observable.just(1,2,3,4,5,6).flatMap(new Func1>() \{

@Override

public Observable extends String> call(Integer arg0) \{

// TODO Auto-generated method stub

return Observable.just(arg0+"");

\}).subscribe(new Subscriber()\{

@Override

public void onCompleted() \{

System.out.println("onCompleted()");

@Override

public void onError(Throwable arg0) \{

@Override

public void onNext(String arg0) \{

System.out.println("onNext():"+arg0);

\});

private static void groupby() \{

Observable.just(1,2,3,4,5,6).groupBy(new Func1() \{

@Override

public Integer call(Integer arg0) \{

// TODO Auto-generated method stub

return arg0 % 2; //分组的规则

\}).subscribe(new Observer>()\{

@Override

public void onCompleted() \{

System.out.println("onCompleted()");

@Override

public void onError(Throwable arg0) \{

@Override

public void onNext(final GroupedObservable arg) \{

arg.subscribe(new Subscriber()\{

@Override

public void onCompleted() \{

@Override

public void onError(Throwable arg0) \{

@Override

public void onNext(Integer arg0) \{

System.out.println("groupbt onNext():"+arg.getKey()+"data:"+arg0);

\});

private static void buffer() \{

Observable.range(1, 5).buffer(2).subscribe(new Observer>()\{

@Override

public void onCompleted() \{

@Override

public void onError(Throwable arg0) \{

@Override

public void onNext(List arg0) \{

System.out.println("onNext:"+arg0);

\});

private static void scan() \{ //每次据取两个数据项，然后向观察者发射

Observable.range(1, 5).scan(new Func2()\{

@Override

public Integer call(Integer sum, Integer arg1) \{

// TODO Auto-generated method stub

return sum + arg1;

\}).subscribe(new Observer()\{

@Override

public void onCompleted() \{

// TODO Auto-generated method stub

@Override

public void onError(Throwable arg0) \{

// TODO Auto-generated method stub

@Override

public void onNext(Integer arg0) \{

// TODO Auto-generatd method stube

System.out.println("onNext:"+arg0);

\});

4.3Filtering Observables(过滤Observable)

1.Debounce: 一定的时间内 数据项后面没有产生数据 将该数据发送出去

2.Distinct:去掉重复的

3.ElementAt:指定位置的数据

4.Filter:按照我们指定的规则来进行过滤 得道我们想要的数据

5.First:取列表中的第一位数据

6.IgnoreElements: 忽略列表上的所有元素 事件结束的时候回调Complement函数

7.Last:我们指定的最后数位项数据

8.Sample:对数据样本数据的采集，采集后发射到观察者 定时的取样 取到的数据放到一个列表中 再发射。

9.Skip:跳跃指定数据项的数据 然后取后面的数据

10.SkipLast:跳过数据的后几项

11.Take:取值的意思 在数据列表中 如果只取数据项的前几项

12.TakeLast:取数据列表的最后几项数据 指定的几位数据

private static void testDebounce() \{ //在一定的时间内 数据项后面没有产生数据 将该数据发送出去

Observable.create(new OnSubscribe() \{

@Override

public void call(Subscriber super Integer> arg0) \{

for(int i = 0; i <10; i++) \{

try \{

Thread.sleep(1000);

arg0.onNext(1);

\} catch (InterruptedException e) \{

arg0.onError(e);

\}).debounce(1,TimeUnit.SECONDS).subscribe(new Subscriber()\{

@Override

public void onCompleted() \{

// TODO Auto-generated method stub

System.out.println("onCompleted():" + "\\n");

@Override

public void onError(Throwable arg0) \{

// TODO Auto-generated method stub

System.out.println("onError:" + arg0+ "\\n");

@Override

public void onNext(Integer arg0) \{

// TODO Auto-generated method stub

System.out.println("onNext:" + arg0+ "\\n");

\});

private static void testDistinct() \{ //去掉重复的

Observable.just(1, 2, 3, 3, 2).distinct().subscribe(new Subscriber()\{

@Override

public void onCompleted() \{

// TODO Auto-generated method stub

System.out.println("onCompleted():" + "\\n");

@Override

public void onError(Throwable arg0) \{

// TODO Auto-generated method stub

System.out.println("onError:" + arg0+ "\\n");

@Override

public void onNext(Integer arg0) \{

// TODO Auto-generated method stub

System.out.println("onNext:" + arg0+ "\\n");

\});

｝

private static void testElementAt() \{ //指定位置的数据

Observable.just(1, 2, 2, 4).elementAt(2).subscribe(new Subscriber()\{

@Override

public void onCompleted() \{

System.out.println("onCompleted():" + "\\n");

@Override

public void onError(Throwable arg0) \{

System.out.println("onError:" + arg0+ "\\n");

@Override

public void onNext(Integer arg0) \{

System.out.println("onNext:" + arg0+ "\\n");

\});

private static void testFilter() \{ //去掉重复的 过滤

Observable.just(1, 2, 3, 3, 2).distinct().filter(new Func1()\{

public Boolean call(Integer arg0) \{

return arg0 > 2;

\}).subscribe(new Subscriber()\{

@Override

public void onCompleted() \{

// TODO Auto-generated method stub

System.out.println("onCompleted():" + "\\n");

@Override

public void onError(Throwable arg0) \{

// TODO Auto-generated method stub

System.out.println("onError:" + arg0+ "\\n");

@Override

public void onNext(Integer arg0) \{

// TODO Auto-generated method stub

System.out.println("onNext:" + arg0+ "\\n");

\});

private static void testFirst() \{ //指定位置的数据

Observable.just(1, 2, 2, 4).first().subscribe(new Subscriber()\{

@Override

public void onCompleted() \{

System.out.println("onCompleted():" + "\\n");

@Override

public void onError(Throwable arg0) \{

System.out.println("onError:" + arg0+ "\\n");

@Override

public void onNext(Integer arg0) \{

System.out.println("onNext:" + arg0+ "\\n");

\});

private static void testIgnoreElement() \{

Observable.create(new OnSubscribe()\{

@Override

public void call(Subscriber super Integer> arg0) \{

// TODO Auto-generated method stub

arg0.onNext(233);

//throw new NullPointerException();

\}).ignoreElements().subscribe(new Subscriber()\{

@Override

public void onCompleted() \{

// TODO Auto-generated method stub

System.out.println("onCompleted():" + "\\n");

@Override

public void onError(Throwable arg0) \{

// TODO Auto-generated method stub

System.out.println("onError:" + arg0+ "\\n");

@Override

public void onNext(Integer arg0) \{

// TODO Auto-generated method stub

System.out.println("onNext:" + arg0+ "\\n");

\});

private static void testSample() \{ //定时对数据进行采样 再发送出去

Observable.create(new OnSubscribe() \{

@Override

public void call(Subscriber super Integer> arg0) \{

for(int i = 0; i <10; i++) \{

try \{

Thread.sleep(1000);

arg0.onNext(i);

\} catch (InterruptedException e) \{

arg0.onError(e);

arg0.onCompleted();

\}).sample(1,TimeUnit.SECONDS).subscribe(new Subscriber()\{

@Override

public void onCompleted() \{

// TODO Auto-generated method stub

System.out.println("onCompleted():" + "\\n");

@Override

public void onError(Throwable arg0) \{

// TODO Auto-generated method stub

System.out.println("onError:" + arg0+ "\\n");

@Override

public void onNext(Integer arg0) \{

// TODO Auto-generated method stub

System.out.println("onNext:" + arg0+ "\\n");

\});

private static void testSkip() \{ //跳过多少项目

Observable.just(1, 2, 2, 4).skip(3).subscribe(new Subscriber()\{

@Override

public void onCompleted() \{

System.out.println("onCompleted():" + "\\n");

@Override

public void onError(Throwable arg0) \{

System.out.println("onError:" + arg0+ "\\n");

@Override

public void onNext(Integer arg0) \{

System.out.println("onNext:" + arg0+ "\\n");

\});

private static void testTake() \{ //取出多少项目

Observable.just(1, 2, 2, 4).take(3).subscribe(new Subscriber()\{

@Override

public void onCompleted() \{

System.out.println("onCompleted():" + "\\n");

@Override

public void onError(Throwable arg0) \{

System.out.println("onError:" + arg0+ "\\n");

@Override

public void onNext(Integer arg0) \{

System.out.println("onNext:" + arg0+ "\\n");

\});

4.3Combing Observables(组合Observable)

1.Zip:在一段时间内将最近的元素进行组合在一起

2.Merge: 按照时间先后顺序 进行组合

3.StartWith: 在当前的数据之前插入 我们想要的数据 (数据类型或者Observable)

4.CombineLatest: 最新接触 组合数据

5.Join:

6.SwithchOnNext:后面的覆盖前面的数据组合

/\*\*

\* 用来合并两个Observable发射的数据项，根据Func2函数生成一个新的值并发射出去。其中一个Observable发送数据结束或者出现异常后，

\* 另外一个Observable也将停在发射数据

\*/

private static void testZip() \{

Observable observable1 = Observable.just(10, 20, 30);

Observable observable2 = Observable.just(4, 8, 12, 16);

Observable.zip(observable1, observable2, new Func2()\{

@Override

public Integer call(Integer arg0, Integer arg1) \{

return arg0 + arg1;

\}).subscribe(new Subscriber()\{

@Override

public void onCompleted() \{

System.out.println("onCompleted");

@Override

public void onError(Throwable arg0) \{

System.out.println("onError():"+arg0.getMessage());

@Override

public void onNext(Integer arg0) \{

System.out.println("onNext arg0 "+arg0);

\});

/\*\*

\* 将两个Observable发射的事件序列组合合并成一个事件序列，就像是Observable发射的一样。你可以简单的将它理解成两个Observable合并成一个Observable对象

\*/

private static void testMerg() \{

Observable observable1 = Observable.just(10, 20, 30);

Observable observable2 = Observable.just(4, 8, 12, 16);

Observable.merge(observable1,observable2).subscribe(new Subscriber()\{

@Override

public void onCompleted() \{

System.out.println("onCompleted");

@Override

public void onError(Throwable arg0) \{

System.out.println("onError():"+arg0.getMessage());

@Override

public void onNext(Integer arg0) \{

System.out.println("onNext arg0 "+arg0);

\});

/\*\*

\* 用于在源Observable发射的数据前插入数据.使用startWith(Interable)我们还可以在源Observable发射的数据前插入Iterable

\*/

private static void testStartWith() \{

Observable first = Observable.just(1, 3, 5);

Observable second = Observable.just(2, 4, 6);

first.startWith(second).subscribe(new Subscriber()\{

@Override

public void onCompleted() \{

System.out.println("onCompleted");

@Override

public void onError(Throwable arg0) \{

System.out.println("onError():"+arg0.getMessage());

@Override

public void onNext(Integer arg0) \{

System.out.println("onNext arg0 "+arg0);

\});

/\*\*

\* 用于将两个Observable最近发射的数据已经Func2函数的规则进展集合 5+2 5+4 5+6

\*/

private static void testCombineLatest() \{

Observable first = Observable.just(1, 3, 5);

Observable second = Observable.just(2, 4, 6);

first.combineLatest(first, second, new Func2()\{

@Override

public Integer call(Integer arg0, Integer arg1) \{

// TODO Auto-generated method stub

return arg0 + arg1;

\}).subscribe(new Subscriber()\{

@Override

public void onCompleted() \{

System.out.println("onCompleted");

@Override

public void onError(Throwable arg0) \{

System.out.println("onError():"+arg0.getMessage());

@Override

public void onNext(Integer arg0) \{

System.out.println("onNext arg0 "+arg0);

\});

/\*\*

\* join()函数基于时间窗口将两个Observables发射的数据结合在一起，每个Observable在自己的时间窗口内独有有效的，都可以拿来组合

\*/

private static void testJoin() \{

Observable first = Observable.just(1, 3, 5);

Observable second = Observable.create(new OnSubscribe()\{

@Override

public void call(Subscriber super Integer> arg0) \{

try\{

for(int i = 0; i < 5; i++) \{

arg0.onNext(i);

arg0.onCompleted();

\}catch(Exception e)\{

e.printStackTrace();

\});

4.4 Error Handling Operators(处理错误)

Catch:捕获异常处理

\*\*onErrorReturn: \*\*遇见异常正常结束数据流

onErrorResumeNext:当数据遇见异常，用正常数据代替错误数据项 不可以捕获到异常信息

onExceptionResumeNext:当有异常的时候，可以用正确的数据流替换掉，并且和、可以捕获到异常信息

五、Schedulers(调度器)种类

调度器(Scheduler)是RxJava以一种极其简单的方式解决多线程问题机制。

Schedulers(调度器)种类：

.io() I/O操作

.computation()计算 计算类型

.immediate() 这个调度器允许你立即在当前线程执行你指定的工作

.newThread() 建立一个新的线程 它为指定任务启动一个新的线程

.trampoline() 按照顺序排队处理的 调度器将会按序处理队列，并运行队列中每一个任务

[1e79ebbb88b7]: /images/20221023/c5bfe976c42e41a5b76b303dab6ea057.png

java响应式编程框架_Android框架第（三）篇--响应式编程

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 Reactor响应式编程

相关响应式编程介绍

相关 java响应式编程框架_Android框架第（三）篇--响应式编程

相关响应式编程html,Reactive响应式编程Spring webFlux

相关异步编程与响应式框架

相关响应式编程笔记

相关 java异步响应式编程

相关响应式编程规范

相关 Bootstrap响应式框架

相关 Angular--响应式编程