核心思想
- 观察者模式
- RxJava最核心的两个东西是Observables（被观察者，事件源）和Subscribers（观察者）。Observables发出一系列事件，Subscribers处理这些事件。
- 一个Observable可以发出零个或者多个事件，直到结束或者出错。每发出一个事件，就会调用它的Subscriber的onNext方法，最后调用Subscriber.onNext()或者Subscriber.onError()结束。
- 如果一个Observerble没有任何的的Subscriber，那么这个Observable是不会发出任何事件的。
异步
链式调用
基本使用
创建与订阅
- 创建观察者（事件处理）和被观察者（事件源）
- 订阅
// 创建被观察者
Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("Hello, world!");
subscriber.onCompleted();
}
});
// 创建观察者
Subscriber<String> subscriber = new Subscriber<String>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
System.out.println(s);
showMsg(s);
}
};
// 订阅事件
observable.subscribe(subscriber);
简洁模式：
Observable<String> observable1 = Observable.just("你好！");
Action1<String> onNextAction = new Action1<String>() {
@Override
public void call(String s) {
System.out.println(s);
showMsg(s);
}
};
observable1.subscribe(onNextAction);
变换
map()
它将一种数据类型变换为另一种数据类型
同数据类型：
// 同数据类型
Observable.just("Hello, world!")
.map(new Func1<String, String>() {
@Override
public String call(String s) {
return s + " -fengmlo";
}
})
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
System.out.println(s);
showMsg(s);
}
});
不同数据类型：
// 不同数据类型
Observable.just("你好！")
.map(new Func1<String, Integer>() {
@Override
public Integer call(String s) {
return s.hashCode();
}
})
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
System.out.println(integer);
showMsg(integer + "");
}
});
多次变换：
// 多次变换
Observable.just("你好！")
.map(new Func1<String, Integer>() {
@Override
public Integer call(String s) {
return s.hashCode();
}
})
.map(new Func1<Integer, String>() {
@Override
public String call(Integer integer) {
return Integer.toString(integer);
}
})
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
System.out.println(s);
showMsg(s);
}
});
集合操作
from()
它接收一个集合作为输入，然后每次输出一个元素给subscriber
//集合操作
Observable.from(new String[]{"url1", "url2", "url3"})//它接收一个集合作为输入，然后每次输出一个元素给subscriber
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
System.out.println(s);
showMsg(s);
}
});
flatMap()
它接收一个Observable的输出作为输入，同时输出另外一个Observable
query("Hello, world!")
.flatMap(new Func1<List<String>, Observable<String>>() {//flatMap()接收一个Observable的输出作为输入，同时输出另外一个Observable。
@Override
public Observable<String> call(List<String> strings) {
return Observable.from(strings);
}
})
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
System.out.println(s);
showMsg(s);
}
});
现在假设有这么个函数：
Observable<List<String>> query(String text) {
List<String> list = Arrays.asList(new String[]{"url1", "url2", "url3"});
return Observable.just(list);
}
flatMap把一个Observable变成一堆Observable
query("Hello, world!")
.flatMap(new Func1<List<String>, Observable<String>>() {//flatMap()接收一个Observable的输出作为输入，同时输出另外一个Observable。
@Override
public Observable<String> call(List<String> strings) {
return Observable.from(strings);
}
})
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
System.out.println(s);
showMsg(s);
}
});
flatMap把一个Observable变成一个Observable，可以用map替换
query("Hello, world!")
.flatMap(new Func1<List<String>, Observable<String>>() {//flatMap()接收一个Observable的输出作为输入，同时输出另外一个Observable。
@Override
public Observable<String> call(List<String> strings) {
return Observable.from(strings);
}
})
.flatMap(new Func1<String, Observable<String>>() {
@Override
public Observable<String> call(String s) {
return Observable.just(s+s);
}
})
/* 与前一个flatMap功能一致
.map(new Func1<String, String>() {
@Override
public String call(String s) {
return s+s;
}
})
*/
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
System.out.println(s);
showMsg(s);
}
});
filter()
输出和输入相同的元素，并且会过滤掉那些不满足检查条件的
query("数据过滤")
.flatMap(new Func1<List<String>, Observable<String>>() {
@Override
public Observable<String> call(List<String> strings) {
return Observable.from(strings);
}
})
.filter(new Func1<String, Boolean>() {
@Override
public Boolean call(String s) {
return !s.equals("url1");
}
}) //filter()输出和输入相同的元素，并且会过滤掉那些不满足检查条件的。
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
showMsg(s);
}
});
指定结果个数
query("数据过滤")
.flatMap(new Func1<List<String>, Observable<String>>() {
@Override
public Observable<String> call(List<String> strings) {
return Observable.from(strings);
}
})
.take(2) //take()输出最多指定数量的结果。
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
showMsg(s);
}
});
订阅前对数据操作
//订阅之前对数据进行操作
query("数据过滤")
.flatMap(new Func1<List<String>, Observable<String>>() {
@Override
public Observable<String> call(List<String> strings) {
return Observable.from(strings);
}
})
.take(2) //take()输出最多指定数量的结果。
.doOnNext(new Action1<String>() {
@Override
public void call(String s) {
// TODO: 2016/8/15 something
}
}) //doOnNext()允许我们在每次输出一个元素之前做一些额外的事情
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
showMsg(s);
}
});
调度器
异步是RxJava最重要的特性之一，RxJava使用subscribeOn()和observeOn()来分别指定订阅的线程和观察的线程，订阅的线程即获取、处理数据的线程，观察的线程则一般是对数据进行展示的线程
eg：
retrofitService.getImage(url)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(bitmap -> myImageView.setImageBitmap(bitmap));
RxAndroid
AndroidSchedulers
retrofitService.getImage(url)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(bitmap -> myImageView.setImageBitmap(bitmap));
AndroidObservable
AndroidObservable，它提供了跟多的功能来配合Android的生命周期。bindActivity()和bindFragment()方法默认使用AndroidSchedulers.mainThread()来执行观察者代码，这两个方法会在Activity或者Fragment结束的时候通知被观察者停止发出新的消息。
AndroidObservable.bindActivity(this, retrofitService.getImage(url))
.subscribeOn(Schedulers.io())
.subscribe(bitmap -> myImageView.setImageBitmap(bitmap);
AndroidObservable.fromBroadcast()方法允许你创建一个类似BroadcastReceiver的Observable对象。下面的例子展示了如何在网络变化的时候被通知到：
IntentFilter filter = new IntentFilter(ConnectivityManager.CONNECTIVITY_ACTION);
AndroidObservable.fromBroadcast(context, filter)
.subscribe(intent -> handleConnectivityChange(intent));
ViewObservable
ViewObservable可以给View添加了一些绑定。如果你想在每次点击view的时候都收到一个事件，可以使用ViewObservable.clicks()，或者你想监听TextView的内容变化，可以使用ViewObservable.text()
ViewObservable.clicks(mCardNameEditText, false)
.subscribe(view -> handleClick(view));
参考链接
深入浅出RxJava（一：基础篇）
深入浅出RxJava(二：操作符)
深入浅出RxJava三--响应式的好处
深入浅出RxJava四-在Android中使用响应式编程
