前言
这篇文章是android-architecture源码分析系列文章的第四篇，我们将对todo-mvp-rxjava项目Demo进行源码分析。todo-mvp-rxjava项目是在todo‑mvp项目的基础上添加了RxJava响应式编程，希望读者有一定的Rxjava的基础。
引入RxJava和Lambda
RxJava
RxJava是近来最火的一个响应式编程的库，在项目中引入RxJava的根本目的在于：使用RxJava可以在程序逻辑复杂的情况下，在不断做线程调度的同时依然保持代码简洁。
RxJava：https://github.com/ReactiveX/RxJava
RxAndroid：https://github.com/ReactiveX/RxAndroid
在项目中引入RxJava：
compile "io.reactivex:rxjava:$rootProject.rxjavaVersion"
compile "io.reactivex:rxandroid:$rootProject.rxandroidVersion"
在写本文时RxJava早已迭代出2.0版本，但由于todo-mvp-rxjava的Demo使用的还是1.0版本的API，这里并没有做出修改，还是使用1.0版本。
Lambda
Java8开始引入了Lambda表达式，它让代码变得更加简洁，但是也降低了代码的可读性。
在Android项目中引入Lambda时，需要在App下的 build.gradle 文件中输入以下内容：
android {
...
defaultConfig {
jackOptions {
enabled true
}
}
compileOptions {
sourceCompatibility JavaVersion.VERSION_1_8
targetCompatibility JavaVersion.VERSION_1_8
}
...
源码分析
我们开始分析todo-mvp-rxjava的代码，项目结构其实与todo-mvp差别不大，重点就是分析RxJava在项目中的使用。
从上图可以看到，util包里面添加了对RxJava的线程调度的封装，我们进一步分析具体实现类SchedulerProvider的源码。
/**
* Provides different types of schedulers.
*/
public class SchedulerProvider implements BaseSchedulerProvider {
@Nullable
private static SchedulerProvider INSTANCE;
// Prevent direct instantiation.
private SchedulerProvider() {
}
public static synchronized SchedulerProvider getInstance() {
if (INSTANCE == null) {
INSTANCE = new SchedulerProvider();
}
return INSTANCE;
}
@Override
@NonNull
public Scheduler computation() {
return Schedulers.computation();
}
@Override
@NonNull
public Scheduler io() {
return Schedulers.io();
}
@Override
@NonNull
public Scheduler ui() {
return AndroidSchedulers.mainThread();
}
}
SchedulerProvider类对外提供了三种不同的可调度的线程类型。
- Schedulers.computation()：CPU密集型计算所使用的Scheduler。
- Schedulers.io()：I/O操作（读写文件、读写数据库、网络信息交互等）所使用的 Scheduler。
- AndroidSchedulers.mainThread()：运行在Android UI线程上。
BasePresenter类也基于RxJava做了更改：
public interface BasePresenter {
void subscribe();
void unsubscribe();
}
Presenter需要根据Activity或者View的生命周期来调用subscribe()和unsubscribe();
我们还是以AddEditTask模块为例，接着分析AddEditTaskPresenter
/**
* Listens to user actions from the UI ({@link AddEditTaskFragment}), retrieves the data and updates
* the UI as required.
*/
public class AddEditTaskPresenter implements AddEditTaskContract.Presenter {
@NonNull
private final TasksDataSource mTasksRepository;
@NonNull
private final AddEditTaskContract.View mAddTaskView;
@NonNull
private final BaseSchedulerProvider mSchedulerProvider;
@Nullable
private String mTaskId;
private boolean mIsDataMissing;
@NonNull
private CompositeSubscription mSubscriptions;
/**
* Creates a presenter for the add/edit view.
*
* @param taskId ID of the task to edit or null for a new task
* @param tasksRepository a repository of data for tasks
* @param addTaskView the add/edit view
* @param shouldLoadDataFromRepo whether data needs to be loaded or not (for config changes)
*/
public AddEditTaskPresenter(@Nullable String taskId, @NonNull TasksDataSource tasksRepository,
@NonNull AddEditTaskContract.View addTaskView, boolean shouldLoadDataFromRepo,
@NonNull BaseSchedulerProvider schedulerProvider) {
mTaskId = taskId;
mTasksRepository = checkNotNull(tasksRepository);
mAddTaskView = checkNotNull(addTaskView);
mIsDataMissing = shouldLoadDataFromRepo;
mSchedulerProvider = checkNotNull(schedulerProvider, "schedulerProvider cannot be null!");
mSubscriptions = new CompositeSubscription();
mAddTaskView.setPresenter(this);
}
@Override
public void subscribe() {
if (!isNewTask() && mIsDataMissing) {
populateTask();
}
}
@Override
public void unsubscribe() {
mSubscriptions.clear();
}
...
@Override
public void populateTask() {
if (isNewTask()) {
throw new RuntimeException("populateTask() was called but task is new.");
}
mSubscriptions.add(mTasksRepository
.getTask(mTaskId)
.subscribeOn(mSchedulerProvider.computation()) //执行在computation线程中
.observeOn(mSchedulerProvider.ui()) //回调方法执行在主线程中
.subscribe(
// onNext
task -> {
if (mAddTaskView.isActive()) {
mAddTaskView.setTitle(task.getTitle());
mAddTaskView.setDescription(task.getDescription());
mIsDataMissing = false;
}
}, // onError
__ -> {
if (mAddTaskView.isActive()) {
mAddTaskView.showEmptyTaskError();
}
}));
}
...
}
CompositeSubscription就是用来持有所有的Subscriptions，然后在生命周期结束(unsubscribe方法)时取消所有的订阅。
mAddTaskView.setPresenter(this);是将Presenter赋值到View中，由View的生命周期管理Presenter的状态:
- 在View的onResume()方法中调用Presenter的subscribe();
- 在View的onPause()方法中调用Presenter的unsubscribe();
Observable.subscribe绑定的时候会返回一个Subscription的对象，add到mSubscriptions里面统一管理。
mTasksRepository.getTask(mTaskId)返回的是Observable<Task>。Observable 即被观察者，它决定什么时候触发事件以及触发怎样的事件。
TasksRepository.getTask(taskId)方法：
/**
* Gets tasks from local data source (sqlite) unless the table is new or empty. In that case it
* uses the network data source. This is done to simplify the sample.
*/
@Override
public Observable<Task> getTask(@NonNull final String taskId) {
checkNotNull(taskId);
final Task cachedTask = getTaskWithId(taskId);
// Respond immediately with cache if available
if (cachedTask != null) {
return Observable.just(cachedTask);
}
// Load from server/persisted if needed.
// Do in memory cache update to keep the app UI up to date
if (mCachedTasks == null) {
mCachedTasks = new LinkedHashMap<>();
}
// Is the task in the local data source? If not, query the network.
Observable<Task> localTask = getTaskWithIdFromLocalRepository(taskId);
Observable<Task> remoteTask = mTasksRemoteDataSource
.getTask(taskId)
.doOnNext(task -> {
mTasksLocalDataSource.saveTask(task);
mCachedTasks.put(task.getId(), task);
});
return Observable.concat(localTask, remoteTask).first()
.map(task -> {
if (task == null) {
throw new NoSuchElementException("No task found with taskId " + taskId);
}
return task;
});
}
简单描述getTask(taskId)方法的流程：
- 在本地数据中查询对应taskId的数据。
- 从远程数据中查询对应taskId的数据。
- 将获得的本地数据对象和远程数据对象依次发送出来。
- 取得满足指定条件的第一个数据。
- 将这条数据转换成Task对象。
- 返回这个Task对象。
concat：接收若干个Observables，发射数据是有序的，不会交叉。
first：获取源Observable产生的第一个数据或者满足指定条件的第一个数据。
map：事件对象的直接变换。
总结
todo-mvp-rxjava是在todo-mvp的基础上集成了RxJava库，相比之下，从程序架构看来并没有太多的不一样，主要区别还是在Model层对数据操作过程中进行了多次线程调度，使用到了RxJava的响应式编程简化了代码流程。而本文并没有对RxJava做过多的介绍，这也不是本文的初衷。
从RxJava的火爆程度来看，在Android项目架构中引入RxJava，以后或许会成为常态。
