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

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

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

public class Thread implements Runnable {

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

 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)

 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() {... }

MyTask mTask = new MyTask();

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

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

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ù)
    ...
   }
  };
 }

 //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;
 }

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

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) {
   ...
}

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

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

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

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

 //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;
 }

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

 //進(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類
  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);
 }

  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）
   ...
  }

//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.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);//交給線程池
  }

public final class AndroidSchedulers {

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

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