Rxjava、retrofit初探

刚开始学习，先记录一些东西，以后再做整理

建议查看深入浅出RxJava

Awesome-RxJava lzyzsd收集的关于Rxjava的知识

给 Android 开发者的 RxJava 详解

———————–所以下面就不用看了，只是搜集一些我觉得挺重要的，要条理的话看上面的

package com.example.kevin.rxjava;

import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;

import rx.Observable;
import rx.Subscriber;
import rx.functions.Action1;
import rx.functions.Func1;

public class MainActivity extends AppCompatActivity {

@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);

Observable<String> myObservable = Observable.create(
new Observable.OnSubscribe<String>(){

@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("Hello Rxjava");
subscriber.onCompleted();
}
}
);

Subscriber<String> mySubscriber = new Subscriber<String>() {
@Override
public void onCompleted() {

}

@Override
public void onError(Throwable e) {

}

@Override
public void onNext(String s) {
System.out.println(s);
}
};

myObservable.subscribe(mySubscriber);

Observable.just("Hello RxJava")
.map(new Func1<String, Integer>() {
@Override
public Integer call(String s) {
return s.hashCode();
}
}).subscribe(new Action1<Integer>() {
@Override
public void call(Integer s) {
System.out.println(s);
}
});
}
}

需要搞懂：

Observable、OnSubscribe、Subscriber

flatMap()：接收一个Observable的输出作为输入，同时输出另外一个Observable

filter()：输出和输入相同的元素，并且会过滤掉那些不满足检查条件的。

take()

from()：它接收一个集合作为输入，然后每次输出一个元素给subscriber：

doOnNext()允许我们在每次输出一个元素之前做一些额外的事情，比如这里的保存标题。

.subscribeOn(Schedulers.io())

observerOn()

a. 打印字符串数组

将字符串数组 names 中的所有字符串依次打印出来：

String[] names = ...;
Observable.from(names)
.subscribe(new Action1<String>() {
@Override
public void call(String name) {
Log.d(tag, name);
}
});

b. 由 id 取得图片并显示

由指定的一个 drawable 文件 id drawableRes 取得图片，并显示在 ImageView 中，并在出现异常的时候打印 Toast 报错：

int drawableRes = ...;
ImageView imageView = ...;
Observable.create(new OnSubscribe<Drawable>() {
@Override
public void call(Subscriber<? super Drawable> subscriber) {
Drawable drawable = getTheme().getDrawable(drawableRes));
subscriber.onNext(drawable);
subscriber.onCompleted();
}
}).subscribe(new Observer<Drawable>() {
@Override
public void onNext(Drawable drawable) {
imageView.setImageDrawable(drawable);
}

@Override
public void onCompleted() {
}

@Override
public void onError(Throwable e) {
Toast.makeText(activity, "Error!", Toast.LENGTH_SHORT).show();
}
});

线程控制 —— Scheduler,有了Scheduler ，就可以使用 subscribeOn() 和 observeOn() 两个方法来对线程进行控制了。

Schedulers.immediate(): 直接在当前线程运行，相当于不指定线程。这是默认的 Scheduler。

Schedulers.newThread(): 总是启用新线程，并在新线程执行操作。

Schedulers.io(): I/O 操作（读写文件、读写数据库、网络信息交互等）所使用的 Scheduler。行为模式和

newThread() 差不多，区别在于 io() 的内部实现是是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下 io()

比 newThread() 更有效率。不要把计算工作放在 io() 中，可以避免创建不必要的线程。

Schedulers.computation(): 计算所使用的 Scheduler。这个计算指的是 CPU 密集型计算，即不会被 I/O

等操作限制性能的操作，例如图形的计算。这个 Scheduler 使用的固定的线程池，大小为 CPU 核数。不要把 I/O 操作放在

computation() 中，否则 I/O 操作的等待时间会浪费 CPU。

另外， Android 还有一个专用的 AndroidSchedulers.mainThread()，它指定的操作将在 Android

主线程运行。

Observable.just(1, 2, 3, 4)
.subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
.observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer number) {
Log.d(tag, "number:" + number);
}
});

上面这段代码中，由于 subscribeOn(Schedulers.io()) 的指定，被创建的事件的内容 1、2、3、4 将会在 IO 线程发出；而由于 observeOn(AndroidScheculers.mainThread()) 的指定，因此 subscriber 数字的打印将发生在主线程 。事实上，这种在 subscribe() 之前写上两句 subscribeOn(Scheduler.io()) 和 observeOn(AndroidSchedulers.mainThread()) 的使用方式非常常见，它适用于多数的 『后台线程取数据，主线程显示』的程序策略。

Scheduler 的原理

RxJava 提供了对事件序列进行变换的支持，这是它的核心功能之一，也是大多数人说『RxJava 真是太好用了』的最大原因。所谓变换，就是将事件序列中的对象或整个序列进行加工处理，转换成不同的事件或事件序列。

retrofit + rxjava

Retrofit

大名鼎鼎的Retrofit库内置了对RxJava的支持。通常调用发可以通过使用一个Callback对象来获取异步的结果：

@GET("/user/{id}/photo")
void getUserPhoto(@Path("id") int id, Callback<Photo> cb);

两个请求并发的发出:

Observable.zip(
service.getUserPhoto(id),
service.getPhotoMetadata(id),
(photo, metadata) -> createPhotoWithData(photo, metadata))
.subscribe(photoWithData -> showPhoto(photoWithData));

@GET("/token")
public Observable<String> getToken();

@GET("/user")
public Observable<User> getUser(@Query("token") String token, @Query("userId") String userId);

...

getToken()
.flatMap(new Func1<String, Observable<User>>() {
@Override
public Observable<User> onNext(String token) {
return getUser(token, userId);
})
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<User>() {
@Override
public void onNext(User user) {
userView.setUser(user);
}

@Override
public void onCompleted() {
}

@Override
public void onError(Throwable error) {
// Error handling
...
}
});

RxJava使用场景小结

取数据先检查缓存的场景

取数据，首先检查内存是否有缓存

然后检查文件缓存中是否有

最后才从网络中取

前面任何一个条件满足，就不会执行后面的

final Observable<String> memory = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
if (memoryCache != null) {
subscriber.onNext(memoryCache);
} else {
subscriber.onCompleted();
}
}
});
Observable<String> disk = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
String cachePref = rxPreferences.getString("cache").get();
if (!TextUtils.isEmpty(cachePref)) {
subscriber.onNext(cachePref);
} else {
subscriber.onCompleted();
}
}
});

Observable<String> network = Observable.just("network");

//主要就是靠concat operator来实现
Observable.concat(memory, disk, network)
.first()
.subscribeOn(Schedulers.newThread())
.subscribe(s -> {
memoryCache = "memory";
System.out.println("--------------subscribe: " + s);
});

界面需要等到多个接口并发取完数据，再更新

//拼接两个Observable的输出，不保证顺序，按照事件产生的顺序发送给订阅者
private void testMerge() {
Observable<String> observable1 = DemoUtils.createObservable1().subscribeOn(Schedulers.newThread());
Observable<String> observable2 = DemoUtils.createObservable2().subscribeOn(Schedulers.newThread());

Observable.merge(observable1, observable2)
.subscribeOn(Schedulers.newThread())
.subscribe(System.out::println);
}

一个接口的请求依赖另一个API请求返回的数据

举个例子，我们经常在需要登陆之后，根据拿到的token去获取消息列表。

这里用RxJava主要解决嵌套回调的问题，有一个专有名词叫Callback hell

NetworkService.getToken("username", "password")
.flatMap(s -> NetworkService.getMessage(s))
.subscribe(s -> {
System.out.println("message: " + s);
});

界面按钮需要防止连续点击的情况

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

响应式的界面

比如勾选了某个checkbox，自动更新对应的preference

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());

复杂的数据变换

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