Android 与并发 1

发表于 2016-05-04 更新于 2022-08-30 分类于 Android 阅读次数：

这个主题分为两个部分来讲，第一部分主要讲 Android 中的 Handler，AsyncTask，Thread 以及一些相关的开源库（RxJava，Eventbus），第二部分主要讲 Atomic，volatile，synchronized 等 JSR-133 中的东西。本文是第一部分。

1.什么是并发

一个逻辑流的执行在时间上于另一个流重叠，称为并发流，多个流并发的执行称为并发。这里的逻辑流简单来说就是代码片段。下面用一个图简单说明下：

CPU 并不由某一个进程独占，CPU 会在运行的时候在多个进程之间切换让用户以为当前的进程是独占 CPU 的，在这个图中，A 和 B，A 和 C 都是并发运行，而 B 和 C 不是，因为 B 和 C 的运行时间并不重叠。

由于进程创建的开销比较大，并且进程之间通信的开销也比较大，后来就有了线程的概念。线程是运行在进程上下文中的逻辑流，每个线程都有自己的上下文，并且线程之间共享进程的空间。这里的线程就是 Java 中平时我们用的 Thread，在 Linux 的 C/C++ 中是 pthread。需要注意的是，并发和 CPU 的核数没有必然的联系。

2.Thread 和 Runnable

在 Java 中 Runnable 仅仅是个接口，是一个抽象上的概念，实现了这个抽象的就是 Thread。Thread 本身的实现是 native 的，不同操作系统的实现不一样，例如在 Linux 上是用 pthread 实现的，在 Android 上当然也是 pthread。

Thread 可以算是并发运行中的一个基本单位了，下面介绍下 Java 中并发的写法。

2.1 ExecutorService

Java 为我们提供了 ExecutorService 接口用来管理并发的线程，先看一段代码：

public static void main(String[] args) {
    ExecutorService executorService = Executors.newCachedThreadPool();
    Runnable task = new Runnable() {
        public void run() {
            System.out.println("xxxx");
        }
    };
    Future<?> submit = executorService.submit(task);
    //submit.cancel(true);
}

submit 方法用于将 Runnable 放到线程池中，然后由线程池来管理，什么时候执行由线程池说了算，并且执行的时候不会阻塞当前的线程。Future 是 submit 方法创建的一个默认 Future，用来返回异步的结果，也可以用 Future 来取消对应线程的执行。

2.2 线程池

线程池实际上是一种管理线程运行的模式，具体的实现一般是通过一个队列来管理线程的执行顺序。

在前一节中 newCachedThreadPool 方法创建一个带缓存功能的线程池，用于重用 Thread 避免创建新的 Thread 带来的开销。Exectors 还为我们提供了其他常用的线程池，并且我们可以对这些线程池进行自定义：

newSingleThreadExecutor 大小为1的线程池，同时只运行一个线程
newScheduledThreadPool 可以添加 delay 的线程池
newFixedThreadPool 固定大小的线程池，可控制最大并发数，多出来的会放在队列中等待
newSingleThreadScheduledExecutor 看名字就知道是前两种的组合

Executors 中的方法还为我们提供了一些参数，用来定义线程池所使用的工作队列，线程池最大并发数等。当线程池被显式的 shutdown 后，其中运行的线程也会 shutdown。

在 Android 中绝大部分第三方库中的后台操作都是使用这些线程池实现的，同事谷歌在文档中也建议开发者将耗时的操作使用线程池来处理而不是用 AsyncTask。

3.Handler

Handler 有两个用途，一个是处理 Message 或者 Runnable 的回调，另一个是让别的线程执行一些操作。这些用途都是依赖于 Looper 这个类，这里我们看下 Looper 的原理和作用。

3.1 Looper

Looper 是一个和线程相关的东西，每一个线程有一个对应的 Looper，他用到了 ThreadLocal 这个概念。ThreadLocal 是线程内全局，线程间独立的一个东西，它用来保存那些你只想在线程内共享的变量，例如 Looper。Looper 在线程中创建了一个 MessageQueue，通过 loop 方法 Looper 会去遍历这个队列，将其中的 Message 取出来，并通过 Handler 将 Message 分发到 Handler 的 handleMessage 方法中。我们可以看下 Looper 的源码：

public final class Looper {
    
    ...
    static final ThreadLocal<Looper> sThreadLocal = new ThreadLocal<Looper>();
    private static Looper sMainLooper;  // guarded by Looper.class
    ...
    public static void prepare() {
        prepare(true);
    }
    private static void prepare(boolean quitAllowed) {
        if (sThreadLocal.get() != null) {
            throw new RuntimeException("Only one Looper may be created per thread");
        }
        sThreadLocal.set(new Looper(quitAllowed));
    }
}

这里只摘录一部分源码，可以看到 Looper 有一个 sMainLooper，这个是在主线程中创建的，另一个是 sThreadLocal，sThreadLocal 保存了当前线程的 Looper。sThreadLocal 的 get 方法会从当前线程中取出对应的 Looper。

public T get() {
    Thread t = Thread.currentThread();
    ThreadLocalMap map = getMap(t);
    if (map != null) {
        ThreadLocalMap.Entry e = map.getEntry(this);
        if (e != null)
            return (T)e.value;
    }
    return setInitialValue();
}

这两个变量是 static 的，所以当使用的时候 Looper 总是会保存一个 UI 线程和一个当前线程的 Looper，这其实也就是 Looper 的主要作用，用于别的线程和 UI 线程通信，最常见的用法就是在后台线程中获取数据，然后通过 handler 去更新 UI 线程。

总结下来就是每一个线程对应一个 Looper，当在线程中建立自己的 Looper 的时候需要先调用 prepare，如果想要发消息到 UI 线程，那么就 prepareMainLooper，如果是当前线程则 prepare。如果是在非主线程中使用的话在发出消息后需要手动调用 loop 去循环 MessageQueue 去发消息。通过这些分析我们可以看到这个东西的设计主要是为其它线程和 UI 通信而设计的。

3.2 Handler 和 Looper 的使用

使用很简单，这里就不赘述了，写两个例子，第一个是主线程中使用：

public class Main3Activity extends AppCompatActivity{
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        ...
         Handler h = new Handler(){
            @Override
            public void handleMessage(Message msg) {
                super.handleMessage(msg);
                // handle something
            }
        };
         h.sendEmptyMessage(0);
        ...
    }
}

别的线程中使用:

new Thread(new Runnable() {
    @Override
    public void run() {
        Looper.prepare();
        Handler h = new Handler(){
            @Override
            public void handleMessage(Message msg) {
                super.handleMessage(msg);
                // in this thread
                Looper.myLooper().quit();
            }
        };
        h.sendEmptyMessage(1);
        Looper.loop();
    }
}).start();

Looper.loop() 会阻塞当前线程，所以在使用完之后需要手动 quit，那么为什么 UI 线程中使用不需要呢，下面我们来看下源码。

3.3 Activity 中的 Looper

这个是 Looper 在 ActivityThread 中的使用，由于在 main 函数中使用了，所以我们在主线程中使用 Handler 去发消息的并没有手动调用，最后的 throw 也表示如果 loop 退出了则 UI 线程结束了。

class ActivityThread{
    public static void main(String[] args) {
        
        ...
        Looper.prepareMainLooper();

        ActivityThread thread = new ActivityThread();
        thread.attach(false);

        if (sMainThreadHandler == null) {
            sMainThreadHandler = thread.getHandler();
        }
        ...

        // End of event ActivityThreadMain.
        Trace.traceEnd(Trace.TRACE_TAG_ACTIVITY_MANAGER);
        Looper.loop();

        throw new RuntimeException("Main thread loop unexpectedly exited");
    }
}

4.AsyncTask

AsyncTask 是官方提供的一个后台线程和 UI 线程交互的工具类，这个类实际上是封装了前面说到的 ExecutorService。AsyncTask 中有两个 Executor：

/**
 * An {@link Executor} that can be used to execute tasks in parallel.
 */
public static final Executor THREAD_POOL_EXECUTOR
        = new ThreadPoolExecutor(CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE,
                TimeUnit.SECONDS, sPoolWorkQueue, sThreadFactory);

/**
 * An {@link Executor} that executes tasks one at a time in serial
 * order.  This serialization is global to a particular process.
 */
public static final Executor SERIAL_EXECUTOR = new SerialExecutor();
private static volatile Executor sDefaultExecutor = SERIAL_EXECUTOR;

THREAD_POOL_EXECUTOR 支持并发，也就是说多个线程可以同时运行，SERIAL_EXECUTOR 是按照顺序执行，每次执行一个，而且默认的是这个顺序执行的 Executor。文档中介绍的时候建议将这个东西用来处理比较短小的后台任务，一定程度上也是因为这个原因。由于这个 Executor 和队列是 static 的，所以一旦你在某处加了一个比较耗时的任务进来，可能会导致其他地方的任务一直处在等待状态，所以对于一些长时间的后台操作，更适合使用 Executor，ThreadPoolExecutor 等 Java 本身的东西来实现。

那么 THREAD_POOL_EXECUTOR 东西既然 new 出来，所以也不会没有用，AsyncTask 有一个 executeOnExecutor 方法，可以将这些 task 指定一个 Executor 来运行：

1	new AsyncTask().executeOnExecutor(AsyncTask.THREAD_POOL_EXECUTOR);

当这样使用的时候就会出现多个线程同时运行的现象。

这个类的另一个优势是可以操作 UI 线程。那么我们来看下他是怎么操作的：

private static class InternalHandler extends Handler {
    public InternalHandler() {
        super(Looper.getMainLooper());
    }

    @SuppressWarnings({"unchecked", "RawUseOfParameterizedType"})
    @Override
    public void handleMessage(Message msg) {
        AsyncTaskResult<?> result = (AsyncTaskResult<?>) msg.obj;
       switch (msg.what) {
            case MESSAGE_POST_RESULT:
                // There is only one result
                result.mTask.finish(result.mData[0]);
               break;
           case MESSAGE_POST_PROGRESS:
               result.mTask.onProgressUpdate(result.mData);
            break;
       }
    }
}

private void finish(Result result) {
      if (isCancelled()) {
          onCancelled(result);
      } else {
          onPostExecute(result);
      }
      mStatus = Status.FINISHED;
}

public final AsyncTask<Params, Progress, Result> execute(Params... params) {
    return executeOnExecutor(sDefaultExecutor, params);
}

public final AsyncTask<Params, Progress, Result> executeOnExecutor(Executor exec, Params... params) {
    ...

    mStatus = Status.RUNNING;

    onPreExecute();

    mWorker.mParams = params;
    exec.execute(mFuture);

    return this;
}

首先有四个方法在 UI 线程执行：

onPreExecute
onPostExecute
onProgressUpdate
onCancelled

在上面的代码中可以看到，onPreExecute 是放在 execute 中的，其他的是在 InternalHandler 中的，从 InternalHandler 的定义可以看出来他是 handle 在 UI 线程上的。到这里整个过程就比较清晰了，在后台运行中或者结束的时候去调用 handler 发送对应的 message 就可以调用到 UI 线程了。

由于在 3.0 之后默认使用 SERIAL_EXECUTOR，所以在 v4 包中有一个 AsyncTaskCompat，他会根据系统版本帮你创建一个使用 THREAD_POOL_EXECUTOR 的 AsyncTask。

5.RxJava 和 EventBus 中的 UI 线程

有了前面的铺垫，这里就变的好理解了很多。RxJava 和 EventBus 中都有让代码在 UI 线程和后台线程之间切换的能力，简单来说，当要使用 UI 线程的时候就用一个带 MainLooper 的 Handler 发消息就行了。在 EventBus 这一部分被封装成了一个 Poster，在 RxJava 中被封装了 HandlerScheduler。

// eventbus
mainThreadPoster = new HandlerPoster(this, Looper.getMainLooper(), 10);
final class HandlerPoster extends Handler {

    private final PendingPostQueue queue;
    private final int maxMillisInsideHandleMessage;
    private final EventBus eventBus;
    private boolean handlerActive;

    HandlerPoster(EventBus eventBus, Looper looper, int maxMillisInsideHandleMessage) {
    	// handle looper
        super(looper);
        this.eventBus = eventBus;
        this.maxMillisInsideHandleMessage = maxMillisInsideHandleMessage;
        queue = new PendingPostQueue();
    }
    ...
}


// RxJava
public final class AndroidSchedulers {
    private AndroidSchedulers() {
        throw new AssertionError("No instances");
    }

    // See https://github.com/ReactiveX/RxAndroid/issues/238
    // https://en.wikipedia.org/wiki/Initialization-on-demand_holder_idiom
    private static class MainThreadSchedulerHolder {
        static final Scheduler MAIN_THREAD_SCHEDULER =
                new HandlerScheduler(new Handler(Looper.getMainLooper()));
    }

    /** A {@link Scheduler} which executes actions on the Android UI thread. */
    public static Scheduler mainThread() {
        Scheduler scheduler =
                RxAndroidPlugins.getInstance().getSchedulersHook().getMainThreadScheduler();
        return scheduler != null ? scheduler : MainThreadSchedulerHolder.MAIN_THREAD_SCHEDULER;
    }
}

至于其他 io 或者是后台线程则是使用的前面说到的线程池实现。

6.小结

Android 中的并发其实和 Java 差不多，并且在 Android 4.0 之后也使用了 JSR-133 的规范，一个主要区别就是有了 Handler 用于操作 UI 线程的东西。在了解了以上东西后，我们可以使用线程池配合 Handler 来完成很多操作，也可以写自己的异步任务的库来满足日常的开发需求，也帮助我们更好的理解了第三方开源库对异步的处理。下一部分则是具体看下 JSR-133 规范中关于并发的描述，还有我们平时使用的 synchronization 等关键字的原理和用途。

References:
JSR-133
Java 中的进程与线程
 Java Thread Local