一種極簡的異步超時處理機制設計與實現（C#版）

1.引言

當執行某些動做以後，會期待反饋。最終要麼是獲得告終果，要麼就是超時了。當超時發生時，多是指望獲得通知，或是但願能自動重試，等等。因而設計了一種通用的異步超時的處理機制，以期經過簡潔易理解的方式完成超時的處理過程。

2.對外接口設計

從使用的角度，調用方指望的是「指定超時時長，時間到自動執行指定過程」，由此能夠得出外部的操做接口參數。從功能角度來看，對於未超時的狀況，須要提供在超時時長內隨時清除超時任務的功能。

2.1操做接口

在這裏，咱們把設計的機制稱裏「超時任務運行器」，從外部來看，其接口與功能結構以下：
（1）添加超時任務，帶上對象標識用於回調時傳參，指定超時時長和超時回調方法便可把超時處理交給超時任務運行器。返回一個任務標識，用於後續可刪除該超時任務。
（2）刪除超時任務，指定任務標識便可刪除。同時，支持清除一個對象的全部超時任務。

2.2使用過程

發起異步操做的同時，添加超時任務，在異步操做成功時，刪除超時任務。超時則運行器會自動執行超時任務。以下圖，灰色部分由運行器完成：

3.超時任務運行器設計與實現

首先，時長的精確粒度設定爲秒，這表示超時最低可支持秒級（廢話）。設計的基本思路爲：對於加入的超時任務，運行器創建清單，並以秒爲單位對清單中的任務進行檢測，對於時間已經到達或超過的將其移動至超時任務執行隊列中，由獨立的超時任務運行線程來執行隊列中的任務。這裏，移動任務至執行隊列的檢測者謂之「生產者」，任務執行線程謂之「消費者」。

3.1基本結構

運行器維護一個超時任務清單和一個執行隊列，一個超時檢測者，它使用定時器檢測任務是否超時並將超時的加入執行隊列，一個任務執行者負責運行超時任務中的回調過程。

3.2數據結構

超時任務信息，除了調用者傳遞的對象標識、超時時長與回調方法，還包括其它運行過程當中所需的屬性：任務標識、運行時間點。同時爲了在回調時能利用到一些對象相關的上下文信息，再增長一個 context 屬性，它將做爲參數傳遞給回調函數，詳細信息看以下類定義：

/// <summary>
    /// 超時回調的委託
    /// </summary>
    /// <typeparam name="T"></typeparam>
    /// <param name="objectKey"></param>
    /// <param name="context"></param>
    public delegate void TimeoutCallback<T>(T objectKey, String context);

    /// <summary>
    /// 超時任務信息
    /// </summary>
    /// <typeparam name="T"></typeparam>
    public class TimeoutTask<T>
    {
        // 任務標識（由超時任務運行器自動分配）
        public long TaskId { get; set; }
        // 對象標識
        public T ObjectKey { get; set; }
        // 超時秒數
        public int TimeoutSeconds { get; set; }
        /// <summary>
        /// 以秒爲單位的 Tick 值，由超時任務運行器根據當前時間加上超時秒數計算設置
        /// DateTime.Ticks 是以 10ns（10納秒） 爲單位
        /// 將其除以 10 單位爲 ws（微秒），再除以 1000 爲 ms（毫秒），再除以 1000 爲 s（秒）
        /// 累計爲 DateTime.Ticks / 10000000
        /// </summary>
        public long ExecuteSecondTicks { get; set; }
        // 超時回調方法
        public TimeoutCallback<T> Callback { get; set; }
        /// <summary>
        /// 用於保存一些回調時使用的上下文信息
        /// </summary>
        public String Context { get; set; }
    }

3.3超時任務清單

任務清單，在操做粒度上，能夠以任務標識爲單位，也能夠以對象標識爲單位，所以，爲了快速檢索。任務清單分兩種形式存儲，一種以任務標識爲主鍵，另外一種以對象標識爲主鍵，其結構以下：

具體類型結構定義以下，_DictionaryLocker 有於同步加鎖，確保線程安全。

// 以 TaskId(任務標識) 爲 KEY 的任務清單字典
    private Dictionary<long, TimeoutTask<T>> _TaskIdDictionary = new Dictionary<long, TimeoutTask<T>>();
    // 以 ObjectId(任務相關對象標識) 爲 KEY 的任務字典，因每一個對象能夠有多個超時任務，因此爲列表
    private Dictionary<T, List<TimeoutTask<T>>> _TaskObjectKeyDictionary = new Dictionary<T, List<TimeoutTask<T>>>();
    // 用於同步操做上述兩個清單字典，使得線程安全
    private object _DictionaryLocker = new object();

3.4任務執行隊列

一個普通的先進先出的隊列，_RunLocker 用於線程安全加鎖。

// 已超時任務隊列，由任務運行線程逐個執行
    private Queue<TimeoutTask<T>> _TaskRunQueue = new Queue<TimeoutTask<T>>();
    // 用來同步操做任務隊列，使得線程安全（生產者，消費者模式）
    private object _RunLocker = new object();

3.5超時檢測者

以每秒進行一次檢測的粒度運行，使用 System.Timers.Timer 很是合適，它的職能是判斷運行時間到達與否決定是否將任務移至執行隊列。

// 超時檢測者，每秒掃描是否達到超時，超時則加入超時任務隊列
    private System.Timers.Timer _TimeoutChecker = new System.Timers.Timer();

    // 超時檢測者
    _TimeoutChecker.Interval = 1000;
    _TimeoutChecker.Elapsed += new System.Timers.ElapsedEventHandler(CheckTimerTick);
    _TimeoutChecker.Start();

    /// <summary>
    /// 超時任務檢測者
    /// 對於，時間已經超過了設定的超時時間的，加入超時任務執行隊列
    /// </summary>
    /// <param name="sender"></param>
    /// <param name="e"></param>
    private void CheckTimerTick(object sender, System.Timers.ElapsedEventArgs e)
    {
        long secondTicks = DateTime.Now.Ticks / 10000000;
        // 遍歷，把時間已到達超過超時時間的找出來
        lock (_DictionaryLocker)
        {
            foreach (var key in _TaskIdDictionary.Keys.ToList())
            {
                var task = _TaskIdDictionary[key];
                if (_TaskIdDictionary[key].ExecuteSecondTicks <= secondTicks)
                {
                    // 加入超時任務執行隊列，並移除清單
                    lock (_RunLocker)
                    {
                        _TaskRunQueue.Enqueue(task);
                        RemoveTimeoutTask(task.TaskId);
                    }
                    // 有生產，則通知執行線程（消費者）
                    _WaitHandle.Set();
                }
            }
        }
    }

3.6任務執行者

執行隊列中存在任務時就執行，不然等待。線程等待，這裏使用了 EventWaitHandle，EventWaitHandle.WaitOne 等待，生產者使用 EventWaitHandle.Set 方法進行通知，配合起來有效地運行隊列中的任務。

// 超時任務執行線程
    private Thread _TaskRunThread;
    // 用於同步操做任務隊列的線程信號（生產者，消費者通知做用）
    private EventWaitHandle _WaitHandle = new AutoResetEvent(false);
    // 用於退出執行線程的一個標識
    private bool _Working = true;

    /// <summary>
    /// 超時任務執行線程主體
    /// </summary>
    private void TaskRunning()
    {
        while (_Working)
        {
            TimeoutTask<T> task = null;
            lock (_RunLocker)   
            {
                if (_TaskRunQueue.Count > 0)
                {
                    task = _TaskRunQueue.Dequeue();  
                }
            }
            // 存在超時任務執行其回調
            if (task != null)
            {
                task.Callback(task.ObjectKey, task.Context);
            }
            else
            {
                // 等待生產者通知
                _WaitHandle.WaitOne();
            }
        }
    }

3.7向外開放的接口

代碼如是說：

/// <summary>
    /// 指定對象標識，超時時長（秒爲單位），超時執行回調，加入到超時檢測字典中
    /// </summary>
    /// <param name="objectKey"></param>
    /// <param name="timeoutSeconds"></param>
    /// <param name="callback"></param>
    /// <param name="context"></param>
    /// <returns></returns>
    public long AddTimeoutTask(T objectKey, int timeoutSeconds, TimeoutCallback<T> callback, String context)
    {
        TimeoutTask<T> task = new TimeoutTask<T>();
        task.ObjectKey = objectKey;
        task.TimeoutSeconds = timeoutSeconds;
        task.Callback = callback;
        long taskId = GetNextTaskId();
        task.TaskId = taskId;
        task.ExecuteSecondTicks = DateTime.Now.Ticks / 10000000 + timeoutSeconds;
        task.Context = context;

        lock (_DictionaryLocker)
        {
            // 以任務標識爲主鍵的任務清單
            _TaskIdDictionary[taskId] = task;
            // 以對象標識爲主鍵的任務清單
            if (_TaskObjectKeyDictionary.ContainsKey(objectKey))
            {
                _TaskObjectKeyDictionary[objectKey].Add(task);
            }
            else
            {
                List<TimeoutTask<T>> list = new List<TimeoutTask<T>>();
                list.Add(task);
                _TaskObjectKeyDictionary[objectKey] = list;
            }
        }
        return taskId;
    }

    /// <summary>
    /// 根據對象標識移除超時任務設置
    /// </summary>
    /// <param name="objectKey"></param>
    public void RemoveTimeoutTask(T objectKey)
    {
        lock (_DictionaryLocker)
        {
            if (_TaskObjectKeyDictionary.ContainsKey(objectKey))
            {
                // 在任務標識爲主鍵的清單中移除相應的該對象的多個超時任務
                foreach (var task in _TaskObjectKeyDictionary[objectKey])
                {
                    _TaskIdDictionary.Remove(task.TaskId);
                }
                _TaskObjectKeyDictionary[objectKey].Clear();
            }
        }
    }

    /// <summary>
    /// 根據任務標識移除超時任務設置
    /// </summary>
    /// <param name="taskId"></param>
    public void RemoveTimeoutTask(long taskId)
    {
        lock (_DictionaryLocker)
        {
            if (_TaskIdDictionary.ContainsKey(taskId))
            {
                var task = _TaskIdDictionary[taskId];
                _TaskIdDictionary.Remove(taskId);
                // 在對象標識爲主鍵的清單移除相應的超時任務
                _TaskObjectKeyDictionary[task.ObjectKey].Remove(task);
            }
        }
    }

4.應用示例

定義回調處理方法，添加一個超時任務只須要指定簡單的參數便可，以下示例，會按什麼順序輸出什麼呢？

class Program
{
    static void Main(string[] args)
    { 
        TS.Task.TimeoutTaskRunner<string> runner = new TS.Task.TimeoutTaskRunner<string>();

        TS.Task.TimeoutCallback<string> callback = (string key, string context) =>
        {
            Console.WriteLine(key + " is timeout.");
        }; 

        runner.AddTimeoutTask("a", 4, callback, null);
        runner.AddTimeoutTask("b", 3, callback, null);
        runner.AddTimeoutTask("c", 2, callback, null); 

        Console.ReadKey();
        runner.Dispose();
    }
}

運行結果：

5.小結

超時處理在異步通訊中常常會碰到，實現超時處理的通用機制，能有效的複用代碼，提升效率。代碼仍然有不少優化空間，如遍歷檢測超時是否有更合適的的方式等，歡迎探討！

完整代碼請訪問：

https://github.com/triplestudio/TimeoutTask