RxJava

RxJava By 侦跃 & @hi 头 hi

RxJava 扩展的观察者模式 处

观察者模式 Observable 发出事件 Subscriber 订阅事件

bus.post(new AnswerEvent(42)); @Subscribe public void onanswer(answerevent event) {! }!

Observable observable = Observable.create(new Observable.OnSubscribe<String>() { @Override public void call(subscriber<? super String> subscriber) { subscriber.onnext("a"); subscriber.oncompleted(); } }); observable.subscribe(new Subscriber() { });

观察者模式的不 足 时结 错误

RxJava 进 oncomplete Subscriber 结 onerror Subscriber 错

Observable.just("hello").subscribe(new Subscriber<String>() { @Override public void oncompleted() { }! @Override public void onerror(throwable e) { }! @Override public void onnext(string s) { System.out.println(s); } });

unsubscribe Subscription subscription = observable.subscribe(system.out::println); subscription.unsubscribe();

处 观 产 线 订阅 线 处

Observable.just("hello") // 观 线 产.subscribeOn(Schedulers.newThread()) // 观 UI 线.observeOn(AndroidSchedulers.mainThread()) // 为 UI 线程接收数据, // 这 UI.subscribe(System.out::println);

Scheduler 认 RxJava 产 订阅 线 运! Scheduler 换 产 订阅 执 线! RxJava 认 Scheduler 针对 务选择

Schedulers Schedulers.immediate() Schedulers.trampoline() Schedulers.newThread() Schedulers.computation() Schedulers.io() AndroidSchedulers.mainThread()

处 -Operator 对 进 变换处 operators 义 operator

Observable.just(1, 2, 3).map(new Func1<Integer, String>() { @Override public String call(integer integer) { return integer.tostring(); } }).subscribe(new Action1<String>() { @Override public void call(string s) { System.out.println(s); } }); Observable.just(1, 2, 3).map(integer -> integer.tostring()).subscribe(system.out::println);

务 组 组 转换 过滤掉 小于 1

Observable.just("1", "2", "2", "3", "4", "5").map(Integer::parseInt).filter(s -> s > 1).distinct().take(3).reduce((integer, integer2) -> integer.intvalue() + integer2.intvalue( ).subscribe(system.out::println);//9

operators 创建 Observable create just 变换 Observable map flatmap 过滤 Observable filter first last Observable merge zip 错误处理 catch 过滤 all skiputil takewhile average reduce count

义 operator Observable.just( a ).lift(subscriber -> { return new Subscriber<String>() { @Override public void oncompleted() { subscriber.oncompleted(); } @Override public void onerror(throwable e) { subscriber.onerror(e); } @Override public void onnext(string s) { subscriber.onnext(1); } }; }).map(i -> i).subscribe(system.out::println);//1

Android 应 Retrofit RxAndroid RxBinding rx-preferences sqlbrite RxLifecycle

Retrofit @GET("/story/{id}") Observable<NewsDetail> getnewsdetailobservable(@path("id") long id);

RxAndroid AndroidSchedulers.mainThread()

RxBinding RxJava 对 Android UI 进

RxView.clicks(findViewById(R.id.btn_throttle)).throttleFirst(1, TimeUnit.SECONDS).subscribe(aVoid -> { System.out.println("click"); });

RxPreference RxJava SharePreference

SharedPreferences preferences = PreferenceManager.getDefaultSharedPreferences(context);! RxSharedPreferences rxpreferences = RxSharedPreferences.create(preferences);! Preference<String> username = rxpreferences.getstring( username");! username.asobservable().subscribe(new Action1<String>() { @Override public void call(string username) { Log.d(TAG, "Username: " + username); } }

RxBinding 结合 RxPreference SharedPreferences preferences = PreferenceManager.getDefaultSharedPreferences(this); RxSharedPreferences rxpreferences = RxSharedPreferences.create(preferences); Preference<Boolean> checked = rxpreferences.getboolean("checked", true); CheckBox checkbox = (CheckBox) findviewbyid(r.id.cb_test); RxCompoundButton.checkedChanges(checkBox).subscribe(checked.asAction()); checked.asobservable().subscribe(aboolean -> { System.out.println("----------------checked: " + aboolean); });

RxLifecycle Activity Fragment 结

myobservable.compose(rxlifecycle.binduntilactivityevent(lifecycle, ActivityEvent.DESTROY)).subscribe(); myobservable.compose(rxlifecycle.bindactivity(lifecycle)).subscribe();

RxJava 场 现 层 调 (Callback hell) 杂 处 应 UI 杂 线 换

层 调 gettoken(new Callback<String>() { @Override public void success(string token) { getuser(userid, new Callback<User>() { @Override public void success(user user) { userview.setuser(user); }! @Override public void failure(retrofiterror error) { // Error handling } }; }! @Override public void failure(retrofiterror error) { // Error handling } });

gettoken("username", "password").flatmap(token -> getuser(token)).subscribe(user -> { System.out.println("user: " + user.tojson()); });

杂 处 Observable.just("1", "2", "2", "3", "4", "5").map(Integer::parseInt).filter(s -> s > 1).distinct().take(3).reduce((integer, integer2) -> integer.intvalue() + integer2.intvalue()).subscribe(system.out::println);//9

应 UI CheckBox checkbox = (CheckBox) findviewbyid(r.id.cb_test); RxCompoundButton.checkedChanges(checkBox).subscribe(checked.asAction());

杂 线 换 Observable<String> observable1 = createobservable1().subscribeon(schedulers.newthread()); Observable<String> observable2 = createobservable2().subscribeon(schedulers.io()); Observable.concat(observable1, observable2).subscribeon(schedulers.computation()).subscribe(system.out::println);

RxJava 库体积稍 大 (900K 对 时 unsubscr RxBinding 库 ) Java6 lambda 简化代码 (Retrolambda & Gr

http://reactivex.io/ 扔物线的 : 给 Android 发 RxJava 详解 http://gank.io/post/ 560e15be2dca930e00da1083 RxJava http://blog.csdn.net/ lzyzsd/article/details/41833541

应 用案例 Square SoundCloud Flipboard 动

Thanks

Q&A