RxJava 和 RxAndroid 二（操做符的使用）
前言：对Rx不了解的朋友能够先看个人第一篇博文 RxJava 和 RxAndroid 一 (基础)，是对Rxjava的基本介绍html
一、merge操做符，合并观察对象
19 List<String> list1 = new ArrayList<>() ; 20 List<String> list2 = new ArrayList<>() ; 21 22 list1.add( "1" ) ; 23 list1.add( "2" ) ; 24 list1.add( "3" ) ; 25 26 list2.add( "a" ) ; 27 list2.add( "b" ) ; 28 list2.add( "c" ) ; 29 30 Observable observable1 = Observable.from( list1 ) ; 31 Observable observable2 = Observable.from( list2 ) ; 32 33 //合并数据 先发送observable2的所有数据，而后发送 observable1的所有数据 34 Observable observable = Observable.merge( observable2 , observable1 ) ; 35 36 observable.subscribe(new Action1() { 37 @Override 38 public void call(Object o) { 39 System.out.println( "rx-- " + o ); 40 } 41 }) ; 42
运行结果java
二、zip 操做符，合并多个观察对象的数据。而且容许 Func2（）函数从新发送合并后的数据
List<String> list1 = new ArrayList<>() ;
List<String> list2 = new ArrayList<>() ;
list1.add( "1" ) ;
list1.add( "2" ) ;
list1.add( "3" ) ;
list2.add( "a" ) ;
list2.add( "b" ) ;
list2.add( "c" ) ;
list2.add( "d" ) ;
Observable observable1 = Observable.from( list1 ) ;
Observable observable2 = Observable.from( list2 ) ;
Observable observable3 = Observable.zip(observable1, observable2, new Func2<String , String , String >() {
@Override
public String call(String s1 , String s2 ) {
return s1 + s2 ;
}
}) ;
observable3.subscribe(new Action1() {
@Override
public void call(Object o) {
System.out.println( "zip-- " + o );
}
}) ;
运行效果：从效果图上能够看出，合并两个的观察对象数据项应该是相等的；若是出现了数据项不等的状况，合并的数据项以最小数据队列为准。android
三、scan累加器操做符的使用
Observable observable = Observable.just( 1 , 2 , 3 , 4 , 5 ) ;
observable.scan(new Func2<Integer,Integer,Integer>() {
@Override
public Integer call(Integer o, Integer o2) {
return o + o2 ;
}
})
.subscribe(new Action1() {
@Override
public void call(Object o) {
System.out.println( "scan-- " + o );
}
}) ;
运行效果：缓存
第一次发射获得1，做为结果与2相加；发射获得3，做为结果与3相加，以此类推，打印结果：安全
四、filter 过滤操做符的使用
Observable observable = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
observable.filter(new Func1<Integer , Boolean>() {
@Override
public Boolean call(Integer o) {
//数据大于4的时候才会被发送
return o > 4 ;
}
})
.subscribe(new Action1() {
@Override
public void call(Object o) {
System.out.println( "filter-- " + o );
}
}) ;
运行效果服务器
五、 消息数量过滤操做符的使用
- take ：取前n个数据
- takeLast：取后n个数据
- first 只发送第一个数据
- last 只发送最后一个数据
- skip() 跳过前n个数据发送后面的数据
- skipLast() 跳过最后n个数据，发送前面的数据
//take 发送前3个数据
Observable observable = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
observable.take( 3 )
.subscribe(new Action1() {
@Override
public void call(Object o) {
System.out.println( "take-- " + o );
}
}) ;
//takeLast 发送最后三个数据
Observable observable2 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
observable2.takeLast( 3 )
.subscribe(new Action1() {
@Override
public void call(Object o) {
System.out.println( "takeLast-- " + o );
}
}) ;
//first 只发送第一个数据
Observable observable3 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
observable3.first()
.subscribe(new Action1() {
@Override
public void call(Object o) {
System.out.println( "first-- " + o );
}
}) ;
//last 只发送最后一个数据
Observable observable4 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
observable4.last()
.subscribe(new Action1() {
@Override
public void call(Object o) {
System.out.println( "last-- " + o );
}
}) ;
//skip() 跳过前2个数据发送后面的数据
Observable observable5 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
observable5.skip( 2 )
.subscribe(new Action1() {
@Override
public void call(Object o) {
System.out.println( "skip-- " + o );
}
}) ;
//skipLast() 跳过最后两个数据，发送前面的数据
Observable observable6 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
observable5.skipLast( 2 )
.subscribe(new Action1() {
@Override
public void call(Object o) {
System.out.println( "skipLast-- " + o );
}
}) ;
效果图
六、elementAt 、elementAtOrDefault
//elementAt() 发送数据序列中第n个数据 ，序列号从0开始
//若是该序号大于数据序列中的最大序列号，则会抛出异常，程序崩溃
//因此在用elementAt操做符的时候，要注意判断发送的数据序列号是否越界
Observable observable7 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
observable7.elementAt( 3 )
.subscribe(new Action1() {
@Override
public void call(Object o) {
System.out.println( "elementAt-- " + o );
}
}) ;
//elementAtOrDefault( int n , Object default ) 发送数据序列中第n个数据 ，序列号从0开始。
//若是序列中没有该序列号，则发送默认值
Observable observable9 = Observable.just( 1 , 2 , 3 , 4 , 5 ) ;
observable9.elementAtOrDefault( 8 , 666 )
.subscribe(new Action1() {
@Override
public void call(Object o) {
System.out.println( "elementAtOrDefault-- " + o );
}
}) ;
运行结果网络
七、startWith() 插入数据
//插入普通数据
//startWith 数据序列的开头插入一条指定的项 , 最多插入9条数据
Observable observable = Observable.just( "aa" , "bb" , "cc" ) ;
observable
.startWith( "11" , "22" )
.subscribe(new Action1() {
@Override
public void call(Object o) {
System.out.println( "startWith-- " + o );
}
}) ;
//插入Observable对象
List<String> list = new ArrayList<>() ;
list.add( "ww" ) ;
list.add( "tt" ) ;
observable.startWith( Observable.from( list ))
.subscribe(new Action1() {
@Override
public void call(Object o) {
System.out.println( "startWith2 -- " + o );
}
}) ;
运行结果app
八、delay操做符，延迟数据发送
Observable<String> observable = Observable.just( "1" , "2" , "3" , "4" , "5" , "6" , "7" , "8" ) ;
//延迟数据发射的时间，仅仅延时一次，也就是发射第一个数据前延时。发射后面的数据不延时
observable.delay( 3 , TimeUnit.SECONDS ) //延迟3秒钟
.subscribe(new Action1() {
@Override
public void call(Object o) {
System.out.println("delay-- " + o);
}
}) ;
九、Timer 延时操做符的使用
使用场景：xx秒后，执行xx ide
//5秒后输出 hello world , 而后显示一张图片
Observable.timer( 5 , TimeUnit.SECONDS )
.observeOn(AndroidSchedulers.mainThread() )
.subscribe(new Action1<Long>() {
@Override
public void call(Long aLong) {
System.out.println( "timer--hello world " + aLong );
findViewById( R.id.image).setVisibility(View.VISIBLE );
}
}) ;
timer 返回一个 Observable , 它在延迟一段给定的时间后发射一个简单的数字0函数
timer 操做符默认在computation调度器上执行，固然也能够用 Scheduler 在定义执行的线程。
delay 、timer 总结：
- 相同点：delay 、 timer 都是延时操做符。
- 不一样点：delay 延时一次，延时完成后，能够连续发射多个数据。timer延时一次，延时完成后，只发射一次数据。
十、interval 轮询操做符，循环发送数据，数据从0开始递增
package app.com.myapplication;
import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import java.util.concurrent.TimeUnit;
import rx.Observable;
import rx.Subscription;
import rx.functions.Action1;
public class IntervalActivity extends AppCompatActivity {
Subscription subscription ;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_interval);
//参数一：延迟时间 参数二：间隔时间 参数三：时间颗粒度
Observable observable = Observable.interval(3000, 3000, TimeUnit.MILLISECONDS) ;
subscription = observable.subscribe(new Action1() {
@Override
public void call(Object o) {
System.out.println( "interval- " + o );
}
}) ;
}
@Override
protected void onDestroy() {
super.onDestroy();
if ( subscription != null ){
subscription.unsubscribe();
}
}
}
十一、doOnNext() 操做符，在每次 OnNext() 方法被调用前执行
使用场景：从网络请求数据，在数据被展现前，缓存到本地
Observable observable = Observable.just( "1" , "2" , "3" , "4" ) ;
observable.doOnNext(new Action1() { @Override public void call(Object o) { System.out.println( "doOnNext--缓存数据" + o ); } }) .subscribe(new Observer() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(Object o) { System.out.println( "onNext--" + o ); } }) ;
十二、Buffer 操做符
- Buffer( int n ) 把n个数据打成一个list包，而后再次发送。
- Buffer( int n , int skip) 把n个数据打成一个list包，而后跳过第skip个数据。
使用场景：一个按钮每点击3次，弹出一个toast
List<String> list = new ArrayList<>();
for (int i = 1; i < 10; i++) {
list.add("" + i);
}
Observable<String> observable = Observable.from(list);
observable
.buffer(2) //把每两个数据为一组打成一个包，而后发送
.subscribe(new Action1<List<String>>() {
@Override
public void call(List<String> strings) {
System.out.println( "buffer---------------" );
Observable.from( strings ).subscribe(new Action1<String>() {
@Override
public void call(String s) {
System.out.println( "buffer data --" + s );
}
}) ;
}
});
例子2：
//第一、2 个数据打成一个数据包，跳过第三个数据 ； 第四、5个数据打成一个包，跳过第6个数据
observable.buffer( 2 , 3 ) //把每两个数据为一组打成一个包，而后发送。第三个数据跳过去
.subscribe(new Action1<List<String>>() {
@Override
public void call(List<String> strings) {
System.out.println( "buffer22---------------" );
Observable.from( strings ).subscribe(new Action1<String>() {
@Override
public void call(String s) {
System.out.println( "buffer22 data --" + s );
}
}) ;
}
}) ;
1三、throttleFirst 操做符
在一段时间内，只取第一个事件，而后其余事件都丢弃。
使用场景：一、button按钮防抖操做，防连续点击 二、百度关键词联想，在一段时间内只联想一次，防止频繁请求服务器
Observable.interval( 1 , TimeUnit.SECONDS) .throttleFirst( 3 , TimeUnit.SECONDS ) .subscribe(new Action1<Long>() { @Override public void call(Long aLong) { System.out.println( "throttleFirst--" + aLong ); } }) ;
这段代码，是循环发送数据，每秒发送一个。throttleFirst( 3 , TimeUnit.SECONDS ) 在3秒内只取第一个事件，其余的事件丢弃。
运行结果
1四、distinct 过滤重复的数据
List<String> list = new ArrayList<>() ; list.add( "1" ) ; list.add( "2" ) ; list.add( "1" ) ; list.add( "3" ) ; list.add( "4" ) ; list.add( "2" ) ; list.add( "1" ) ; list.add( "1" ) ; Observable.from( list ) .distinct() .subscribe(new Action1<String>() { @Override public void call(String s) { System.out.println( "distinct--" + s ); } }) ;
从结果能够看出，重复的数据已经被过滤掉了
distinctUntilChanged() 过滤连续重复的数据
List<String> list = new ArrayList<>() ; list.add( "1" ) ; list.add( "2" ) ; list.add( "1" ) ; list.add( "3" ) ; list.add( "4" ) ; list.add( "4" ) ; list.add( "2" ) ; list.add( "1" ) ; list.add( "1" ) ; Observable.from( list ) .distinctUntilChanged() .subscribe(new Action1<String>() { @Override public void call(String s) { System.out.println( "distinctUntilChanged--" + s ); } }) ;
运行结果
从结果能够看出，连续重复的数据已经被过滤掉了
1五、debounce() 操做符
一段时间内没有变化，就会发送一个数据。
使用场景：百度关键词联想提示。在输入的过程当中是不会从服务器拉数据的。当输入结束后，在400毫秒没有输入就会去获取数据。
避免了，屡次请求给服务器带来的压力.
1六、doOnSubscribe()
使用场景： 能够在事件发出以前作一些初始化的工做，好比弹出进度条等等
注意：
一、doOnSubscribe() 默认运行在事件产生的线程里面，然而事件产生的线程通常都会运行在 io 线程里。那么这个时候作一些，更新UI的操做，是线程不安全的。
因此若是事件产生的线程是io线程，可是咱们又要在doOnSubscribe() 更新UI ， 这时候就须要线程切换。
二、若是在 doOnSubscribe() 以后有 subscribeOn() 的话，它将执行在离它最近的 subscribeOn() 所指定的线程。
三、 subscribeOn() 事件产生的线程 ； observeOn() : 事件消费的线程
Observable.create(onSubscribe) .subscribeOn(Schedulers.io()) .doOnSubscribe(new Action0() { @Override public void call() { progressBar.setVisibility(View.VISIBLE); // 须要在主线程执行 } }) .subscribeOn(AndroidSchedulers.mainThread()) // 指定主线程 .observeOn(AndroidSchedulers.mainThread()) .subscribe(subscriber);
1七、range 操做符的使用
首先看range 方法的源码
public static Observable<Integer> range(int start, int count) {
if (count < 0) {
throw new IllegalArgumentException("Count can not be negative");
}
if (count == 0) {
return Observable.empty();
}
if (start > Integer.MAX_VALUE - count + 1) {
throw new IllegalArgumentException("start + count can not exceed Integer.MAX_VALUE");
}
if(count == 1) {
return Observable.just(start);
}
return Observable.create(new OnSubscribeRange(start, start + (count - 1)));
}
//能够经过第三个参数控制range执行的线程
public static Observable<Integer> range(int start, int count, Scheduler scheduler) {
return range(start, count).subscribeOn(scheduler);
}
Range操做符发射一个范围内的有序整数序列，你能够指定范围的起始和长度。
RxJava将这个操做符实现为range函数，它接受两个参数，一个是范围的起始值，一个是范围的数据的数目。若是你将第二个参数设为0，将致使Observable不发射任何数据（若是设置为负数，会抛异常）。
range默认不在任何特定的调度器上执行。有一个变体能够经过可选参数指定Scheduler。
例子
Observable.range( 10 , 3 )
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.v( "rx_range " , "" + integer ) ;
}
}) ;
结果
/rx_range: 10
/rx_range: 11
/rx_range: 12
1八、defer 操做符
例子
package app.com.myapplication;
import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.util.Log;
import rx.Observable;
import rx.functions.Action1;
import rx.functions.Func0;
public class DeferActivity extends AppCompatActivity {
String i = "10" ;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_defer);
i = "11 " ;
Observable<String> defer = Observable.defer(new Func0<Observable<String>>() {
@Override
public Observable<String> call() {
return Observable.just( i ) ;
}
}) ;
Observable test = Observable.just( i ) ;
i = "12" ;
defer.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.v( "rx_defer " , "" + s ) ;
}
}) ;
test.subscribe(new Action1() {
@Override
public void call(Object o) {
Log.v( "rx_just " , "" + o ) ;
}
}) ;
}
}
结果
/rx_defer: 12
/rx_just: 11
- 能够看到，just操做符是在建立Observable就进行了赋值操做，而defer是在订阅者订阅时才建立Observable，此时才进行真正的赋值操做。
Defer操做符会一直等待直到有观察者订阅它，而后它使用Observable工厂方法生成一个Observable。它对每一个观察者都这样作，所以尽管每一个订阅者都觉得本身订阅的是同一个Observable，事实上每一个订阅者获取的是它们本身的单独的数据序列。- 在某些状况下，等待直到最后一分钟（就是知道订阅发生时）才生成Observable能够确保Observable包含最新的数据。
RxJava 和 RxAndroid 三（生命周期控制和内存优化）
相关文章
- 1. RxJava 和 RxAndroid 二（操作符的使用）
- 2. RxJava和RxAndroid
- 3. RxJava操做符（二）----转换操做符
- 4. RxJava操做符使用
- 5. RxJava操做符
- 6. RxJava 和 RxAndroid 四（RxBinding的使用）
- 7. Rxjava学习(二操做符)
- 8. RxJava 操做符系列二
- 9. 初识RXJava 和 RxAndroid
- 10. Rxjava(三)、操做符的使用
- 更多相关文章...
- • SQL BETWEEN 操作符 - SQL 教程
- • SQL UNION 操作符 - SQL 教程
- • RxJava操作符（二）Transforming Observables
- • RxJava操作符（一）Creating Observables
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. vs2019运行opencv图片显示代码时，窗口乱码
- 2. app自动化 - 元素定位不到？别慌，看完你就能解决
- 3. 在Win8下用cisco ××× Client连接时报Reason 422错误的解决方法
- 4. eclipse快速补全代码
- 5. Eclipse中Java/Html/Css/Jsp/JavaScript等代码的格式化
- 6. idea+spring boot +mabitys(wanglezapin)+mysql (1)
- 7. 勒索病毒发生变种 新文件名将带有“.UIWIX”后缀
- 8. 【原创】Python 源文件编码解读
- 9. iOS9企业部署分发问题深入了解与解决
- 10. 安装pytorch报错CondaHTTPError：******
欢迎关注本站公众号,获取更多信息
