RXjava学习资料：
https://www.gitbook.com/book/yuxingxin/rxjava-essentials-cn/details
如下只是学习笔记而已，后面添加实战案例，现在只是理论总结：
Rxjava语言特点：
1，易于并发从而更好的利用服务器的能力；
2，易于有条件的异步执行；
3，一种更好的方式来避免回调地狱；
4，一种响应式方法。
RXjava源于观察者模式：
添加了如下三个缺少的功能：
1，生产者在没有更多数据可用时能够发出信号通知：oncompleted()事件。
2，生产者在发生错误的时候能够发出信号通知：onError()事件。
3，RxJavaObservables能够组合而不是嵌套，从而避免开发者陷入回调地狱。
RXjava中的四个角色：
1，Observable:观察得到的，看的见得
2，Observer：观察者
3，Subscriber：订阅者
4，Subjects：服从。
Observables和Subjects是两个“生产”实体
Observers和Subscribers是两个“消费”实体。
Rxjava之Observable
onNext(T)检索数据
onError(Throwable)发现错误
onCompleted()完成
RXjava之热Observables和冷Observables
热Observables:只要创建完就发射数据，订阅他的观察者从序列中某位置接受数据。
冷Observables:一直等待，知道观察者订阅他才发射数据，可确保能收到数据序列。
Observale创建的两种方式：
1：create
Observable.create(new Observable.OnSubscribe<Object>() {
@Override
public void call(Subscriber<? super Object> subscriber) {
}
});2：from:从列表，数组来创建Observable,并一个个发射数据
List<Integer> items = new ArrayList<Integer>();
items.add(1);
Observable<Integer> observableString = Observable.from(items);
Subscription subscriptionPrint = observableString.subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
System.out.println("Observable completed");
}
@Override
public void onError(Throwable e) {
System.out.println("Oh,no! Something wrong happened！");
}
@Override
public void onNext(Integer item) {
System.out.println("Item is " + item);
}
});
3，just
just()方法可以传入一到九个参数，它们会按照传入的参数的顺序来发射它
们。
无理由不发射数据并结束操作：
Observable.empty(),Observable.never(),和
Observable.throw()
RxJava提供四种不同的Subject：
PublishSubject
BehaviorSubject:向订阅者发送截止订阅前最新的一个数据对象，然后发送订阅后数据流；
ReplaySubject:缓存所订阅的所有数据，向任意一个订阅他的观察者重发数据流；
AsyncSubject：当Observable完成AnsySubject只会发布最后一个数据给已定订阅的每一个观察者。
as中开发使用RXjava,RXandroid:
gradle依赖：
compile 'io.reactivex:rxandroid:1.1.0'
compile 'io.reactivex:rxjava:1.1.0'
其他推荐的as插件：
1，lombok：getsettostringequal等注解代码
2，butterknife：findviewbyidonclick代码等的注解代码
3，retrolambda：java8lambda相关函数
具体作用情百度或者谷歌进行搜索
just主要是得到原始的observable版本，在一个新的响应式架构的基础上迁移已存在的代码，这个方法可
能是一个有用的开始点。
private void loadApps(AppInfo appOne, AppInfo appTwo, AppInfo appThree)
mRecyclerView.setVisibility(View.VISIBLE);
Observable.just(appOne, appTwo, appThree)
.subscribe(new Observer<AppInfo>() {
@Override
public void onCompleted() {
mSwipeRefreshLayout.setRefreshing(false);
Toast.makeText(getActivity(), "Here is the list!"
}
@Override
public void onError(Throwable e) {
Toast.makeText(getActivity(), "Something went wrong!"
mSwipeRefreshLayout.setRefreshing(false);
}
@Override
public void onNext(AppInfo appInfo) {
mAddedApps.add(appInfo);
mAdapter.addApplication(mAddedApps.size() - 1, appInfo);
}
});
}
repeat()重复发数据：
repeat(3)：重复发送三次数据；
private void loadApps(AppInfo appOne, AppInfo appTwo, AppInfo appThree)
mRecyclerView.setVisibility(View.VISIBLE);
Observable.just(appOne, appTwo, appThree)
.repeat(3)
.subscribe(new Observer<AppInfo>() {
@Override
public void onCompleted() {
mSwipeRefreshLayout.setRefreshing(false);
Toast.makeText(getActivity(), "Here is the list!"
}
@Override
public void onError(Throwable e) {
Toast.makeText(getActivity(), "Something went wrong!"
mSwipeRefreshLayout.setRefreshing(false);
}
@Override
public void onNext(AppInfo appInfo) {
mAddedApps.add(appInfo);
mAdapter.addApplication(mAddedApps.size() - 1, appInfo);
}
});
}
defer()声明一个Observable但是你又想推迟这个Observable的创建直到观察者订阅时
1
private Observable<Integer> getInt() {
return Observable.create(subscriber -> {
if(subscriber.isUnsubscribed()) {
return;
}
App.L.debug("GETINT");
subscriber.onNext(42);
subscriber.onCompleted();
});
} 2
Observable<Integer> deferred = Observable.defer(this::getInt); 3
deferred.subscribe(number -> {
App.L.debug(String.valueOf(number));
});
range()
你需要从一个指定的数字X开始发射N个数字吗？你可以用 range，下面是从10开始，发送3个数字：
Observable.range(10, 3)
.subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
Toast.makeText(getActivity(), "Yeaaah!", Toast.LENGTH_LONG).show();
}
@Override
public void onError(Throwable e) {
Toast.makeText(getActivity(), "Something went wrong!", Toast.LENGTH_SHORT).show();
}
@Override
public void onNext(Integer number) {
Toast.makeText(getActivity(), "I say " + number, Toast.LENGTH_SHORT).show();
}
});
interval()
interval() 函数在你需要创建一个轮询程序时非常好用。
如下每隔3秒toast下数据：
Subscription stopMePlease = Observable.interval(3, TimeUnit.SECONDS)
.subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
Toast.makeText(getActivity(), "Yeaaah!", Toast.LENGTH_LONG).show();
}
@Override
public void onError(Throwable e) {
Toast.makeText(getActivity(), "Something went wrong!", Toast.LENGTH_SHORT).show();
}
@Override
public void onNext(Integer number) {
Toast.makeText(getActivity(), "I say " + number, Toast.LENGTH_SHORT).show();
}
});
timer()
如果你需要一个一段时间之后才发射的Observable，你可以像下面的例子使用timer()：
如下3秒后发送数据：
Observable.timer(3, TimeUnit.SECONDS)
.subscribe(new Observer<Long>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Long number) {
Log.d("RXJAVA", "I say " + number);
}
});
如何过滤数据 Observables？
RxJava让我们使用 filter() 方法来过滤我们观测序列中不想要的值，filter((appInfo) ->appInfo.getName().startsWith("C"))是只想展示c开头的数据。
1，过滤序列：
private void loadList(List<AppInfo> apps) {
mRecyclerView.setVisibility(View.VISIBLE);
Observable.from(apps)
.filter((appInfo) ->
appInfo.getName().startsWith("C"))
.subscribe(new Observer<AppInfo>() {
@Override
public void onCompleted() {
mSwipeRefreshLayout.setRefreshing(false);
}
@Override
public void onError(Throwable e) {
Toast.makeText(getActivity(), "Something went wrong!"
mSwipeRefreshLayout.setRefreshing(false);
}
@Override
public void onNext(AppInfo appInfo) {
mAddedApps.add(appInfo);
mAdapter.addApplication(mAddedApps.size() - 1, appInfo);
}
});
}过滤空数据：
.filter(new Func1<AppInfo, Boolean>() {
@Override
public Boolean call(AppInfo appInfo) {
return appInfo != null;
}
}) 获取开头或者结尾的几个数据：take() 或 takeLast()。
take前几个数据；
takelast后几个数据。
如下是获取前三个数据：
private void loadList(List<AppInfo> apps) {
mRecyclerView.setVisibility(View.VISIBLE);
Observable.from(apps)
.take(3)
.subscribe(new Observer<AppInfo>() {
@Override
public void onCompleted() {
mSwipeRefreshLayout.setRefreshing(false);
}
@Override
public void onError(Throwable e) {
Toast.makeText(getActivity(), "Something went wrong!"
mSwipeRefreshLayout.setRefreshing(false);
}
@Override
public void onNext(AppInfo appInfo) {
mAddedApps.add(appInfo);
mAdapter.addApplication(mAddedApps.size() - 1, appInfo);
}
});
}获取后三个数据：
Observable.from(apps)
.takeLast(3)
.subscribe(...);
数据去重Distinct:
制造重复数据：
Observable<AppInfo> fullOfDuplicates = Observable.from(apps)
.take(3)
.repeat(3);
去重：
fullOfDuplicates.distinct()
.subscribe(new Observer<AppInfo>() {
@Override
public void onCompleted() {
mSwipeRefreshLayout.setRefreshing(false);
}
@Override
public void onError(Throwable e) {
Toast.makeText(getActivity(), "Something went wrong!"
mSwipeRefreshLayout.setRefreshing(false);
}
@Override
public void onNext(AppInfo appInfo) {
mAddedApps.add(appInfo);
mAdapter.addApplication(mAddedApps.size() - 1, appInfo);
}
});
}
忽略掉重复的值并且在温度确实改变的时候才得到通知：ditinctuntilchanged()
也就是说有重复数据的时候，利用ditinctuntilchanged可以去重并展示重复数据中的一个数据。
发射第一个数据 first()
发射最后一个数据last()
观测序列完成，不在发射任何值得时候，firstOrDefault(),lastOrDefault()
跳过前两个数据 skip(2)
跳过后两个数据 skipLast(2)，也就是跳过后两个的其他数据发射。
发射指定位置的元素：elementAt(2)，从零开始，第三个元素发射
指定时间间隔由Observable发射最近一次的数值：
如下每隔三秒，updateDisplay(currentTemperature) 更新当前温度信息
Observable<Integer> sensor = [...]
sensor.sample(30, TimeUnit.SECONDS)
.subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Integer currentTemperature) {
updateDisplay(currentTemperature)
}
});
timeout: timeout() 为一个Observable的限时的副本,如果在指定的时间间隔内Observable不发射值的话，它监听的原始的Observable时就会触发 onError() 函
数，也就是限定时间的字段:
下面代码意思就是每隔两秒刷新当前温度，如果不发射数据，就会打印：You should go check the sensor, dude。
Subscription subscription = getCurrentTemperature()
.timeout(2, TimeUnit.SECONDS)
.subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
Log.d("RXJAVA", "You should go check the sensor, dude");
}
@Override
public void onNext(Integer currentTemperature) {
updateDisplay(currentTemperature)
}
});
Debounce
过滤掉由Observable发射的速率过快的数据；如果在一个指定的时间间隔过去了仍旧没有发射一个，那么它将发射最后的那个。
2016年4月6日23:02:38
转换Observables
Map() :将函数作用于每一个发出去的元素上面去，以此创建一个新的Observable来发射转换的数据。
如下就是将原来的apps的名称小写，并发出去。
private void loadList(List<AppInfo> apps) {
mRecyclerView.setVisibility(View.VISIBLE);
Observable.from(apps)
.map(new Func1<AppInfo, AppInfo>() {
@Override
public Appinfo call(AppInfo appInfo) {
String currentName = appInfo.getName();
String lowerCaseName = currentName.toLowerCase();
appInfo.setName(lowerCaseName);
return appInfo;
}
})
.subscribe(new Observer<AppInfo>() {
@Override
public void onCompleted() {
mSwipeRefreshLayout.setRefreshing(false);
}
@Override
public void onError(Throwable e) {
Toast.makeText(getActivity(), "Something went wrong!"
mSwipeRefreshLayout.setRefreshing(false);
}
@Override
public void onNext(AppInfo appInfo) {
mAddedApps.add(appInfo);
mAdapter.addApplication(mAddedApps.size() - 1, appInfo);
}
});
}FlatMap RxJava的 flatMap() 函数提供一种铺平序列的方式，然后合并这些Observables发射的数据，最后将合并后的结果作为最终的Observable。
ConcatMap：解决了 flatMap() 的交叉问题，提供了一种能够把发射的值连续在一起的铺平函数，而不是合并它们。
FlatMapIterable：flatMapInterable() 和 flatMap() 很像。仅有的本质不同是它将源数据两两结成对并生成Iterable，而不是原始数据项和生成的Observables。
SwitchMap：switchMap() 和 flatMap() 很像，除了一点：每当源Observable发射一个新的数据项（Observable）时，它将取消订阅并停止监视之前那个数据项
产生的Observable，并开始监视当前发射的这一个。
Scan：scan() 函数对原始
Observable发射的每一项数据都应用一个函数，计算出函数的结果值。
Observable.just(1, 2, 3, 4, 5)
.scan((sum, item) -> sum + item)
.subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
Log.d("RXJAVA", "Sequence completed.");
}
@Override
public void onError(Throwable e) {
Log.e("RXJAVA", "Something went south!");
}
@Override
public void onNext(Integer item) {
Log.d("RXJAVA", "item is: " + item);
}
});
groupBy() 来分组元素：
如下按照根据最近更新时间来进行分组。
Observable<GroupedObservable<String, AppInfo>> groupedItems = Observable.from(apps)
.groupBy(new Func1<AppInfo, String>() {
@Override
public String call(AppInfo appInfo) {
SimpleDateFormat formatter = new SimpleDateFormat("MM/yyyy"
return formatter.format(new Date(appInfo.getLastUpdateTime()));
}
});
Observable.concat(groupedItems)
.subscribe(new Observer<AppInfo>() {
@Override
public void onCompleted() {
mSwipeRefreshLayout.setRefreshing(false);
}
@Override
public void onError(Throwable e) {
Toast.makeText(getActivity(), "Something went wrong!", Toast.LENGTH_SHORT).show();
mSwipeRefreshLayout.setRefreshing(false);
}
@Override
public void onNext(AppInfo appInfo) {
mAddedApps.add(appInfo);
mAdapter.addApplication(mAddedApps.size() - 1, appInfo);
}
});
Buffer RxJava中的 buffer() 函数将源Observable变换一个新的Observable，这个新的Observable每次发射一组列表值而不是一个一个发射。
buffer(count = 3)一组三个的方式进行发射一个列表的obserable。
buffer(count = 2，skip = 2)一组两个，跳过第三个来进行发射一个列表的obserable。
buffer(timespan = 2,count = 2)每隔2秒，每组2个发射一个列表的obserable.
Window:RxJava的 window() 函数和 buffer() 很像，但是它发射的是Observable而不是列表。
window(count = 3)原来observable的自己，每隔obserable三个元素。
window(skip = 3,count = 2),每隔第三个，每隔obserable中有2个元素。
Cast:
map的特殊版本。将原observale每一项转换为新的类型，把他变成不同class.
组合observable
2016年4月7日22:23:38
Merge:
从单词角度分析，合并，rx中将帮助你把两个甚至更多的Observables合并到他们发射的数据项里。
private void loadList(List<AppInfo> apps) {
mRecyclerView.setVisibility(View.VISIBLE);
List reversedApps = Lists.reverse(apps);
Observable<AppInfo> observableApps = Observable.from(apps);
Observable<AppInfo> observableReversedApps = Observable.from(reversedApps);
//合并操作
Observable<AppInfo> mergedObserbable = Observable.merge(observableApps, observableReversedApps);
mergedObserbable.subscribe(new Observer<AppInfo>() {
@Override
public void onCompleted() {
mSwipeRefreshLayout.setRefreshing(false);
Toast.makeText(getActivity(), "Here is the list!", Toast.LENGTH_LONG).show();
}
@Override
public void onError(Throwable e) {
Toast.makeText(getActivity(), "One of the two Observable threw an error!"
mSwipeRefreshLayout.setRefreshing(false);
}
@Override
public void onNext(AppInfoappInfo) {
mAddedApps.add(appInfo);
mAdapter.addApplication(mAddedApps.size() - 1, appInfo);
}
});
}Zip操作：
zip() 合并两个或者多个Observables发射出的数据项，根据指定的函数 Func* 变换它们，并发射一个新值。
private void loadList(List<AppInfo> apps) {
mRecyclerView.setVisibility(View.VISIBLE);
Observable<AppInfo> observableApp = Observable.from(apps);
Observable<Long> tictoc = Observable.interval(1, TimeUnit.SECONDS);
Observable.zip(observableApp, tictoc,
(AppInfo appInfo, Long time) -> updateTitle(appInfo, time))
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<AppInfo>() {
@Override
public void onCompleted() {
Toast.makeText(getActivity(), "Here is the list!", Toast.LENGTH_LONG).show();
}
@Override
public void onError(Throwable e) {
mSwipeRefreshLayout.setRefreshing(false);
Toast.makeText(getActivity(), "Something went wrong!", Toast.LENGTH_SHORT).show();
}
@Override
public void onNext(AppInfoappInfo) {
if (mSwipeRefreshLayout.isRefreshing()) {
mSwipeRefreshLayout.setRefreshing(false);
}
mAddedApps.add(appInfo);
int position = mAddedApps.size() - 1;
mAdapter.addApplication(position, appInfo);
mRecyclerView.smoothScrollToPosition(position);
}
});
} Join操作：
RxJava的 join() 函数基于时间窗口将两个Observables发射的数据结合在一起。
如下意思：tictoc 这个Observable数据每秒只发射一个新的 Long 型整数。
为了合并它们，我们需要指定两个 Func1 变量：
appInfo -> Observable.timer(2, TimeUnit.SECONDS)
time -> Observable.timer(0, TimeUnit.SECONDS)
上面描述了两个时间窗口。下面一行描述我们如何使用 Func2 将两个发射的数据
结合在一起。
this::updateTitle
private void loadList(List<AppInfo> apps) {
mRecyclerView.setVisibility(View.VISIBLE);
Observable<AppInfo> appsSequence =
Observable.interval(1000, TimeUnit.MILLISECONDS)
.map(position -> {
return apps.get(position.intValue());
});
Observable<Long> tictoc = Observable.interval(1000, TimeUnit.MILLISECONDS);
appsSequence.join(
tictoc,
appInfo -> Observable.timer(2, TimeUnit.SECONDS),
time -> Observable.timer(0, TimeUnit.SECONDS),
this::updateTitle)
.observeOn(AndroidSchedulers.mainThread())
.take(10)
.subscribe(new Observer<AppInfo>() {
@Override
public void onCompleted() {
Toast.makeText(getActivity(), "Here is the list!"
}
@Override
public void onError(Throwable e) {
mSwipeRefreshLayout.setRefreshing(false);
Toast.makeText(getActivity(), "Something went wrong!"
}
@Override
public void onNext(AppInfoappInfo) {
if (mSwipeRefreshLayout.isRefreshing()) {
mSwipeRefreshLayout.setRefreshing(false);
}
mAddedApps.add(appInfo);
int position = mAddedApps.size() - 1;
mAdapter.addApplication(position, appInfo);
mRecyclerView.smoothScrollToPosition(position);
}
});
} combineLatest操作：
zip() 作用于最近未打包的两个Observables。相反， combineLatest() 作用于最近发射的数据项：如果 Observable1 发射了A并且 Observable2 发射了B和C， combineLatest() 将会分组处理AB和AC。
如下：
一个是每秒钟从我们已安装的应用列表发射一个App数据，第二个是每隔1.5秒发射一个 Long 型整数。我们将他们结合起来并执行 updateTitle() 函数。
private void loadList(List<AppInfo> apps) {
mRecyclerView.setVisibility(View.VISIBLE);
Observable<AppInfo> appsSequence = Observable.interval(1000, TimeUnit.MILLISECONDS)
.map(position ->apps.get(position.intValue()));
Observable<Long> tictoc = Observable.interval(1500, TimeUnit.MILLISECONDS);
Observable.combineLatest(appsSequence, tictoc,
this::updateTitle)
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<AppInfo>() {
@Override
public void onCompleted() {
Toast.makeText(getActivity(), "Here is the list!", Toast.LENGTH_LONG).show();
}
@Override
public void onError(Throwable e) {
mSwipeRefreshLayout.setRefreshing(false);
Toast.makeText(getActivity(), "Something went wrong!", Toast.LENGTH_SHORT).show();
}
@Override
public void onNext(AppInfoappInfo) {
if (mSwipeRefreshLayout.isRefreshing()) {
mSwipeRefreshLayout.setRefreshing(false);
}
mAddedApps.add(appInfo);
int position = mAddedApps.size() - 1;
mAdapter.addApplication(position, appInfo);
mRecyclerView.smoothScrollToPosition(position);
}
});
}And,Then和When：
我们有两个发射的序列， observableApp ，发射我们安装的应用列表数据， tictoc 每秒发射一个 Long 型整数。现在我们用 and() 连接源
Observable和第二个Observable。
private void loadList(List<AppInfo> apps) {
mRecyclerView.setVisibility(View.VISIBLE);
Observable<AppInfo> observableApp = Observable.from(apps);
Observable<Long> tictoc = Observable.interval(1, TimeUnit.SECONDS);
Pattern2<AppInfo, Long> pattern = JoinObservable.from(observableApp).and(tictoc);
Plan0<AppInfo> plan = pattern.then(this::updateTitle);
JoinObservable
.when(plan)
.toObservable()
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<AppInfo>() {
@Override
public void onCompleted() {
Toast.makeText(getActivity(), "Here is the list!", Toast.LENGTH_LONG).show();
}
@Override
public void onError(Throwable e) {
mSwipeRefreshLayout.setRefreshing(false);
Toast.makeText(getActivity(), "Something went wrong!"
}
@Override
public void onNext(AppInfoappInfo) {
if (mSwipeRefreshLayout.isRefreshing()) {
mSwipeRefreshLayout.setRefreshing(false);
}
mAddedApps.add(appInfo);
int position = mAddedApps.size() - 1;
mAdapter.addApplication(position, appInfo);
mRecyclerView.smoothScrollToPosition(position);
}
});
}Switch：
subscribe-unsubscribe 的序列里我们能够从一个Observable自动取消订阅来订阅一个新的Observable。
将一个发射多个Observables的Observable转换成另一个单独的Observable，后者发射那些Observables最近发射的数据项。
StartWith：
startWith() 是 concat() 的对应部分。正如 concat() 向发射数据的Observable追加数据那样，在Observable开始发射他们的数据之前，
startWith() 通过传递一个参数来先发射一个数据序列。
Schedulers调度器
处理多线程异步操作：
RxJava提供了5种调度器：
.io() ：IO操作，增长缩减自适应线程池（线程池无限制）
.computation：计算工作默认的调度器，调度器：buffer(),debounce(),delay(),interval(),sample,skip()。
.immediate()：允许在当前线程执行你指定的工作，是timeout(),timeInterval(),timestamp()默认调度器。
.newThread()：为指定任务启动一个线程。
.trampoline()：任务入队列，按需处理队列中的任务，是repeat()和retry()默认调度器。
如下就是简单将对图片文件的IO操作通过RX的io调度器来线程池中进行处理。
public static void storeBitmap(Context context, Bitmap bitmap, String filename)
Schedulers.io().createWorker().schedule(() -> {
blockingStoreBitmap(context, bitmap, filename);
});
}
private static void blockingStoreBitmap(Context context, Bitmap bitmap, String filename)
FileOutputStream fOut = null;
try {
fOut = context.openFileOutput(filename, Context.MODE_PRIVATE);
bitmap.compress(Bitmap.CompressFormat.PNG, 100, fOut);
fOut.flush();
fOut.close();
} catch (Exception e) {
throw new RuntimeException(e);
} finally {
try {
if (fOut != null) {
fOut.close();
}
} catch (IOException e) {
throw new RuntimeException(e);
}
}
}
在主线程中返回结果。
onBackpressureBuffer() 方法将告诉Observable发射的数据如果比观察者消费的数据要更快的话，它必须把它们存储在缓存中并提供一个合适的时间给它们。
getApps()
.onBackpressureBuffer()
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<AppInfo>() {
[...]
处理耗时的任务:
getObservableApps(apps)
.onBackpressureBuffer()
.subscribeOn(Schedulers.computation())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<AppInfo>() {
[...]