干货! RxJava 常见使用场景及小技巧!

总结Rxjava最常用的使用场景和小技巧

本文基于RxJava1.x.x ,若用 2.x版本请自行替换部分代码

基础部分:给 Android 开发者的 RxJava 详解 ——扔物线
对RxJava的概念以及基本特性做了详细介绍

本系列相关进阶:

带你入门学习Rxjava ;RxJava2.x使用以及操作符详解

正文部分:

RxJava+RxBinding在View上的一些使用技巧

//防抖操作防止多次点击

注"依赖RxView

RxView.clicks(btn_click)

.throttleFirst(3, TimeUnit.SECONDS)

.subscribe(new Action1<Void>() {

@Override

public void call(Void aVoid) {

Toast.makeText(getActivity(), R.string.des_demo_not_more_click, Toast.LENGTH_SHORT).show();

}

});

[/code]

//CheckBox状态更新-响应式界面

RxCompoundButton.checkedChanges(checkBox2)

.subscribe(new Action1<Boolean>() {

@Override

public void call(Boolean aBoolean) {

//dosome...

}

});

[/code]

//使用debounce做textSearch 如:可以使用debounce减少频繁的网络请求。避免每输入（删除）一个字就做一次联想

RxTextView.textChangeEvents(inputEditText)

.debounce(400, TimeUnit.MILLISECONDS)

.observeOn(AndroidSchedulers.mainThread())

.subscribe(new Observer<TextViewTextChangeEvent>() ...

[/code]

RxJava轮询器：interval

////延时3000 ，每间隔3000

private Subscription subscribe_auto;

//开启轮询

private void autoLoop() {

if (subscribe_auto == null || subscribe_auto.isUnsubscribed()) {

subscribe_auto = Observable.interval(3000, 3000, TimeUnit.MILLISECONDS)

//延时3000 ，每间隔3000，时间单位

.observeOn(AndroidSchedulers.mainThread())

.subscribe(new Action1<Long>() {

@Override

public void call(Long aLong) {

//do...

}

});

}

}

//关闭询器

void stopLoop() {

if (subscribe_auto != null && !subscribe_auto.isUnsubscribed()) {

subscribe_auto.unsubscribe();

}

[/code]

RxJava定时器：timer

Observable.timer(3000 , TimeUnit.MILLISECONDS)

.observeOn(AndroidSchedulers.mainThread())

.subscribe(new Action1<Long>() {

@Override

public void call(Long aLong) {

iv_welcome.setVisibility(View.VISIBLE);

//do...

}

});

[/code]

RxJava操作符：Buffer

//点击按钮达到指定次数后，触发某事件

RxView.clicks(btn_buffer_count)

.buffer(3)

.compose(this.<List<Void>>bindToLifecycle())

.observeOn(AndroidSchedulers.mainThread())

.subscribe(new Action1<List<Void>>() {

@Override

public void call(List<Void> voids) {

Toast.makeText(BufferFragment.this.getActivity(), R.string.des_demo_buffer_count, Toast.LENGTH_SHORT).show();

}

});

[/code]

//案例：输入123456 ， 两两发送，并跳过之后的一个数据，即最后订阅者接收到[1,2] [4,5] 两个数据包裹， 3和6 分别被跳过

Observable.from(chs)

.buffer(2, 3)

.observeOn(AndroidSchedulers.mainThread())

.subscribe(new Action1<List<Character>>() {

@Override

public void call(List<Character> characters) {

//do....

}

});

[/code]

R

R

RxJava combineLatest合并最近N个结点

//例如：注册的时候所有输入信息（邮箱、密码、电话号码等）合法才点亮注册按钮

Observable<CharSequence> _emailChangeObservable = RxTextView.textChanges(_email).skip(1);

Observable<CharSequence> _passwordChangeObservable = RxTextView.textChanges(_password).skip(1);

Observable<CharSequence> _numberChangeObservable = RxTextView.textChanges(_number).skip(1);

Observable.combineLatest(_emailChangeObservable,

_passwordChangeObservable,

_numberChangeObservable,

new Func3<CharSequence, CharSequence, CharSequence, Boolean>() {

@Override

public Boolean call(CharSequence newEmail,

CharSequence newPassword,

CharSequence newNumber) {

Log.d("xiayong",newEmail+" "+newPassword+" "+newNumber);

boolean emailValid = !isEmpty(newEmail) &&

 EMAIL_ADDRESS.matcher(newEmail).matches();

if (!emailValid) {

_email.setError("Invalid Email!");

}

boolean passValid = !isEmpty(newPassword) && newPassword.length() > 8;

if (!passValid) {

_password.setError("Invalid Password!");

}

boolean numValid = !isEmpty(newNumber);

if (numValid) {

int num = Integer.parseInt(newNumber.toString());

numValid = num > 0 && num <= 100;

}

if (!numValid) {

_number.setError("Invalid Number!");

}

return emailValid && passValid && numValid;

}

})//

.subscribe(new Observer<Boolean>() {

@Override

public void onCompleted() {

log.d("completed");

}

@Override

public void onError(Throwable e) {

 log.d("Error");

}

@Override

public void onNext(Boolean formValid) {

 _btnValidIndicator.setEnabled(formValid);

}

}); 

[/code]

RxJava操作符：zip数据合并操作

//合并联系人模拟一部分数据来自本地，一部分来自网络

 /**

 * 模拟网络联系人列表

 */

private Observable<List<Contacter>> queryContactsForNet() {

return Observable.create(new Observable.OnSubscribe<List<Contacter>>() {

@Override

public void call(Subscriber<? super List<Contacter>> subscriber) {

try {

Thread.sleep(3000);

} catch (InterruptedException e) {

e.printStackTrace();

}

ArrayList<Contacter> contacters = new ArrayList<>();

contacters.add(new Contacter("net:Zeus"));

contacters.add(new Contacter("net:Athena"));

contacters.add(new Contacter("net:Prometheus"));

subscriber.onNext(contacters);

subscriber.onCompleted();

}

});

}

/**

 * 模拟手机本地联系人查询

 */

private Observable<List<Contacter>> queryContactsFromLocation() {

return Observable.create(new Observable.OnSubscribe<List<Contacter>>() {

@Override

public void call(Subscriber<? super List<Contacter>> subscriber) {

ArrayList<Contacter> contacters = new ArrayList<>();

contacters.add(new Contacter("location:张三"));

contacters.add(new Contacter("location:李四"));

contacters.add(new Contacter("location:王五"));

subscriber.onNext(contacters);

subscriber.onCompleted();

}

});

}

Observable.zip(

queryContactsFromLocation(),

queryContactsForNet(),

new Func2<List<Contacter>, List<Contacter>, List<Contacter>>() {

@Override

public List<Contacter> call(List<Contacter> contacters, List<Contacter> contacters2) {

contacters.addAll(contacters2);

return contacters;

}

}

)

.subscribeOn(Schedulers.io())

.observeOn(AndroidSchedulers.mainThread())

.subscribe(new Action1<List<Contacter>>() {

@Override

public void call(List<Contacter> contacters) {

initPage(contacters);

}

});

[/code]

RxJava操作符：merge合并操作符

//模拟网络请求中，1秒后先显示本地数据，3秒后显示网络数据

 private Observable<List<Contacter>> getDataFromNet() {

return Observable.create(new Observable.OnSubscribe<List<Contacter>>() {

@Override

public void call(Subscriber<? super List<Contacter>> subscriber) {

try {

Thread.sleep(3000);

} catch (InterruptedException e) {

e.printStackTrace();

}

ArrayList<Contacter> contacters = new ArrayList<>();

contacters.add(new Contacter(NET + "Zeus"));

contacters.add(new Contacter(NET + "Athena"));

contacters.add(new Contacter(NET + "Prometheus"));

 // subscriber.onError(new Throwable("模拟出错"));

subscriber.onNext(contacters);

subscriber.onCompleted();

}

});

}

private Observable<List<Contacter>> getDataFromLocation() {

return Observable.create(new Observable.OnSubscribe<List<Contacter>>() {

@Override

public void call(Subscriber<? super List<Contacter>> subscriber) {

try {

Thread.sleep(1000);

} catch (InterruptedException e) {

e.printStackTrace();

}

List<Contacter> datas = new ArrayList<>();

datas.add(new Contacter(LOCATION + "张三"));

datas.add(new Contacter(LOCATION + "李四"));

datas.add(new Contacter(LOCATION + "王五"));

subscriber.onNext(datas);

subscriber.onCompleted();

//合并请求

Zip，还望大家不要跟Merge操作符混淆，Merge/concat其实只是将多个Obsevables的输出序列变为一个，方便订阅者统一处理，对于订阅者来说就仿佛只订阅了一个观察者一样。

Observable.concat(

getDataFromLocation(),

getDataFromNet()

)

.subscribeOn(Schedulers.io())

.observeOn(AndroidSchedulers.mainThread())

.subscribe(new Action1<List<Contacter>>() {

@Override

public void call(List<Contacter> contacters) {

initPage(contacters);

}

}, new Action1<Throwable>() {

@Override

public void call(Throwable throwable) {

XgoLog.e(throwable.getMessage());

}

}, new Action0() {

@Override

public void call() {

XgoLog.d("onCompleted()");

}

});

[/code]