淺談Android中多線程切換的幾種方法

更新時(shí)間：2018年01月02日 08:40:01 作者：藍(lán)灰_q

本篇文章主要介紹了淺談Android中多線程切換的幾種方法，小編覺(jué)得挺不錯(cuò)的，現(xiàn)在分享給大家，也給大家做個(gè)參考。一起跟隨小編過(guò)來(lái)看看吧

我們知道，多線程是Android開(kāi)發(fā)中必現(xiàn)的場(chǎng)景，很多原生API和開(kāi)源項(xiàng)目都有多線程的內(nèi)容，這里簡(jiǎn)單總結(jié)和探討一下常見(jiàn)的多線程切換方式。

我們先回顧一下Java多線程的幾個(gè)基礎(chǔ)內(nèi)容，然后再分析總結(jié)一些經(jīng)典代碼中對(duì)于線程切換的實(shí)現(xiàn)方式。

幾點(diǎn)基礎(chǔ)

多線程切換，大概可以切分為這樣幾個(gè)內(nèi)容：如何開(kāi)啟多個(gè)線程，如何定義每個(gè)線程的任務(wù)，如何在線程之間互相通信。

Thread

Thread可以解決開(kāi)啟多個(gè)線程的問(wèn)題。

Thread是Java中實(shí)現(xiàn)多線程的線程類，每個(gè)Thread對(duì)象都可以啟動(dòng)一個(gè)新的線程，注意是可以啟動(dòng)，也可以不啟動(dòng)新線程：

thread.run();//不啟動(dòng)新線程，在當(dāng)前線程執(zhí)行
thread.start();//啟動(dòng)新線程。

另外就是Thread存在線程優(yōu)先級(jí)問(wèn)題，如果為T(mén)hread設(shè)置較高的線程優(yōu)先級(jí)，就有機(jī)會(huì)獲得更多的CPU資源，注意這里也是有機(jī)會(huì)，優(yōu)先級(jí)高的Thread不是必然會(huì)先于其他Thread執(zhí)行，只是系統(tǒng)會(huì)傾向于給它分配更多的CPU。

默認(rèn)情況下，新建的Thread和當(dāng)前Thread的線程優(yōu)先級(jí)一致。

設(shè)置線程優(yōu)先級(jí)有兩種方式：

thread.setPriority(Thread.MAX_PRIORITY);//1~10，通過(guò)線程設(shè)置
Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND);//-20~19，通過(guò)進(jìn)程設(shè)置

這兩種設(shè)置方式是相對(duì)獨(dú)立的，在Android中，一般建議通過(guò)Process進(jìn)程設(shè)置優(yōu)先級(jí)。

ThreadPool

Thread本身是需要占用內(nèi)存的，開(kāi)啟/銷毀過(guò)量的工作線程會(huì)造成過(guò)量的資源損耗，這種場(chǎng)景我們一般會(huì)通過(guò)對(duì)資源的復(fù)用來(lái)進(jìn)行優(yōu)化，針對(duì)IO資源我們會(huì)做IO復(fù)用（例如Http的KeepAlive），針對(duì)內(nèi)存我們會(huì)做內(nèi)存池復(fù)用（例如Fresco的內(nèi)存池），針對(duì)CPU資源，我們一般會(huì)做線程復(fù)用，也就是線程池。

所以，在Android開(kāi)發(fā)中，一般不會(huì)直接開(kāi)啟大量的Thread，而是會(huì)使用ThreadPool來(lái)復(fù)用線程。

Runnable

Runnable主要解決如何定義每個(gè)線程的工作任務(wù)的問(wèn)題。
Runnable是Java中實(shí)現(xiàn)多線程的接口，相對(duì)Thread而言，Runnable接口更容易擴(kuò)展（不需要單繼承），而且，Thread本身也是一種Runnable：

public class Thread implements Runnable {

相比Thread而言，Runnable不關(guān)注如何調(diào)度線程，只關(guān)心如何定義要執(zhí)行的工作任務(wù)，所以在實(shí)際開(kāi)發(fā)中，多使用Runnable接口完成多線程開(kāi)發(fā)。

Callable

Callable和Runnable基本類似，但是Callable可以返回執(zhí)行結(jié)果。

線程間通信

Thread和Runnable能實(shí)現(xiàn)切換到另一個(gè)線程工作（Runnable需要額外指派工作線程），但它們完成任務(wù)后就會(huì)退出，并不注重如何在線程間實(shí)現(xiàn)通信，所以切換線程時(shí)，還需要在線程間通信，這就需要一些線程間通信機(jī)制。

Future

一般來(lái)說(shuō)，如果要做簡(jiǎn)單的通信，我們最常用的是通過(guò)接口回調(diào)來(lái)實(shí)現(xiàn)。

Future就是這樣一種接口，它可以部分地解決線程通信的問(wèn)題，F(xiàn)uture接口定義了done、canceled等回調(diào)函數(shù)，當(dāng)工作線程的任務(wù)完成時(shí)，它會(huì)（在工作線程中）通過(guò)回調(diào)告知我們，我們?cè)俨捎闷渌侄瓮ㄖ渌€程。

  mFuture = new FutureTask<MyBizClass>(runnable) {
   @Override
   protected void done() {
    ...//還是在工作線程里
   }
  };

Condition

Condition其實(shí)是和Lock一起使用的，但如果把它視為一種線程間通信的工具，也說(shuō)的通。

因?yàn)?，Condition本身定位就是一種多線程間協(xié)調(diào)通信的工具，Condition可以在某些條件下，喚醒等待線程。

 Lock lock = new ReentrantLock();
 Condition notFull = lock.newCondition(); //定義Lock的Condition
...
 while (count == items.length)
    notFull.await();//等待condition的狀態(tài)
...
 notFull.signal();//達(dá)到condition的狀態(tài)

Handler

其實(shí)，最完整的線程間通信機(jī)制，也是我們最熟悉的線程間通信機(jī)制，莫過(guò)于Handler通信機(jī)制，Handler利用線程封閉的ThreadLocal維持一個(gè)消息隊(duì)列，Handler的核心是通過(guò)這個(gè)消息隊(duì)列來(lái)傳遞Message，從而實(shí)現(xiàn)線程間通信。

AsyncTask的多線程切換

回顧完多線程的幾個(gè)基礎(chǔ)概念，先來(lái)看看簡(jiǎn)單的多線程切換，Android自帶的AsyncTask。

AsyncTask主要在doInBackground函數(shù)中定義工作線程的工作內(nèi)容，在其他函數(shù)中定義主線程的工作內(nèi)容，例如

onPostExecute，這里面必然涉及兩個(gè)問(wèn)題：
1.如何實(shí)現(xiàn)把doInBackground拋給工作線程
2.如何實(shí)現(xiàn)把onPostExecute拋給主線程

其實(shí)非常簡(jiǎn)單，我們先看第一個(gè)

1.如何實(shí)現(xiàn)把doInBackground拋給工作線程

在使用AsyncTask時(shí)，我們一般會(huì)創(chuàng)建一個(gè)基于AsyncTask的擴(kuò)展類或匿名類，在其中實(shí)現(xiàn)幾個(gè)抽象函數(shù)，例如：

 private class MyTask extends AsyncTask<String, Object, Long> {
  @Override
  protected void onPreExecute() {... }
  @Override
  protected Long doInBackground(String... params) {... }
  @Override
  protected void onProgressUpdate(Object... values) {... }
  @Override
  protected void onPostExecute(Long aLong) {... }
  @Override
  protected void onCancelled() {... }

然后，我們會(huì)實(shí)例化這個(gè)AsyncTask：

MyTask mTask = new MyTask();

在AsyncTask源碼中，我們看到，構(gòu)造函數(shù)里會(huì)創(chuàng)建一個(gè)WorkerRunnable：

 public AsyncTask() {
  mWorker = new WorkerRunnable<Params, Result>() {//這是一個(gè)Callable
   public Result call() throws Exception {
    ...
     result = doInBackground(mParams);//在工作線程中執(zhí)行
     ...

WorkerRunnable實(shí)際上是一個(gè)Callable對(duì)象，所以，doInBackground是被包在一個(gè)Callable對(duì)象中了，這個(gè)Callable還會(huì)被繼續(xù)包裝，最終被交給一個(gè)線程池去執(zhí)行：

Runnable mActive;
...
if ((mActive = mTasks.poll()) != null) {
 THREAD_POOL_EXECUTOR.execute(mActive);//交給線程池執(zhí)行
}

梳理一下，大致過(guò)程為：

定義doInBackground-->被一個(gè)Callable調(diào)用-->層層包為一個(gè)Runnable-->交給線程池執(zhí)行。

這樣就解決了第一個(gè)問(wèn)題，如何實(shí)現(xiàn)把doInBackground拋給工作線程。

我們?cè)賮?lái)看第二個(gè)問(wèn)題。

2.如何實(shí)現(xiàn)把onPostExecute拋給主線程

首先，我們要知道工作任務(wù)何時(shí)執(zhí)行完畢，就需要在工作完成時(shí)觸發(fā)一個(gè)接口回調(diào)，也就是前面說(shuō)過(guò)的Future，還是看AsyncTask源碼：

public AsyncTask() {
  mWorker = new WorkerRunnable<Params, Result>() {
   public Result call() throws Exception {
    ...
  };

  mFuture = new FutureTask<Result>(mWorker) {
   @Override
   protected void done() {//Future的回調(diào)
    try {
     postResultIfNotInvoked(get());//get()是FutureTask接口函數(shù)
    ...
   }
  };
 }

這樣，我們就知道可以處理onPostExecute函數(shù)了，但是，我們還需要把它拋給主線程，主要源碼如下：

 //mWorker、mFuture和都會(huì)指向postResult函數(shù)
 private Result postResult(Result result) {
  @SuppressWarnings("unchecked")
  Message message = getHandler().obtainMessage(MESSAGE_POST_RESULT,
    new AsyncTaskResult<Result>(this, result));
  message.sendToTarget();
  return result;
 }
 //getHandler()會(huì)指向InternalHandler
 private static class InternalHandler extends Handler {
  public InternalHandler() {
   super(Looper.getMainLooper());//指向MainLooper
  }
  @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]);//通過(guò)handler機(jī)制，回到主線程，調(diào)用finish函數(shù)
  ...
 }
 //在Handler中，最終會(huì)在主線程中調(diào)用finish
 private void finish(Result result) {
  if (isCancelled()) {
   onCancelled(result);
  } else {
   onPostExecute(result);//調(diào)用onPostExecute接口函數(shù)
  }
  mStatus = Status.FINISHED;
 }

從源碼可以看到，其實(shí)AsyncTask還是通過(guò)Handler機(jī)制，把任務(wù)拋給了主線程。

總體來(lái)說(shuō)，AsyncTask的多線程任務(wù)是通過(guò)線程池實(shí)現(xiàn)的工作線程，在完成任務(wù)后利用Future的done回調(diào)來(lái)通知任務(wù)完成，并通過(guò)handler機(jī)制通知主線程去執(zhí)行onPostExecute等回調(diào)函數(shù)。

EventBus的多線程切換

EventBus會(huì)為每個(gè)訂閱事件注冊(cè)一個(gè)目標(biāo)線程，所以需要從發(fā)布事件的線程中，根據(jù)注冊(cè)信息，實(shí)時(shí)切換到目標(biāo)線程中，所以，這是個(gè)很典型的多線程切換場(chǎng)景。
根據(jù)EventBus源碼，多線程切換的主要判斷代碼如下：

  switch (subscription.subscriberMethod.threadMode) {
   case POSTING:
    invokeSubscriber(subscription, event);//直接在當(dāng)前線程執(zhí)行
    break;
   case MAIN:
    if (isMainThread) {
     invokeSubscriber(subscription, event);//在當(dāng)前主線程執(zhí)行
    } else {
     mainThreadPoster.enqueue(subscription, event);//當(dāng)然不是主線程，交給主線程執(zhí)行
    }
    break;
   case BACKGROUND:
    if (isMainThread) {
     backgroundPoster.enqueue(subscription, event);//當(dāng)前線程為主線程，交給工作線程
    } else {
     invokeSubscriber(subscription, event);//直接在當(dāng)前工作線程執(zhí)行
    }
    break;
   case ASYNC:
    asyncPoster.enqueue(subscription, event);//異步執(zhí)行
    break;
   default:
    throw new IllegalStateException("Unknown thread mode: " + subscription.subscriberMethod.threadMode);
  }

所以，在EventBus里，如果需要做線程間切換，主要是拋給不同的任務(wù)隊(duì)列，實(shí)現(xiàn)線程間切換。

從任務(wù)隊(duì)列判斷，切換目標(biāo)包括主線程Poster、backgroundPoster和asyncPoster這樣三種。

我們先看任務(wù)隊(duì)列的設(shè)計(jì)：

任務(wù)隊(duì)列

因?yàn)镋ventBus不能判斷有哪些任務(wù)會(huì)并行，所以它采用了隊(duì)列的設(shè)計(jì)，多線程任務(wù)（EventBus的事件）會(huì)先進(jìn)入隊(duì)列，然后再處理隊(duì)列中的工作任務(wù)，這是典型的生產(chǎn)--消費(fèi)場(chǎng)景。
主線程Poster、backgroundPoster和asyncPoster都是任務(wù)隊(duì)列的不同實(shí)現(xiàn)。

主線程Poster

負(fù)責(zé)處理主線程的mainThreadPoster是Handler的子類：

final class HandlerPoster extends Handler {
...
 void enqueue(Subscription subscription, Object event) {
  ...
  synchronized (this) {//因?yàn)橹骶€程只有一個(gè)，需要線程安全
   queue.enqueue(pendingPost);
   ...
    if (!sendMessage(obtainMessage())) {//作為handler發(fā)送消息
     ...
 //在主線程中處理消息
 @Override
 public void handleMessage(Message msg) {
   ...
}

從源碼可以看出，這個(gè)Poster其實(shí)是一個(gè)Handler，它采用了哪個(gè)線程的消息隊(duì)列，就決定了它能和哪個(gè)線程通信，我們確認(rèn)一下：

 EventBus(EventBusBuilder builder) {
  ...
  mainThreadPoster = new HandlerPoster(this, Looper.getMainLooper(), 10);//獲取主線程的MainLooper

所以，EventBus是擴(kuò)展了一個(gè)Handler，作為主線程的Handler，通過(guò)Handler消息機(jī)制實(shí)現(xiàn)的多線程切換。

當(dāng)然，這個(gè)Handler本事，又多了一層queue。

backgroundPoster和asyncPoster

backgroundPoster和asyncPoster其實(shí)都是使用了EventBus的線程池，默認(rèn)是個(gè)緩存線程池：

private final static ExecutorService DEFAULT_EXECUTOR_SERVICE = Executors.newCachedThreadPool();

所以，backgroundPoster和asyncPoster都是把任務(wù)交給線程池處理，這樣實(shí)現(xiàn)的多線程切換。

不過(guò)，backgroundPoster和asyncPoster也有一些不同，我們知道，在newCachedThreadPool中，最大線程數(shù)就是Integer的最大值，相當(dāng)于不設(shè)上限，所以可以盡可能多的啟動(dòng)線程，asyncPoster就是這樣做的，enqueu和run都沒(méi)做同步，為每個(gè)事件單獨(dú)開(kāi)啟新線程處理。

而在backgroundPoster中，可以盡量復(fù)用線程，主要方法是在run的時(shí)候，做個(gè)1秒的等待：

 @Override
 public void run() { 
  ...
  PendingPost pendingPost = queue.poll(1000);//允許等待1秒

因?yàn)樽隽诉@一秒的掛起等待，在enqueue和run時(shí)，都需要用synchronized (this) 來(lái)確保線程安全。

另外，其實(shí)這里面還有個(gè)很重要的用法，就是Executors.newCachedThreadPool()中的SynchronousQueue：

 public static ExecutorService newCachedThreadPool() {
  return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
          60L, TimeUnit.SECONDS,
          new SynchronousQueue<Runnable>());//用于輔助線程切換的阻塞隊(duì)列
 }

這個(gè)SynchronousQueue，在OkHttp中也使用了：

 //okhttp3.Dispatcher源碼
 public synchronized ExecutorService executorService() {
 if (executorService == null) {
  executorService = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
   new SynchronousQueue<Runnable>(), Util.threadFactory("OkHttp Dispatcher", false));//用于輔助線程切換的阻塞隊(duì)列
 }
 return executorService;
 }

SynchronousQueue與普通隊(duì)列不同，不是數(shù)據(jù)等線程，而是線程等數(shù)據(jù)，這樣每次向SynchronousQueue里傳入數(shù)據(jù)時(shí)，都會(huì)立即交給一個(gè)線程執(zhí)行，這樣可以提高數(shù)據(jù)得到處理的速度。

總的來(lái)看，EventBus還是采用線程池實(shí)現(xiàn)工作線程，采用handler機(jī)制通知到主線程。不同在于，它采用的queue的隊(duì)列方式來(lái)管理所有的跨線程請(qǐng)求，而且它利用了SynchronousQueue阻塞隊(duì)列來(lái)輔助實(shí)現(xiàn)線程切換。

RxJava的多線程切換

其實(shí)在多線程管理這方面，RxJava的線程管理能力是非常令人贊嘆的。
RxJava的主要概念是工作流，它可以把一序列工作流定義在一個(gè)線程類型上：

myWorkFlow.getActResponse(myParam)
    .subscribeOn(Schedulers.io())//指定線程
    .xxx//其他操作

這個(gè)構(gòu)建工作流的過(guò)程其實(shí)挺復(fù)雜的，不過(guò)如果我們只看線程操作這部分，其實(shí)流程非常清晰，我們追蹤一下subscribeOn的源碼（RxJava2）：

 //進(jìn)入subscribeOn
 public final Flowable<T> subscribeOn(@NonNull Scheduler scheduler) {
  ObjectHelper.requireNonNull(scheduler, "scheduler is null");
  return subscribeOn(scheduler, !(this instanceof FlowableCreate));
 }
 //繼續(xù)進(jìn)入subscribeOn
 public final Flowable<T> subscribeOn(@NonNull Scheduler scheduler, boolean requestOn) {
  ObjectHelper.requireNonNull(scheduler, "scheduler is null");
  return RxJavaPlugins.onAssembly(new FlowableSubscribeOn<T>(this, scheduler, requestOn));
 }

然后，進(jìn)入FlowableSubscribeOn類

 //進(jìn)入FlowableSubscribeOn類
  public FlowableSubscribeOn(Flowable<T> source, Scheduler scheduler, boolean nonScheduledRequests) {
  ...
  this.scheduler = scheduler;
  ...
 }

 @Override
 public void subscribeActual(final Subscriber<? super T> s) {
  Scheduler.Worker w = scheduler.createWorker();//根據(jù)參數(shù)值，如Schedulers.io()創(chuàng)建worker
  final SubscribeOnSubscriber<T> sos = new SubscribeOnSubscriber<T>(s, w, source, nonScheduledRequests);//根據(jù)worker創(chuàng)建SubscribeOnSubscriber
  s.onSubscribe(sos);
  w.schedule(sos);
 }

這個(gè)SubscribeOnSubscriber是個(gè)內(nèi)部類：

  SubscribeOnSubscriber(Subscriber<? super T> actual, Scheduler.Worker worker, Publisher<T> source, boolean requestOn) {
   ...
   this.worker = worker;
   ...
   }
...
  void requestUpstream(final long n, final Subscription s) {
   ...
    worker.schedule(new Request(s, n));//worker會(huì)安排如何執(zhí)行runnable（Request是一個(gè)runnable）
   ...
  }

而這個(gè)worker，其實(shí)就是我們輸入的線程參數(shù)，如Schedulers.io()，這個(gè)io是這樣定義的：

//io.reactivex.schedulers.Schedulers源碼
 static {
  SINGLE = RxJavaPlugins.initSingleScheduler(new SingleTask());

  COMPUTATION = RxJavaPlugins.initComputationScheduler(new ComputationTask());

  IO = RxJavaPlugins.initIoScheduler(new IOTask());

  TRAMPOLINE = TrampolineScheduler.instance();

  NEW_THREAD = RxJavaPlugins.initNewThreadScheduler(new NewThreadTask());
 }
...
 static final class IOTask implements Callable<Scheduler> {
  @Override
  public Scheduler call() throws Exception {
   return IoHolder.DEFAULT;
  }
 }

 static final class NewThreadTask implements Callable<Scheduler> {
  @Override
  public Scheduler call() throws Exception {
   return NewThreadHolder.DEFAULT;
  }
 }

 static final class SingleTask implements Callable<Scheduler> {
  @Override
  public Scheduler call() throws Exception {
   return SingleHolder.DEFAULT;
  }
 }

 static final class ComputationTask implements Callable<Scheduler> {
  @Override
  public Scheduler call() throws Exception {
   return ComputationHolder.DEFAULT;
  }
 }
...
 static final class SingleHolder {
  static final Scheduler DEFAULT = new SingleScheduler();
 }

 static final class ComputationHolder {
  static final Scheduler DEFAULT = new ComputationScheduler();
 }

 static final class IoHolder {
  static final Scheduler DEFAULT = new IoScheduler();
 }

 static final class NewThreadHolder {
  static final Scheduler DEFAULT = new NewThreadScheduler();
 }

這里的IO，最終會(huì)指向一個(gè)Scheduler，如IoScheduler：

//io.reactivex.internal.schedulers.IoScheduler源碼
...
 static final class EventLoopWorker extends Scheduler.Worker {//Scheduler.Worker的實(shí)現(xiàn)類
  ...
  @NonNull
  @Override
  public Disposable schedule(@NonNull Runnable action, long delayTime, @NonNull TimeUnit unit) {
   if (tasks.isDisposed()) {
    // don't schedule, we are unsubscribed
    return EmptyDisposable.INSTANCE;
   }

   return threadWorker.scheduleActual(action, delayTime, unit, tasks);//交給線程池
  }

這樣，Scheculer中的具體任務(wù)就交給了某個(gè)線程池來(lái)處理。

需要特別說(shuō)明的是，RxJava中調(diào)用Android主線程(AndroidSchedulers.mainThread)，其實(shí)還是使用了Handler機(jī)制：

public final class AndroidSchedulers {

 ...
  static final Scheduler DEFAULT = new HandlerScheduler(new Handler(Looper.getMainLooper()));

這個(gè)HandlerScheduler其實(shí)就是實(shí)現(xiàn)了Scheduler和Scheduler.Worker內(nèi)部類。

···
final class HandlerScheduler extends Scheduler {
private final Handler handler;

HandlerScheduler(Handler handler) {
 this.handler = handler;
}
private static final class HandlerWorker extends Worker {
 ...
 @Override
 public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
  ...
  handler.sendMessageDelayed(message, Math.max(0L, unit.toMillis(delay)));
···

總的來(lái)看，RxJava的多線程切換其實(shí)是利用了Scheculer.Worker這個(gè)內(nèi)部類，把任務(wù)交給Scheculer的Worker去做，而這個(gè)Scheculer的Worker是根據(jù)定義的線程來(lái)實(shí)現(xiàn)了不同的線程池，其實(shí)還是交給線程池去處理了。
至于主線程，RxJava也是使用了Handler機(jī)制。

總結(jié)

小小總結(jié)一下，基本上來(lái)說(shuō)，Android中的多線程切換，主要使用Runnable和Callable來(lái)定義工作內(nèi)容，使用線程池來(lái)實(shí)現(xiàn)異步并行，使用Handler機(jī)制來(lái)通知主線程，有些場(chǎng)景下會(huì)視情況需要，使用Future的接口回調(diào)，使用SynchronousQueue阻塞隊(duì)列等。

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: