怎樣實現一個非阻塞的超時重試任務隊列

時間  2019-11-16
標籤 怎樣 實現 一個 阻塞 超時 重試 任務 隊列 简体版
原文   原文鏈接

原由

最近接手一個項目，要把其中的阻塞任務隊列，重構成非阻塞。在客戶端不多有機會直接處理任務隊列。項目完成須要總結經驗html

阻塞的發生

我這裏先說明我遇到的阻塞問題，我這裏的阻塞不是多線程訪問的阻塞，概念上是任務執行的阻塞。具體是：java

  • 任務開始客戶端準備數據，經過socket向服務器發送數據。
  • 阻塞等待服務器socket的ack迴應。
  • 獲得服務器的socket迴應完成任務，取出隊列的後續任務繼續執行。

這樣的阻塞隊列優勢就是：android

  • 代碼看起來很是簡潔且彙集，開始代碼對應結束代碼。
  • 邏輯上能夠保證任務的完成，由於若是沒有完成，就會阻塞直到任務的完成。

可是致命的缺點也是阻塞等待，由於直接的socket通訊使用是不保證送達，若是服務器一直沒有迴應，客戶端的任務隊列就一直阻塞在隊頭。除非經過其餘方式強制終止任務隊列。git

非阻塞的隊列

肯定了問題的發生的緣由，就能夠一步步的解決問題。 首先阻塞就是由於在等待迴應，只有迴應後才能完成任務。任務以本地客戶端開啓，以服務器迴應結束，期間阻塞。構成一個任務的概念。github

拆任務

其實客戶端沒必要執着等待迴應，只要把任務拆分紅服務器

  1. 發送任務
  2. 迴應任務

而期間再也不阻塞，只要迴應任務可以找到對應的發送任務，客戶端就能夠肯定該任務的完成。多線程

HandlerThread實現任務處理隊列

這裏socket的通訊確定是發生在子線程的，而子線程想要維護任務處理隊列，最好的方式就是直接使用HandlerThread，它封裝在子線程中Handler的配置，而Handler自己就是的任務處理隊列。app

package com.example.licola.myandroiddemo.java;

import android.os.Handler;
import android.os.HandlerThread;
import java.util.HashSet;

/** * Created by LiCola on 2018/4/10. * 簡化版非阻塞任務隊列 */
public class Dispatcher {

  private static final String THREAD_NAME="dispatcher-worker";

  private Handler mHandler;
  private HandlerThread handlerThread;

  private HashSet<String> tasks = new HashSet<>();//任務集合

  public void run(){
    handlerThread = new HandlerThread(THREAD_NAME);
    handlerThread.start();
    mHandler = new Handler(handlerThread.getLooper());
  }

  public void postSendTask(String id,String data){
    mHandler.post(new Runnable() {
      @Override
      public void run() {
         //發送任務的操做 如準備數據等

        tasks.add(id);
      }
    });
  }

  public void postAckTask(final String id){
    mHandler.post(new Runnable() {
      @Override
      public void run() {
        //迴應任務的操做 如解析迴應等

        tasks.remove(id);
      }
    });
  }
}

複製代碼

上面的代碼已經很是簡化，不涉及具體的任務處理，只有關鍵代碼。實現了前文的拆任務的理念。socket

可是拆任務也帶來了一個很嚴重的問題，任務怎樣保證完成。由於不阻塞，發送任務只管發送，發送完成迎來的多是下一個發送任務，而對應的迴應任務卻一直沒有到來。概念上這個任務始終沒有完成。代碼上就是tasks堆積愈來愈多等待迴應的任務。ide

超時機制

爲了應對可能堆積的tasks任務集合，就須要引入超時機制，就是給一個任務設定最長等待時間，若是超過這個時間尚未完成就重試。有了前面的代碼基礎加入超時檢測處理是很容易的。

  • 首先想到的就是在運行過程當中加入按期循環執行的檢測代碼。
  • 給發送任務加入時間變量，用於檢測超時。
  • 任務集合保存任務id和對應的發送數據，用於重試。

超時重試機制的任務隊列

package com.example.licola.myandroiddemo.java;

import android.os.Handler;
import android.os.HandlerThread;
import android.util.Pair;
import com.example.licola.myandroiddemo.utils.Logger;
import java.util.HashMap;
import java.util.Map.Entry;

/** * Created by LiCola on 2018/4/10. * 支持超時重試機制版非阻塞任務隊列 */
public class Dispatcher {

  private static final String THREAD_NAME = "dispatcher-worker";

  //超時檢測時間
  private static final long CHECK_ACK_TIME_OUT = 10 * 1000;
  //任務限定等待時間，即任務超時時間
  private static final long ACK_TIME_OUT = 4 * 1000;

  private Handler mHandler;
  private HandlerThread handlerThread;

  private HashMap<String, Pair<Long, String>> tasks=new HashMap<>();//任務集合

  public void run() {
    handlerThread = new HandlerThread(THREAD_NAME);
    handlerThread.start();
    mHandler = new Handler(handlerThread.getLooper());

    //開啓循環檢測
    mHandler.postDelayed(checkTimeOutTask(), CHECK_ACK_TIME_OUT);
  }

  public void postSendTask(final String id, final String data) {
    mHandler.post(new Runnable() {
      @Override
      public void run() {
        //發送任務的操做 如準備數據等

        Logger.d("開始發送任務");
        tasks.put(id, new Pair<>(System.currentTimeMillis(), data));
      }
    });
  }

  public void postAckTask(final String id) {
    mHandler.post(new Runnable() {
      @Override
      public void run() {
        //迴應任務的操做 如解析迴應等

        Logger.d("開始迴應任務");
        tasks.remove(id);
      }
    });
  }

  public Runnable checkTimeOutTask() {
    return new Runnable() {
      @Override
      public void run() {
        int count = 0;
        long curTime = System.currentTimeMillis();

        if (!tasks.isEmpty()) {
          for (Entry<String, Pair<Long, String>> entry : tasks.entrySet()) {
            String id = entry.getKey();
            Pair<Long, String> pair = entry.getValue();
            Long time = pair.first;
            String data = pair.second;
            if (curTime - time >= ACK_TIME_OUT) {
              postSendTask(id, data);
              count++;
            }
          }
        }

        if (count > 0) {
          Logger.d(String.format("檢測到超時任務%d", count));
        }

        //循環檢測
        mHandler.postDelayed(checkTimeOutTask(), CHECK_ACK_TIME_OUT);

      }
    };
  }
}

複製代碼

上面的代碼已經實現超時重試機制。仔細想一想這段代碼的運行狀況。仍是問題和有優化空間的。

檢測時機

仔細想一想按期檢測的時間和限定的超時時間，二者的關係。

//超時檢測時間
  private static final long CHECK_ACK_TIME_OUT = 10 * 1000;
  //任務限定等待時間，即任務超時時間
  private static final long ACK_TIME_OUT = 4 * 1000;
複製代碼

爲了檢測儘量的高效，且不影響整個任務隊列處理性能。讓檢測時間間隔比較大，且大於任務超時時間。 實際的運行狀況極可能以下圖所示：

運行時間圖

咱們以時間點check爲基準分析：

  1. check-1時間點以前：開始任務task-一、task-2。
  2. check-1時間點：檢測開始，發現任務集合中有2個等待任務，可是它們都沒有超時，沒有任何處理。
  3. check-2時間點以前：task-1正常完成，任務集合中刪除它。
  4. check-2時間定：檢測開始，發現任務集合中有1個等待任務，且已經超時。task-2任務重試，task-2的計時重置到當前時間點。

這是一種假設運行狀況，可是仍是暴露出了兩個問題：

  • 不夠高效：雖然檢測時間間隔足夠大，一個間隔內可以完成整個發送迴應的正常任務，可是檢測並無很高效，仍是在check-1時間點中觀察到了兩個不該該被觀察到的任務。其中task-1：它剛開始且能夠正常完成的。
  • 不精確的超時：在check-2以前任務task-2它已經超時了，可是在超時一段時間後才發現。

這兩個問題其實不嚴重，根據實際狀況選擇。 若是任務的超時小几率發生，且不要求精確的超時檢測。超時重試機制的任務處理隊列-非精確控制時間，仍是足夠知足開發需求的。

精確的控制超時時間

怎樣作到精確的控制超時時間，且讓檢測更高效。在Android開發中有沒有遇到精確控制任務時間的狀況，而其餘工程師們怎樣實現高效處理的。雖然咱們平常開發中沒有感知，可是這個狀況其實很是很是的廣泛存在。把這個問題換個角度：

怎樣精確的控制任務時間？

再想一想你開發的各類系統處理：

  • 長按點擊事件的監聽
  • ANR（Application Not Response）的檢測和發生

這兩個系統處理本質上就是精確控制任務時間的處理。

源碼的智慧

肯定了上面這兩個源碼目標，咱們來看看系統是怎樣實現的。

長按點擊事件

一個點擊的事件序列由ACTION_DOWN開始，後續的事件action不肯定。

開始：

任務的開始就是在View.onTouchEvent(MotionEvent event)的action事件處理cast:MotionEvent.ACTION_DOWN中的方法checkForLongClick(0, x, y) 核心代碼就一行：

private void checkForLongClick(int delayOffset, float x, float y) {
        if ((mViewFlags & LONG_CLICKABLE) == LONG_CLICKABLE || (mViewFlags & TOOLTIP) == TOOLTIP) {
		      
		      //發送延遲任務
            postDelayed(mPendingCheckForLongPress,
                    ViewConfiguration.getLongPressTimeout() - delayOffset);
        }
    }
複製代碼

結束：

點擊任務處理已經開始，而典型點擊任務結束就是ACTION_UP事件，一樣在代碼中cast:MotionEvent.ACTION_UP中的方法removeLongPressCallback()

private void removeLongPressCallback() {
        if (mPendingCheckForLongPress != null) {
            removeCallbacks(mPendingCheckForLongPress);
        }
    }
複製代碼

超時：

由於在開始就已經肯定固定時間點後執行超時處理，在這個時間點以前沒有其餘action操做來及時remove掉超時處理。從而超時處理獲得執行，具體就是執行長按事件。

private final class CheckForLongPress implements Runnable {
    
        @Override
        public void run() {
            if ((mOriginalPressedState == isPressed()) && (mParent != null)
                    && mOriginalWindowAttachCount == mWindowAttachCount) {
                if (performLongClick(mX, mY)) {
                    mHasPerformedLongPress = true;
                }
            }
        }
    }
複製代碼

ANR的檢查與發生

總所周知ANR的發生有不少種，這裏就挑Service的建立超時來舉例說明

Service Timeout:好比前臺服務在20s內未執行完成。

這裏參考理解Android ANR的觸發原理的分析流程。做者很形象的總結整個ANR檢測的理念：

埋炸彈-拆炸彈

由於ANR的處理比較複雜，咱們省略自動寫日誌和進程通訊等流程。

開始：埋炸彈

ActiveServices源碼部分

private final void realStartServiceLocked(ServiceRecord r, ProcessRecord app, boolean execInFg) throws RemoteException {
    ...
    //發送delay消息(SERVICE_TIMEOUT_MSG)
    bumpServiceExecutingLocked(r, execInFg, "create");
    try {
        ...
        //最終執行服務的onCreate()方法
        app.thread.scheduleCreateService(r, r.serviceInfo,
                mAm.compatibilityInfoForPackageLocked(r.serviceInfo.applicationInfo),
                app.repProcState);
    } catch (DeadObjectException e) {
        mAm.appDiedLocked(app);
        throw e;
    } finally {
        ...
    }
}
複製代碼
private final void bumpServiceExecutingLocked(ServiceRecord r, boolean fg, String why) {
    ... 
    scheduleServiceTimeoutLocked(r.app);
}

void scheduleServiceTimeoutLocked(ProcessRecord proc) {
    if (proc.executingServices.size() == 0 || proc.thread == null) {
        return;
    }
    long now = SystemClock.uptimeMillis();
    Message msg = mAm.mHandler.obtainMessage(
            ActivityManagerService.SERVICE_TIMEOUT_MSG);
    msg.obj = proc;
    
    //當超時後仍沒有remove該SERVICE_TIMEOUT_MSG消息，則執行service Timeout流程
    mAm.mHandler.sendMessageAtTime(msg,
        proc.execServicesFg ? (now+SERVICE_TIMEOUT) : (now+ SERVICE_BACKGROUND_TIMEOUT));
}
複製代碼

結束：拆炸彈

在Service的啓動前，已經埋下了炸彈，那就在啓動完成後拆掉炸彈。 ActiveServices源碼部分

private void serviceDoneExecutingLocked(ServiceRecord r, boolean inDestroying, boolean finishing) {
    ...
    if (r.executeNesting <= 0) {
        if (r.app != null) {
            r.app.execServicesFg = false;
            r.app.executingServices.remove(r);
            if (r.app.executingServices.size() == 0) {
                //當前服務所在進程中沒有正在執行的service
                mAm.mHandler.removeMessages(ActivityManagerService.SERVICE_TIMEOUT_MSG, r.app);
        ...
    }
    ...
}
複製代碼

超時：炸彈爆炸

若是Service沒有限定時間內完成啓動，拆掉炸彈，炸彈就會爆炸，就是超時任務執行。 就是ActiveService的serviceTimeout方法執行，寫下日誌發出ANR彈框。

總結

咱們從精確控制任務超時時間這角度，分析了長按事件和ANR的發生原理。最終發現他們都是基於一樣的設計方式：埋炸彈-拆炸彈 在任務開始時設置定時任務，及時完成remove掉定時任務，不然任務超時就會執行超時處理，而定時任務精確的時間執行就保證了超時任務精確控制。這個方式徹底不一樣於我前文實現的間隔檢測-非精確時間控制。

超時重試機制的任務隊列-精確控制時間

有對源碼的理解和總結，稍微修改代碼就能夠獲得以下

package com.example.licola.myandroiddemo.java;

import android.os.Handler;
import android.os.HandlerThread;
import com.example.licola.myandroiddemo.utils.Logger;
import java.util.HashMap;

/** * Created by LiCola on 2018/4/10. * 支持超時重試機制版非阻塞任務隊列 */
public class DispatcherTime {

  private static final String THREAD_NAME = "dispatcher-worker";

  //任務限定等待時間，即任務超時時間
  private static final long ACK_TIME_OUT = 2 * 1000;

  private Handler mHandler;
  private HandlerThread handlerThread;

  private HashMap<String, Runnable> timeoutTask = new HashMap<>();//超時集合

  public void run() {
    handlerThread = new HandlerThread(THREAD_NAME);
    handlerThread.start();
    mHandler = new Handler(handlerThread.getLooper());
  }

  public void postSendTask(final String id, final String data) {
    mHandler.post(new Runnable() {
      @Override
      public void run() {
        //發送任務的操做 如準備數據等

        Logger.d("開始發送任務",data);
        Runnable checkTimeOutTask = checkTimeOutTask(id, data);
        timeoutTask.put(id, checkTimeOutTask);
        mHandler.postDelayed(checkTimeOutTask,ACK_TIME_OUT);
      }
    });
  }

  public void postAckTask(final String id) {
    mHandler.post(new Runnable() {
      @Override
      public void run() {
        //迴應任務的操做 如解析迴應等

        Logger.d("開始迴應任務",id);
        Runnable runnable = timeoutTask.remove(id);
        mHandler.removeCallbacks(runnable);
      }
    });
  }

  public Runnable checkTimeOutTask(final String id, final String data) {
    return new Runnable() {
      @Override
      public void run() {

        Logger.d("超時任務執行 ",id,data);
        postSendTask(id, data);
      }
    };
  }
}
複製代碼

上面實現了每次任務發送都會埋下一個延遲任務，若是沒有及時獲得迴應就會重試。 這個實現的缺點若是要說的就是：

  • 每一個發送任務都會建立一個對應的延遲任務，若是發送任務數量較大，且只有小几率任務超時，就會產生大量建立的任務而又短時間存在且沒有機會執行的任務。

固然若是要優化就是使用Handler.handleMessage(Message msg)方法處理超時任務，而不是每次postDelayed都建立Runnable對象。這裏只留下思路就不用代碼了。

總結

  • 其實源碼的理解不是很難，只要找到切入點，從你關心的點出發，就可以理解源碼並應用它。咱們從超時任務的處理爲切入點就很容易理解長按事件的原理和ANR的發生機制。
  • 當咱們瞭解到一個新的解決方案，不要急於去應用它，要分析新方案的利弊，和咱們實際項目的匹配程度，才能很好的應用它和改造它。

參考

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程