Android和iOS开发中的异步处理（四）——异步任务和队列

如上图所示，生产者和消费者都运行在一个线程，即主线程。按照这种思路来实现任务队列，我们需要执行的任务本身必须是异步的，否则整个队列的任务就没法异步化。

我们定义要执行的异步任务的接口如下：

public interface Task {
/**
     * 唯一标识当前任务的ID
     * @return
     */
String getTaskID();

/**
     * 由于任务是异步任务, 
     * 那么start方法被调用只是启动任务;
     * 任务完成后会回调TaskListener.
     *
     * 注: start方法需在主线程上执行.
     */
voID start();

/**
     * 设置回调监听.
     * @param Listener
     */
voID setListener(TaskListener Listener);

/**
     * 异步任务回调接口.
     */
interface TaskListener {
/**
         * 当前任务完成的回调.
         * @param task
         */
voID taskComplete(Task task);
/**
         * 当前任务执行失败的回调.
         * @param task
         * @param cause 失败原因
         */
voID taskFailed(Task task, Throwable cause);
}
}

由于Task是一个异步任务，所以我们为它定义了一个回调接口TaskListener。

getTaskID是为了得到一个能唯一标识当前任务的ID，便于对不同任务进行精确区分。

另外，为了更通用的表达失败原因，我们这里选用一个Throwable对象来表达（注：在实际编程中这未必是一个值得效仿的做法，具体情况请具体分析）。

有人可能会说：这里把Task接口定义成异步的，那如果想执行一个同步的任务该怎么办？这其实很好办。把同步任务改造成异步任务是很简单的，有很多种方法（反过来却很难）。

任务队列的接口，定义如下：

public interface TaskQueue {
/**
     * 向队列中添加一个任务.
     * @param task
     */
voID addTask(Task task);

/**
     * 设置监听器.
     * @param Listener
     */
voID setListener(TaskQueueListener Listener);

/**
     * 销毁队列.
     * 注: 队列在最后不用的时候, 应该主动销毁它.
     */
voID destroy();

/**
     * 任务队列对外监听接口.
     */
interface TaskQueueListener {
/**
         * 任务完成的回调.
         * @param task
         */
voID taskComplete(Task task);
/**
         * 任务最终失败的回调.
         * @param task
         * @param cause 失败原因
         */
voID taskFailed(Task task, Throwable cause);
}
}

任务队列TaskQueue本身的 *** 作也是异步的，addTask只是将任务放入队列，至于它什么时候完成（或失败），调用者需要监听TaskQueueListener接口。

需要注意的一点是，TaskQueueListener的taskFailed，与前面TaskListener的taskFailed不同，它表示任务在经过一定次数的失败后，最终放弃重试从而最终失败。而后者只表示那个任务一次执行失败。

我们重点讨论TaskQueue的实现，而Task的实现我们这里不关心，我们只关心它的接口。TaskQueue的实现代码如下：

public class CallbackBasedTaskQueue implements TaskQueue, Task.TaskListener {
private static final String TAG = "TaskQueue";

/**
     * Task排队的队列. 
     * 不需要thread-safe
     */
private Queue<Task> taskQueue = new linkedList<Task>();

private TaskQueueListener Listener;
private boolean stopped;

/**
     * 一个任务最多重试次数.
     * 若重试次数超过MAX_RETRIES, 
     * 任务则最终失败.
     */
private static final int MAX_RETRIES = 3;
/**
     * 当前任务的执行次数记录
     * (当尝试超过MAX_RETRIES时就最终失败)
     */
private int runcount;

@OverrIDe
public voID addTask(Task task) {
//新任务加入队列
taskQueue.offer(task);
task.setListener(this);

if (taskQueue.size() == 1 && !stopped) {
//当前是第一个排队任务, 立即执行它
launchNextTask();
}
}

@OverrIDe
public voID setListener(TaskQueueListener Listener) {
this.Listener = Listener;
}

@OverrIDe
public voID destroy() {
stopped = true;
}

private voID launchNextTask() {
//取当前队列头的任务, 但不出队列
Task task = taskQueue.peek();
if (task == null) {
//impossible case
Log.e(TAG, "impossible: NO task in queue, unexpected!");
return;
}

Log.d(TAG, "start task (" + task.getTaskID() + ")");
task.start();
runcount = 1;
}

@OverrIDe
public voID taskComplete(Task task) {
Log.d(TAG, "task (" + task.getTaskID() + ") complete");
finishTask(task, null);
}

@OverrIDe
public voID taskFailed(Task task, Throwable error) {
if (runcount < MAX_RETRIES && !stopped) {
//可以继续尝试
Log.d(TAG, "task (" + task.getTaskID() + ") Failed, try again. runcount: " + runcount);
task.start();
runcount++;
}
else {
//最终失败
Log.d(TAG, "task (" + task.getTaskID() + ") Failed, final Failed! runcount: " + runcount);
finishTask(task, error);
}
}

/**
     * 一个任务最终结束(成功或最终失败)后的处理
     * @param task
     * @param error
     */
private voID finishTask(Task task, Throwable error) {
//回调
if (Listener != null && !stopped) {
try {
if (error == null) {
Listener.taskComplete(task);
}
else {
Listener.taskFailed(task, error);
}
}
catch (Throwable e) {
Log.e(TAG, "", e);
}
}
task.setListener(null);

//出队列
taskQueue.poll();

//启动队列下一个任务
if (taskQueue.size() > 0 && !stopped) {
launchNextTask();
}
}
}

在这个实现中，我们需要注意的几点是：

进出队列的所有 *** 作（offer, peek, take）都运行在主线程，所以队列数据结构不再需要线程安全。我们选择了linkedList的实现。

任务的启动执行，依赖两个机会：

任务进队列addTask的时候，如果原来队列为空（当前任务是第一个任务），那么启动它；

一个任务执行完成（成功了，或者最终失败了）后，如果队列里有排队的其它任务，那么取下一个任务启动执行。

任务一次执行失败，并不算失败，还要经过若干次重试。如果重试次数超过MAX_RETRIES，才算最终失败。runcount记录了当前任务的累计执行次数。

CallbackBasedTaskQueue的代码揭示了任务队列的基本实现模式。

任务队列对于失败任务的重试策略，大大提高了最终成功的概率。在GitHub上的演示程序中，我把Task的失败概率设置得很高（高达80%），在重试3次的配置下，当任务执行的时候仍然有比较大的概率能最终执行成功。

关于RxJava到底有什么用？网上有很多讨论。

有人说，RxJava就是为了异步。这个当然没错，但说得不具体。

也有人说，RxJava的真正好处就是它提供的各种lift变换。还有人说，RxJava最大的用处是它的Schedulers机制，能够方便地切换线程。其实这些都不是革命性的关键因素。

那关键的是什么呢？我个人认为，是它对于回调接口设计产生的根本性的影响：它消除了为每个异步接口单独定义回调接口的必要性。

这里马上就有一个例子。我们使用RxJava对TaskQueue接口重新进行改写。

public interface TaskQueue {
/**
     * 向队列中添加一个任务.
     *
     * @param task
     * @param <R> 异步任务执行完要返回的数据类型.
     * @return 一个Observable. 调用者通过这个Observable获取异步任务执行结果.
     */
<R> Observable<R> addTask(Task<R> task);

/**
     * 销毁队列.
     * 注: 队列在最后不用的时候, 应该主动销毁它.
     */
voID destroy();
}

我们仔细看一看这个修改后的TaskQueue接口定义。

原来的回调接口TaskQueueListener没有了。

异步接口addTask原来没有返回值，现在返回了一个Observable。调用者拿到这个Observable，然后去订阅它（subscribe），就能获得任务执行结果（成功或失败）。这里的改动很关键。本来addTask什么也不返回，要想获得结果必须监听一个回调接口，这是典型的异步任务的运作方式。但这里返回一个Observable之后，让它感觉上非常类似一个同步接口了。说得再抽象一点，这个Observable是我们站在当下对于未来的一个指代，本来还没有运行的、发生在未来的虚无缥缈的任务，这时候有一个实实在在的东西被我们抓在手里了。而且我们还能对它在当下就进行很多 *** 作，并可以和其它Observable结合。这是这一思想真正的强大之处。

相应地，Task接口本来也是一个异步接口，自然也可以用这种方式进行修改：

/**
 * 异步任务接口定义.
 *
 * 不再使用TaskListener传递回调, 
 * 而是使用Observable.
 *
 * @param <R> 异步任务执行完要返回的数据类型.
 */
public interface Task <R> {
/**
     * 唯一标识当前任务的ID
     * @return
     */
String getTaskID();

/**
     *
     * 启动任务.
     *
     * 注: start方法需在主线程上执行.
     *
     * @return 一个Observable. 调用者通过这个Observable获取异步任务执行结果.
     */
Observable<R> start();
}

这里把改为RxJava的接口讨论清楚了，具体的队列实现反而不重要了。具体实现代码就不在这里讨论了，想了解详情的同学还是参见GitHub。注意GitHub的实现中用到了一个小技巧：把一个异步的任务封装成Observable，我们可以使用AsyncOnSubscribe。