在我們業務操作時，難免會有多次操作，比如：界面搜索框，輸入內容時時查詢數據庫/后臺數據。多次觸發搜索，我們期望什么結果呢？絕大部分情況，應該是只需要最后一次操作的結果，其它操作應該無效。

自定義等待的任務類

1. 可等待的任務類 AwaitableTask:

/// <summary>
    /// 可等待的任務
    /// </summary>
    public class AwaitableTask
    {
        /// <summary>
        /// 獲取任務是否為不可執行狀態
        /// </summary>
        public bool NotExecutable { get; private set; }


        /// <summary>
        /// 設置任務不可執行
        /// </summary>
        public void SetNotExecutable()
        {
            NotExecutable = true;
        }


        /// <summary>
        /// 獲取任務是否有效
        /// 注：對無效任務，可以不做處理。減少并發操作導致的干擾
        /// </summary>
        public bool IsInvalid { get; private set; } = true;


        /// <summary>
        /// 標記任務無效
        /// </summary>
        public void MarkTaskValid()
        {
            IsInvalid = false;
        }


        #region Task


        private readonly Task _task;
        /// <summary>
        /// 初始化可等待的任務。
        /// </summary>
        /// <param name="task"></param>
        public AwaitableTask(Task task) => _task = task;


        /// <summary>
        /// 獲取任務是否已完成
        /// </summary>
        public bool IsCompleted => _task.IsCompleted;


        /// <summary>
        /// 任務的Id
        /// </summary>
        public int TaskId => _task.Id;


        /// <summary>
        /// 開始任務
        /// </summary>
        public void Start() => _task.Start();


        /// <summary>
        /// 同步執行開始任務
        /// </summary>
        public void RunSynchronously() => _task.RunSynchronously();


        #endregion


        #region TaskAwaiter


        /// <summary>
        /// 獲取任務等待器
        /// </summary>
        /// <returns></returns>
        public TaskAwaiter GetAwaiter() => new TaskAwaiter(this);


        /// <summary>Provides an object that waits for the completion of an asynchronous task. </summary>
        [HostProtection(SecurityAction.LinkDemand, ExternalThreading = true, Synchronization = true)]
        public struct TaskAwaiter : INotifyCompletion
        {
            private readonly AwaitableTask _task;


            /// <summary>
            /// 任務等待器
            /// </summary>
            /// <param name="awaitableTask"></param>
            public TaskAwaiter(AwaitableTask awaitableTask) => _task = awaitableTask;


            /// <summary>
            /// 任務是否完成.
            /// </summary>
            public bool IsCompleted => _task._task.IsCompleted;


            /// <inheritdoc />
            public void OnCompleted(Action continuation)
            {
                var This = this;
                _task._task.ContinueWith(t =>
                {
                    if (!This._task.NotExecutable) continuation?.Invoke();
                });
            }
            /// <summary>
            /// 獲取任務結果
            /// </summary>
            public void GetResult() => _task._task.Wait();
        }


        #endregion


    }

無效的操作可以分為以下倆種：

已經進行中的操作，后續結果應標記為無效

還沒開始的操作，后續不執行

自定義任務類型 AwaitableTask中，添加倆個字段NotExecutable、IsInvalid：

/// <summary>
    /// 獲取任務是否為不可執行狀態
    /// </summary>
    public bool NotExecutable { get; private set; }
    /// <summary>
    /// 獲取任務是否有效
    /// 注：對無效任務，可以不做處理。減少并發操作導致的干擾
    /// </summary>
    public bool IsInvalid { get; private set; } = true;

2. 有返回結果的可等待任務類 AwaitableTask<TResult>：

/// <summary>
    /// 可等待的任務
    /// </summary>
    /// <typeparam name="TResult"></typeparam>
    public class AwaitableTask<TResult> : AwaitableTask
    {
        private readonly Task<TResult> _task;
        /// <summary>
        /// 初始化可等待的任務
        /// </summary>
        /// <param name="task">需要執行的任務</param>
        public AwaitableTask(Task<TResult> task) : base(task) => _task = task;


        #region TaskAwaiter


        /// <summary>
        /// 獲取任務等待器
        /// </summary>
        /// <returns></returns>
        public new TaskAwaiter GetAwaiter() => new TaskAwaiter(this);


        /// <summary>
        /// 任務等待器
        /// </summary>
        [HostProtection(SecurityAction.LinkDemand, ExternalThreading = true, Synchronization = true)]
        public new struct TaskAwaiter : INotifyCompletion
        {
            private readonly AwaitableTask<TResult> _task;


            /// <summary>
            /// 初始化任務等待器
            /// </summary>
            /// <param name="awaitableTask"></param>
            public TaskAwaiter(AwaitableTask<TResult> awaitableTask) => _task = awaitableTask;


            /// <summary>
            /// 任務是否已完成。
            /// </summary>
            public bool IsCompleted => _task._task.IsCompleted;


            /// <inheritdoc />
            public void OnCompleted(Action continuation)
            {
                var This = this;
                _task._task.ContinueWith(t =>
                {
                    if (!This._task.NotExecutable) continuation?.Invoke();
                });
            }


            /// <summary>
            /// 獲取任務結果。
            /// </summary>
            /// <returns></returns>
            public TResult GetResult() => _task._task.Result;
        }


        #endregion
    }

添加任務等待器，同步等待結果返回：

/// <summary>
    /// 獲取任務等待器
    /// </summary>
    /// <returns></returns>
    public new TaskAwaiter GetAwaiter() => new TaskAwaiter(this);


    /// <summary>
    /// 任務等待器
    /// </summary>
    [HostProtection(SecurityAction.LinkDemand, ExternalThreading = true, Synchronization = true)]
    public new struct TaskAwaiter : INotifyCompletion
    {
        private readonly AwaitableTask<TResult> _task;


        /// <summary>
        /// 初始化任務等待器
        /// </summary>
        /// <param name="awaitableTask"></param>
        public TaskAwaiter(AwaitableTask<TResult> awaitableTask) => _task = awaitableTask;


        /// <summary>
        /// 任務是否已完成。
        /// </summary>
        public bool IsCompleted => _task._task.IsCompleted;


        /// <inheritdoc />
        public void OnCompleted(Action continuation)
        {
            var This = this;
            _task._task.ContinueWith(t =>
            {
                if (!This._task.NotExecutable) continuation?.Invoke();
            });
        }


        /// <summary>
        /// 獲取任務結果。
        /// </summary>
        /// <returns></returns>
        public TResult GetResult() => _task._task.Result;
    }

異步任務隊列

/// <summary>
    /// 異步任務隊列
    /// </summary>
    public class AsyncTaskQueue : IDisposable
    {
        /// <summary>
        /// 異步任務隊列
        /// </summary>
        public AsyncTaskQueue()
        {
            _autoResetEvent = new AutoResetEvent(false);
            _thread = new Thread(InternalRunning) { IsBackground = true };
            _thread.Start();
        }


        #region 執行


        /// <summary>
        /// 執行異步操作
        /// </summary>
        /// <typeparam name="T">返回結果類型</typeparam>
        /// <param name="func">異步操作</param>
        /// <returns>isInvalid:異步操作是否有效；result:異步操作結果</returns>
        public async Task<(bool isInvalid, T reslut)> ExecuteAsync<T>(Func<Task<T>> func)
        {
            var task = GetExecutableTask(func);
            var result = await await task;
            if (!task.IsInvalid)
            {
                result = default(T);
            }
            return (task.IsInvalid, result);
        }


        /// <summary>
        /// 執行異步操作
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="func"></param>
        /// <returns></returns>
        public async Task<bool> ExecuteAsync<T>(Func<Task> func)
        {
            var task = GetExecutableTask(func);
            await await task;
            return task.IsInvalid;
        }


        #endregion


        #region 添加任務


        /// <summary>
        /// 獲取待執行任務
        /// </summary>
        /// <param name="action"></param>
        /// <returns></returns>
        private AwaitableTask GetExecutableTask(Action action)
        {
            var awaitableTask = new AwaitableTask(new Task(action));
            AddPenddingTaskToQueue(awaitableTask);
            return awaitableTask;
        }


        /// <summary>
        /// 獲取待執行任務
        /// </summary>
        /// <typeparam name="TResult"></typeparam>
        /// <param name="function"></param>
        /// <returns></returns>
        private AwaitableTask<TResult> GetExecutableTask<TResult>(Func<TResult> function)
        {
            var awaitableTask = new AwaitableTask<TResult>(new Task<TResult>(function));
            AddPenddingTaskToQueue(awaitableTask);
            return awaitableTask;
        }


        /// <summary>
        /// 添加待執行任務到隊列
        /// </summary>
        /// <param name="task"></param>
        /// <returns></returns>
        private void AddPenddingTaskToQueue(AwaitableTask task)
        {
            //添加隊列，加鎖。
            lock (_queue)
            {
                _queue.Enqueue(task);
                //開始執行任務
                _autoResetEvent.Set();
            }
        }


        #endregion


        #region 內部運行


        private void InternalRunning()
        {
            while (!_isDisposed)
            {
                if (_queue.Count == 0)
                {
                    //等待后續任務
                    _autoResetEvent.WaitOne();
                }
                while (TryGetNextTask(out var task))
                {
                    //如已從隊列中刪除
                    if (task.NotExecutable) continue;


                    if (UseSingleThread)
                    {
                        task.RunSynchronously();
                    }
                    else
                    {
                        task.Start();
                    }
                }
            }
        }
        /// <summary>
        /// 上一次異步操作
        /// </summary>
        private AwaitableTask _lastDoingTask;
        private bool TryGetNextTask(out AwaitableTask task)
        {
            task = null;
            while (_queue.Count > 0)
            {
                //獲取并從隊列中移除任務
                if (_queue.TryDequeue(out task) && (!AutoCancelPreviousTask || _queue.Count == 0))
                {
                    //設置進行中的異步操作無效
                    _lastDoingTask?.MarkTaskValid();
                    _lastDoingTask = task;
                    return true;
                }
                //并發操作，設置任務不可執行
                task.SetNotExecutable();
            }
            return false;
        }


        #endregion


        #region dispose


        /// <inheritdoc />
        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }


        /// <summary>
        /// 析構任務隊列
        /// </summary>
        ~AsyncTaskQueue() => Dispose(false);


        private void Dispose(bool disposing)
        {
            if (_isDisposed) return;
            if (disposing)
            {
                _autoResetEvent.Dispose();
            }
            _thread = null;
            _autoResetEvent = null;
            _isDisposed = true;
        }


        #endregion


        #region 屬性及字段


        /// <summary>
        /// 是否使用單線程完成任務.
        /// </summary>
        public bool UseSingleThread { get; set; } = true;


        /// <summary>
        /// 自動取消以前的任務。
        /// </summary>
        public bool AutoCancelPreviousTask { get; set; } = false;


        private bool _isDisposed;
        private readonly ConcurrentQueue<AwaitableTask> _queue = new ConcurrentQueue<AwaitableTask>();
        private Thread _thread;
        private AutoResetEvent _autoResetEvent;


        #endregion

添加異步任務隊列類，用于任務的管理，如添加、執行、篩選等：

1. 自動取消之前的任務 AutoCancelPreviousTask

內部使用線程，循環獲取當前任務列表，如果當前任務被標記NotExecutable不可執行，則跳過。

NotExecutable是何時標記的？

獲取任務時，標記所有獲取的任務為NotExecutable。直到任務列表中為空，那么只執行最后獲取的一個任務。

2. 標記已經進行的任務無效 MarkTaskValid

當前進行的任務，無法中止，那么標記為無效即可。

/// <summary>
    /// 上一次異步操作
    /// </summary>
    private AwaitableTask _lastDoingTask;
    private bool TryGetNextTask(out AwaitableTask task)
    {
        task = null;
        while (_queue.Count > 0)
        {
            //獲取并從隊列中移除任務
            if (_queue.TryDequeue(out task) && (!AutoCancelPreviousTask || _queue.Count == 0))
            {
                //設置進行中的異步操作無效
                _lastDoingTask?.MarkTaskValid();
                _lastDoingTask = task;
                return true;
            }
            //并發操作，設置任務不可執行
            task.SetNotExecutable();
        }
        return false;
    }

后續執行完后，根據此標記，設置操作結果為空。

/// <summary>
    /// 執行異步操作
    /// </summary>
    /// <typeparam name="T">返回結果類型</typeparam>
    /// <param name="func">異步操作</param>
    /// <returns>isInvalid:異步操作是否有效；result:異步操作結果</returns>
    public async Task<(bool isInvalid, T reslut)> ExecuteAsync<T>(Func<Task<T>> func)
    {
        var task = GetExecutableTask(func);
        var result = await await task;
        if (!task.IsInvalid)
        {
            result = default(T);
        }
        return (task.IsInvalid, result);
    }

實踐測試

啟動9個并發任務，測試實際的任務隊列并發操作管理：

public MainWindow()
    {
        InitializeComponent();
        _asyncTaskQueue = new AsyncTaskQueue
        {
            AutoCancelPreviousTask = true,
            UseSingleThread = true
        };
    }
    private AsyncTaskQueue _asyncTaskQueue;
    private void ButtonBase_OnClick(object sender, RoutedEventArgs e)
    {
        // 快速啟動9個任務
        for (var i = 1; i < 10; i++)
        {
            Test(_asyncTaskQueue, i);
        }
    }
    public static async void Test(AsyncTaskQueue taskQueue, int num)
    {
        var result = await taskQueue.ExecuteAsync(async () =>
        {
            Debug.WriteLine("輸入:" + num);
            // 長時間耗時任務
            await Task.Delay(TimeSpan.FromSeconds(5));
            return num * 100;
        });
        Debug.WriteLine($"{num}輸出的:" + result);
    }

測試結果如下：

一共9次操作，只有最后一次操作結果，才是有效的。其它8次操作，一次是無效的，7次操作被取消不執行。

固定時間間隔只保留最后操作

實際業務過程中，大部分場景并不是瞬間丟過來N個任務，而是比如100ms內有20個操作觸發。

這類高并發操作，不止是上方做一個隊列獲取首尾任務，還需要加個延時（時間間隔），減少并發操作的執行，可以基于上方TryGetNextTask方法操作：

/// <summary>
        /// 獲取下一任務
        /// </summary>
        /// <returns></returns>
        private async Task<AwaitableTask> TryGetNextTask()
        {
            //獲取并從隊列中移除任務
            if (_queue.TryDequeue(out var task))
            {
                //設置進行中的異步操作無效
                foreach (var lastDoingTask in _lastDoingTasks)
                {
                    lastDoingTask.MarkTaskInvalid();
                }
                _lastDoingTasks.Add(task);
                //如果有中間任務，則立即取消
                if (_queue.Count != 0)
                {
                    task.SetNotExecutable();
                }
                else
                {
                    //如果沒有中間任務，則設置延遲時間再確認是否有中間任務
                    await Task.Delay(_delay).ConfigureAwait(false);
                    if (_queue.Count != 0)
                    {
                        task.SetNotExecutable();
                    }
                }
                return task;
            }

設置延時取任務后，就能實現在固定的時間內最多只會觸發一次，并且是最后一次操作。

作者：唐宋元明清2188

出處：http://www.cnblogs.com/kybs0/

本文版權歸作者和博客園共有，歡迎轉載，但未經作者同意必須在文章頁面給出原文連接，否則保留追究法律責任的權利。