【Android】Rxjava操做符系列（二）
今每天气不错啊，继续撸操做符！java
1.debounce
做用：debounce操做符是对源Observable间隔期产生的结果进行过滤，若是在这个规定的间隔期内没有别的结果产生，则将这个结果提交给订阅者，不然忽略该结果，原理有点像光学防抖.react
Subscription sub = Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super Integer> subscriber) {
if (subscriber.isUnsubscribed()) return;
try {
for (int i = 0; i < 6; i++) {
Log.e("rxjava","i = "+i);
subscriber.onNext(i);
try {
Thread.currentThread().sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
subscriber.onCompleted();
} catch (Exception e) {
subscriber.onError(e);
}
}
}).subscribeOn(Schedulers.newThread())
.debounce(2000, TimeUnit.MILLISECONDS)
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.e("rxjava", "debounce = " + integer);
}
});
若是2秒内没有发出新的消息，则将这个结果提交给订阅者，运行效果：web
2.distinct
做用：去重app
Observable.just(1, 2, 3, 1, 2, 3, 12, 3).distinct().subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.e("rxjava", "distinct = " + integer);
}
});
结果：ide
3.elementAt
做用：获取指定索引的结果svg
Observable.range(1, 10).elementAt(4).subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Integer integer) {
Log.e("rxjava", "elementAt = " + integer);
}
});
结果：函数
4.filter
做用：根据条件筛选结果gradle
Observable.range(1, 10).filter(new Func1<Integer, Boolean>() {
@Override
public Boolean call(Integer integer) {
return integer % 2 == 0;
}
}).subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Integer integer) {
Log.e("rxjava", "filter = " + integer);
}
});
结果：ui
5.first
做用：只发射第一个数据spa
Observable.just(1, 2,3,4,5).first().subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.e("rxjava", "first = " + integer);
}
});
结果：
6.ignoreElements
做用：ignoreElements操做符忽略全部源Observable产生的结果，只会执行onCpmpleted()或者onError()方法.
Observable.range(1, 10).ignoreElements().filter(new Func1<Integer, Boolean>() {
@Override
public Boolean call(Integer integer) {
return integer % 2 == 0;
}
}).subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
tvShow.setText("onCompleted()");
Log.e("rxjava", "ignoreElement = onCompleted()");
}
@Override
public void onError(Throwable e) {
tvShow.setText("onError---"+e.getMessage());
Log.e("rxjava", "ignoreElement = onError()");
}
@Override
public void onNext(Integer integer) {
sb.append(integer.toString() + "---");
tvShow.setText(sb.toString());
Log.e("rxjava", "ignoreElement = " + integer);
}
});
结果：
7.last
做用：只发射最后一项
Observable.just(1, 2,3,4,5).last().subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.e("rxjava", "last = "+ integer);
}
});
结果：
8.sample
做用：sample操做符是按期扫描源Observable产生的结果，在指定的间隔周期内进行采样
Observable.range(1,3000000).sample(200,TimeUnit.MILLISECONDS).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.e("rxjava", "simple = "+ integer);
}
});
结果：
9.skip
做用：抑制Observable发射的前N项数据
Observable.just(1, 2,3,4,5).skip(2).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.e("rxjava", "simple = "+ integer);
}
});
结果：
10.skiplast
做用：抑制Observable发射的后N项数据
Observable.just(1, 2, 3, 4, 5).skipLast(2).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.e("rxjava", "skiplast = " + integer);
}
});
结果：
11.take
做用：只发射前面的N项数据
Observable.just(1, 2, 3, 4, 5).take(2).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.e("rxjava", "take = " + integer);
}
});
结果：
12.takelast
做用：发射Observable发射的最后N项数据
Observable.just(1, 2, 3, 4, 5).takeLast(2).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.e("rxjava", "takelast = " + integer);
}
});
结果：
13.And,Then和When
做用：And,Then和When操做符是在RxJava的joins包下，使用Pattern和Plan做为中介，将发射的数据集合并到一块儿。须要导包joins包，AndroidStudio中在gradle引用：compile ‘io.reactivex:rxjava-joins:0.22.0’，若是不引用此包，类Pattern、Plan和JoinObservable不存在.
Observable observableApp = Observable.from(new String[]{"小猫","小狗","小猪"});
Observable<Long> tictoc = Observable.interval(1, TimeUnit.SECONDS);
Pattern2 pattern = JoinObservable.from(observableApp).and(tictoc);
Plan0 plan = pattern.then(new Func2() {
@Override
public Object call(Object o, Object o2) {
return o+":"+o2;
}
});
JoinObservable
.when(plan)
.toObservable()
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Object o) {
Log.e("rxjava", "and/then/when = " + o);
}
});
结果：
14.CombineLatest
做用：当多个Observables中的任何一个发射了数据时，使用一个函数结合每一个Observable发射的最近数据项，而且基于这个函数的结果发射数据。
CombineLatest在原始的Observable中任意一个发射了数据时发射一条数据。当原始Observables的任何一个发射了一条数据时，CombineLatest使用一个函数结合它们最近发射的数据，而后发射这个函数的返回值。
Observable<String> o1 = Observable.create(new Observable.OnSubscribe<String>() {
@Override public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("o1");
subscriber.onNext("o2");
subscriber.onNext("o3");
}
});
Observable<String> o2 = Observable.create(new Observable.OnSubscribe<String>() {
@Override public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("o4");
subscriber.onNext("o5");
subscriber.onNext("o6");
}
});
Observable<String> o3 = Observable.create(new Observable.OnSubscribe<String>() {
@Override public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("o7");
subscriber.onNext("o8");
subscriber.onNext("o9");
}
});
Observable.combineLatest(o1, o2, o3, new Func3<String, String,String, String>() {
@Override public String call(String s, String s2,String s3) {
Log.e("combine --- >", "s = " + s + " | s2 = " + s2 +"| s3 = "+ s3 );
return s + s2 +s3;
}
}).subscribe(new Observer<String>() {
@Override public void onCompleted() {
Log.e("onCompleted --- >", "onCompleted");
}
@Override public void onError(Throwable e) {
Log.e("onError --- >", "onError");
}
@Override public void onNext(String o) {
Log.e("onNext --- >", o);
}
});
结果：
15.join
做用：两个数据元素，第二个数据源B，每发射一个数据，咱们都把它和第一个数据源A中已经发射的数据进行一对一匹配
Observable<Integer> obs1 = Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super Integer> subscriber) {
for (int i = 1; i < 7; i++) {
subscriber.onNext(i);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
});
Observable.from(new String[]{"zhangsan +","lisi +"})
.join(obs1,
//接受从源Observable发射来的数据，并返回一个Observable，
//这个Observable的生命周期决定了源Observable发射出来数据的有效期
new Func1<String, Observable<Long>>() {
@Override
public Observable<Long> call(String s) {
return Observable.timer(4000, TimeUnit.MILLISECONDS);
}
},
//接受从目标Observable发射来的数据，并返回一个Observable，
//这个Observable的生命周期决定了目标Observable发射出来数据的有效期
new Func1<Integer, Observable<Long>>() {
@Override
public Observable<Long> call(Integer integer) {
return Observable.timer(2000, TimeUnit.MILLISECONDS);
}
},
//接收从源Observable和目标Observable发射来的数据，并返回最终组合完的数据
new Func2<String,Integer,String>() {
@Override
public String call(String str1, Integer integer) {
return str1 + integer;
}
})
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<String>() {
@Override
public void call(String o) {
Log.e("rxjava","join:"+o);
}
});
结果：
16.merge
做用：合并数据源
Observable<Integer> observableOne = Observable.range(1,5);
Observable<Integer> observableTwo = Observable.range(6,4);
Subscription subscription = Observable.merge(observableOne,observableTwo)
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<Integer>() {
@Override
public void onNext(Integer integer) {
Log.e("rxjava", "merge = " + integer);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onCompleted() {
}
});
结果：
17.startwith
做用：startWith操做符经过传递一个参数来先发射一个数据序列
Integer[] value = new Integer[]{1,2,3,4,5,6,7,8,9};
Subscription sub = Observable.from(value)
.startWith(value[6])
.take(5)
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Integer integer) {
Log.e("rxjava", "startwith = " + integer);
}
});
结果：
18.zip
做用：zip操做符返回一个Obversable，它使用这个函数按顺序结合两个或多个Observables发射 的数据项，而后它发射这个函数返回的结果
Observable<Integer> observableOne = Observable.from(new Integer[]{1,2,3,4,5});
Observable<String> observableTwo = Observable.from(new String[]{"张三","李四","王五","赵六","林七"});
Observable<ZipObject> observable = Observable.zip(observableOne, observableTwo, new Func2<Integer, String, ZipObject>() {
@Override
public ZipObject call(Integer integer, String s) {
ZipObject zipObject = new ZipObject();
zipObject.name = s;
zipObject.number = integer;
return zipObject;
}
});
Subscription subscription = observable.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<ZipObject>() {
@Override
public void call(ZipObject zipObject) {
Log.e("rxjava", "zip = " + zipObject.name+"---"+zipObject.number);
}
});
结果：
欲知后事如何，请看下回分解！