RxJava基本使用
更多文章请点击链接:
转载请标明出处:https://www.cnblogs.com/tangzh/p/12088300.html,
rxjava究竟是啥,从根本上来讲,它就是一个实现异步操作的库,并且能够使代码非常简洁。它的异步是使用观察者模式来实现的。
关于观察者模式的介绍,可以看我这一篇文章:
https://www.cnblogs.com/tangzh/p/11175120.html
这里我主要讲rxjava的一些基本用法,基本案例,原理的话暂时不深究:
一、自己构造事件
observable.create(new observableonsubscribe<integer>() { @override public void subscribe(observableemitter emitter) { int i = getnumber(); if (i < 0) { emitter.oncomplete(); return; } else { log.d(tag, thread.currentthread().getname()); emitter.onnext(i); emitter.oncomplete(); } } }) .subscribeon(schedulers.io()) .observeon(androidschedulers.mainthread()) .subscribe(new consumer<integer>() { @override public void accept(integer integer) throws exception { log.d(tag, thread.currentthread().getname()); log.d(tag, integer + ""); } }, new consumer<throwable>() { @override public void accept(throwable throwable) throws exception { } });
rxjava 有四个基本概念:observable
(可观察者,即被观察者)、 observer
(观察者)、 subscribe
(订阅)、事件。observable
和 observer
通过 subscribe()
方法实现订阅关系,从而 observable
可以在需要的时候发出事件来通知 observer
。
onnext():方法用来发送事件。
下面看看其他两个方法:
oncompleted()
: 事件队列完结。rxjava 不仅把每个事件单独处理,还会把它们看做一个队列。rxjava 规定,当不会再有新的onnext()
发出时,需要触发oncompleted()
方法作为标志。onerror()
: 事件队列异常。在事件处理过程中出异常时,onerror()
会被触发,同时队列自动终止,不允许再有事件发出。- 在一个正确运行的事件序列中,
oncompleted()
和onerror()
有且只有一个,并且是事件序列中的最后一个。需要注意的是,oncompleted()
和onerror()
二者也是互斥的,即在队列中调用了其中一个,就不应该再调用另一个。
讲一下我们上面的例子,上面这个例子是采用简洁的链式调用来写的:
首先使用 create()
方法来创建一个 observable ,并为它定义事件触发规则,然后通过emitter.onnext(i)传递出来,.subscribeon(schedulers.io())便是指定该事件产生的所在的线程为子线程,.observeon(androidschedulers.mainthread())指定观察者执行的线程为主线程。这时候为止返回的对象为observable对象。
然后该observable对象subscribe绑定观察者(也就是观察者进行订阅),里面有接收被观察者发出来的事件,有一个成功的方法,和一个失败的方法,这样就实现了由被观察者向观察传递事件。
二、对集合里的数据进行变换
list<integer> list = new arraylist<integer>() { { add(0); add(1); add(2); } }; observable.fromiterable(list).map(new function() { @override public object apply(object o) throws exception { int i = (int) o + 1; return string.valueof(i); } }) .tolist() .toobservable().subscribeon(schedulers.io()) .subscribeon(androidschedulers.mainthread()) .subscribe(new consumer() { @override public void accept(object o) throws exception { log.d(tag, o.tostring()); } });
且看,我们需要对某个集合里面的数据一一进行变换,然后发送出来执行其他操作。
上面便是对集合里面的每一项进行加一操作,然后再转换为string类型,然后tolist(),组合成集合发送出来,最后在观察者方法中打印出每一项。
三、合并执行
定义两个被观察者,各自产生事件,然后合并在一起,发送给一个观察者。
首先定义我们上面第一个例子的被观察者,用于发送一个数字:
observable observable1 = observable.create(new observableonsubscribe<integer>() { @override public void subscribe(observableemitter emitter) { int i = getnumber(); if (i < 0) { emitter.oncomplete(); return; } else { log.d(tag, thread.currentthread().getname()); emitter.onnext(i); emitter.oncomplete(); } } }) .subscribeon(schedulers.io());
其次再定义我们上面第二个例子的被观察者:
list<integer> list = new arraylist<integer>() { { add(0); add(1); add(2); } }; observable observable2 = observable.fromiterable(list).map(new function() { @override public object apply(object o) { int i = (int) o + 1; return string.valueof(i); } }) .tolist() .toobservable().subscribeon(schedulers.io());
最后将这两个被观察者的事件合并起来发送给一个观察者:
disposable disposable = observable.zip(observable1, observable2, new bifunction() { @override public object apply(object o, object o2) throws exception { int i = (int) o; string k = (string) ((list) o2).get(0); return k + i; } }) .subscribe(new consumer() { @override public void accept(object o) { log.d(tag, (string) o); } }, new consumer<throwable>() { @override public void accept(throwable throwable) { log.d(tag, throwable.getmessage()); } });
zip方法,顾名思义,有点类似与于打包的意思。
o为被观察者1返回的结果,o2为被观察2返回的结果,将这两个结果一起处理后发送给观察者。打印出来。
现在先介绍这几个,找个时间再整理一些其他的用法以及原理实现。
上一篇: 关于换行以及换行属性
推荐阅读
-
使用IDEA创建java项目的步骤详解(hello word)
-
关于PHPstorm 使用技巧,PHPstorm使用技巧
-
Mybatis动态sql、if与where的使用、sql片段、foreach遍历、Mybatis的关联查询一对一、一对多、多对多、Mybatis的延时加载
-
使用准则进行条件查询--1.4.从窗体中选择查询的条件
-
PHP中使用curl伪造IP的简单方法,phpcurl伪造ip
-
微服务统计,分析,图表,监控一体化的HttpReports项目在.Net Core 中的使用
-
python跨文件使用全局变量的实现
-
使用Axure RP原型设计实践02,自定义部件以及熟悉与部件相关面板_html/css_WEB-ITnose
-
Tinymce+jQuery.Validation使用产生的BUG
-
PHP使用递归生成文章树_PHP