一种极简的异步超时处理机制设计与实现（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