C# 异步并发操作，只保留最后一次操作
在我们业务操作时，难免会有多次操作，我们期望什么结果呢？
绝大部分情况，应该是只需要最后一次操作的结果，其它操作应该无效。
自定义等待的任务类
1. 可等待的任务类 AwaitableTask:
1 /// <summary> 2 /// 可等待的任务 3 /// </summary> 4 public class AwaitableTask 5 { 6 /// <summary> 7 /// 获取任务是否为不可执行状态 8 /// </summary> 9 public bool NotExecutable { get; private set; } 10 11 /// <summary> 12 /// 设置任务不可执行 13 /// </summary> 14 public void SetNotExecutable() 15 { 16 NotExecutable = true; 17 } 18 19 /// <summary> 20 /// 获取任务是否有效 21 /// 注：对无效任务，可以不做处理。减少并发操作导致的干扰 22 /// </summary> 23 public bool IsInvalid { get; private set; } = true; 24 25 /// <summary> 26 /// 标记任务无效 27 /// </summary> 28 public void MarkTaskValid() 29 { 30 IsInvalid = false; 31 } 32 33 #region Task 34 35 private readonly Task _task; 36 /// <summary> 37 /// 初始化可等待的任务。 38 /// </summary> 39 /// <param name="task"></param> 40 public AwaitableTask(Task task) => _task = task; 41 42 /// <summary> 43 /// 获取任务是否已完成 44 /// </summary> 45 public bool IsCompleted => _task.IsCompleted; 46 47 /// <summary> 48 /// 任务的Id 49 /// </summary> 50 public int TaskId => _task.Id; 51 52 /// <summary> 53 /// 开始任务 54 /// </summary> 55 public void Start() => _task.Start(); 56 57 /// <summary> 58 /// 同步执行开始任务 59 /// </summary> 60 public void RunSynchronously() => _task.RunSynchronously(); 61 62 #endregion 63 64 #region TaskAwaiter 65 66 /// <summary> 67 /// 获取任务等待器 68 /// </summary> 69 /// <returns></returns> 70 public TaskAwaiter GetAwaiter() => new TaskAwaiter(this); 71 72 /// <summary>Provides an object that waits for the completion of an asynchronous task. </summary> 73 [HostProtection(SecurityAction.LinkDemand, ExternalThreading = true, Synchronization = true)] 74 public struct TaskAwaiter : INotifyCompletion 75 { 76 private readonly AwaitableTask _task; 77 78 /// <summary> 79 /// 任务等待器 80 /// </summary> 81 /// <param name="awaitableTask"></param> 82 public TaskAwaiter(AwaitableTask awaitableTask) => _task = awaitableTask; 83 84 /// <summary> 85 /// 任务是否完成. 86 /// </summary> 87 public bool IsCompleted => _task._task.IsCompleted; 88 89 /// <inheritdoc /> 90 public void OnCompleted(Action continuation) 91 { 92 var This = this; 93 _task._task.ContinueWith(t => 94 { 95 if (!This._task.NotExecutable) continuation?.Invoke(); 96 }); 97 } 98 /// <summary> 99 /// 获取任务结果 100 /// </summary> 101 public void GetResult() => _task._task.Wait(); 102 } 103 104 #endregion 105 106 }
View Code
无效的操作可以分为以下俩种：
- 已经进行中的操作，后续结果应标记为无效
- 还没开始的操作，后续不执行
自定义任务类型 AwaitableTask中，添加俩个字段NotExecutable、IsInvalid：
1 /// <summary> 2 /// 获取任务是否为不可执行状态 3 /// </summary> 4 public bool NotExecutable { get; private set; } 5 /// <summary> 6 /// 获取任务是否有效 7 /// 注：对无效任务，可以不做处理。减少并发操作导致的干扰 8 /// </summary> 9 public bool IsInvalid { get; private set; } = true;
2. 有返回结果的可等待任务类 AwaitableTask<TResult>：
1 /// <summary> 2 /// 可等待的任务 3 /// </summary> 4 /// <typeparam name="TResult"></typeparam> 5 public class AwaitableTask<TResult> : AwaitableTask 6 { 7 private readonly Task<TResult> _task; 8 /// <summary> 9 /// 初始化可等待的任务 10 /// </summary> 11 /// <param name="task">需要执行的任务</param> 12 public AwaitableTask(Task<TResult> task) : base(task) => _task = task; 13 14 #region TaskAwaiter 15 16 /// <summary> 17 /// 获取任务等待器 18 /// </summary> 19 /// <returns></returns> 20 public new TaskAwaiter GetAwaiter() => new TaskAwaiter(this); 21 22 /// <summary> 23 /// 任务等待器 24 /// </summary> 25 [HostProtection(SecurityAction.LinkDemand, ExternalThreading = true, Synchronization = true)] 26 public new struct TaskAwaiter : INotifyCompletion 27 { 28 private readonly AwaitableTask<TResult> _task; 29 30 /// <summary> 31 /// 初始化任务等待器 32 /// </summary> 33 /// <param name="awaitableTask"></param> 34 public TaskAwaiter(AwaitableTask<TResult> awaitableTask) => _task = awaitableTask; 35 36 /// <summary> 37 /// 任务是否已完成。 38 /// </summary> 39 public bool IsCompleted => _task._task.IsCompleted; 40 41 /// <inheritdoc /> 42 public void OnCompleted(Action continuation) 43 { 44 var This = this; 45 _task._task.ContinueWith(t => 46 { 47 if (!This._task.NotExecutable) continuation?.Invoke(); 48 }); 49 } 50 51 /// <summary> 52 /// 获取任务结果。 53 /// </summary> 54 /// <returns></returns> 55 public TResult GetResult() => _task._task.Result; 56 } 57 58 #endregion 59 }
View Code
添加任务等待器，同步等待结果返回：
1 /// <summary> 2 /// 获取任务等待器 3 /// </summary> 4 /// <returns></returns> 5 public new TaskAwaiter GetAwaiter() => new TaskAwaiter(this); 6 7 /// <summary> 8 /// 任务等待器 9 /// </summary> 10 [HostProtection(SecurityAction.LinkDemand, ExternalThreading = true, Synchronization = true)] 11 public new struct TaskAwaiter : INotifyCompletion 12 { 13 private readonly AwaitableTask<TResult> _task; 14 15 /// <summary> 16 /// 初始化任务等待器 17 /// </summary> 18 /// <param name="awaitableTask"></param> 19 public TaskAwaiter(AwaitableTask<TResult> awaitableTask) => _task = awaitableTask; 20 21 /// <summary> 22 /// 任务是否已完成。 23 /// </summary> 24 public bool IsCompleted => _task._task.IsCompleted; 25 26 /// <inheritdoc /> 27 public void OnCompleted(Action continuation) 28 { 29 var This = this; 30 _task._task.ContinueWith(t => 31 { 32 if (!This._task.NotExecutable) continuation?.Invoke(); 33 }); 34 } 35 36 /// <summary> 37 /// 获取任务结果。 38 /// </summary> 39 /// <returns></returns> 40 public TResult GetResult() => _task._task.Result; 41 }
异步任务队列
添加异步任务队列类，用于任务的管理，如添加、执行、筛选等：
1 /// <summary> 2 /// 异步任务队列 3 /// </summary> 4 public class AsyncTaskQueue : IDisposable 5 { 6 /// <summary> 7 /// 异步任务队列 8 /// </summary> 9 public AsyncTaskQueue() 10 { 11 _autoResetEvent = new AutoResetEvent(false); 12 _thread = new Thread(InternalRunning) { IsBackground = true }; 13 _thread.Start(); 14 } 15 16 #region 执行 17 18 /// <summary> 19 /// 执行异步操作 20 /// </summary> 21 /// <typeparam name="T">返回结果类型</typeparam> 22 /// <param name="func">异步操作</param> 23 /// <returns>isInvalid:异步操作是否有效；result:异步操作结果</returns> 24 public async Task<(bool isInvalid, T reslut)> ExecuteAsync<T>(Func<Task<T>> func) 25 { 26 var task = GetExecutableTask(func); 27 var result = await await task; 28 if (!task.IsInvalid) 29 { 30 result = default(T); 31 } 32 return (task.IsInvalid, result); 33 } 34 35 /// <summary> 36 /// 执行异步操作 37 /// </summary> 38 /// <typeparam name="T"></typeparam> 39 /// <param name="func"></param> 40 /// <returns></returns> 41 public async Task<bool> ExecuteAsync<T>(Func<Task> func) 42 { 43 var task = GetExecutableTask(func); 44 await await task; 45 return task.IsInvalid; 46 } 47 48 #endregion 49 50 #region 添加任务 51 52 /// <summary> 53 /// 获取待执行任务 54 /// </summary> 55 /// <param name="action"></param> 56 /// <returns></returns> 57 private AwaitableTask GetExecutableTask(Action action) 58 { 59 var awaitableTask = new AwaitableTask(new Task(action)); 60 AddPenddingTaskToQueue(awaitableTask); 61 return awaitableTask; 62 } 63 64 /// <summary> 65 /// 获取待执行任务 66 /// </summary> 67 /// <typeparam name="TResult"></typeparam> 68 /// <param name="function"></param> 69 /// <returns></returns> 70 private AwaitableTask<TResult> GetExecutableTask<TResult>(Func<TResult> function) 71 { 72 var awaitableTask = new AwaitableTask<TResult>(new Task<TResult>(function)); 73 AddPenddingTaskToQueue(awaitableTask); 74 return awaitableTask; 75 } 76 77 /// <summary> 78 /// 添加待执行任务到队列 79 /// </summary> 80 /// <param name="task"></param> 81 /// <returns></returns> 82 private void AddPenddingTaskToQueue(AwaitableTask task) 83 { 84 //添加队列，加锁。 85 lock (_queue) 86 { 87 _queue.Enqueue(task); 88 //开始执行任务 89 _autoResetEvent.Set(); 90 } 91 } 92 93 #endregion 94 95 #region 内部运行 96 97 private void InternalRunning() 98 { 99 while (!_isDisposed) 100 { 101 if (_queue.Count == 0) 102 { 103 //等待后续任务 104 _autoResetEvent.WaitOne(); 105 } 106 while (TryGetNextTask(out var task)) 107 { 108 //如已从队列中删除 109 if (task.NotExecutable) continue; 110 111 if (UseSingleThread) 112 { 113 task.RunSynchronously(); 114 } 115 else 116 { 117 task.Start(); 118 } 119 } 120 } 121 } 122 /// <summary> 123 /// 上一次异步操作 124 /// </summary> 125 private AwaitableTask _lastDoingTask; 126 private bool TryGetNextTask(out AwaitableTask task) 127 { 128 task = null; 129 while (_queue.Count > 0) 130 { 131 //获取并从队列中移除任务 132 if (_queue.TryDequeue(out task) && (!AutoCancelPreviousTask || _queue.Count == 0)) 133 { 134 //设置进行中的异步操作无效 135 _lastDoingTask?.MarkTaskValid(); 136 _lastDoingTask = task; 137 return true; 138 } 139 //并发操作，设置任务不可执行 140 task.SetNotExecutable(); 141 } 142 return false; 143 } 144 145 #endregion 146 147 #region dispose 148 149 /// <inheritdoc /> 150 public void Dispose() 151 { 152 Dispose(true); 153 GC.SuppressFinalize(this); 154 } 155 156 /// <summary> 157 /// 析构任务队列 158 /// </summary> 159 ~AsyncTaskQueue() => Dispose(false); 160 161 private void Dispose(bool disposing) 162 { 163 if (_isDisposed) return; 164 if (disposing) 165 { 166 _autoResetEvent.Dispose(); 167 } 168 _thread = null; 169 _autoResetEvent = null; 170 _isDisposed = true; 171 } 172 173 #endregion 174 175 #region 属性及字段 176 177 /// <summary> 178 /// 是否使用单线程完成任务. 179 /// </summary> 180 public bool UseSingleThread { get; set; } = true; 181 182 /// <summary> 183 /// 自动取消以前的任务。 184 /// </summary> 185 public bool AutoCancelPreviousTask { get; set; } = false; 186 187 private bool _isDisposed; 188 private readonly ConcurrentQueue<AwaitableTask> _queue = new ConcurrentQueue<AwaitableTask>(); 189 private Thread _thread; 190 private AutoResetEvent _autoResetEvent; 191 192 #endregion 193 194 }
View Code
1. 自动取消之前的任务 AutoCancelPreviousTask
内部使用线程，循环获取当前任务列表，如果当前任务被标记NotExecutable不可执行，则跳过。
NotExecutable是何时标记的？
获取任务时，标记所有获取的任务为NotExecutable。直到任务列表中为空，那么只执行最后获取的一个任务。
2. 标记已经进行的任务无效 MarkTaskValid
当前进行的任务，无法中止，那么标记为无效即可。
1 /// <summary> 2 /// 上一次异步操作 3 /// </summary> 4 private AwaitableTask _lastDoingTask; 5 private bool TryGetNextTask(out AwaitableTask task) 6 { 7 task = null; 8 while (_queue.Count > 0) 9 { 10 //获取并从队列中移除任务 11 if (_queue.TryDequeue(out task) && (!AutoCancelPreviousTask || _queue.Count == 0)) 12 { 13 //设置进行中的异步操作无效 14 _lastDoingTask?.MarkTaskValid(); 15 _lastDoingTask = task; 16 return true; 17 } 18 //并发操作，设置任务不可执行 19 task.SetNotExecutable(); 20 } 21 return false; 22 }
后续执行完后，根据此标记，设置操作结果为空。
1 /// <summary> 2 /// 执行异步操作 3 /// </summary> 4 /// <typeparam name="T">返回结果类型</typeparam> 5 /// <param name="func">异步操作</param> 6 /// <returns>isInvalid:异步操作是否有效；result:异步操作结果</returns> 7 public async Task<(bool isInvalid, T reslut)> ExecuteAsync<T>(Func<Task<T>> func) 8 { 9 var task = GetExecutableTask(func); 10 var result = await await task; 11 if (!task.IsInvalid) 12 { 13 result = default(T); 14 } 15 return (task.IsInvalid, result); 16 }
实践测试
启动10个并发任务，测试实际的任务队列并发操作管理：
1 public MainWindow() 2 { 3 InitializeComponent(); 4 _asyncTaskQueue = new AsyncTaskQueue 5 { 6 AutoCancelPreviousTask = true, 7 UseSingleThread = true 8 }; 9 } 10 private AsyncTaskQueue _asyncTaskQueue; 11 private void ButtonBase_OnClick(object sender, RoutedEventArgs e) 12 { 13 // 快速启动10个任务 14 for (var i = 1; i < 10; i++) 15 { 16 Test(_asyncTaskQueue, i); 17 } 18 } 19 public static async void Test(AsyncTaskQueue taskQueue, int num) 20 { 21 var result = await taskQueue.ExecuteAsync(async () => 22 { 23 Debug.WriteLine("输入:" + num); 24 // 长时间耗时任务 25 await Task.Delay(TimeSpan.FromSeconds(5)); 26 return num * 100; 27 }); 28 Debug.WriteLine($"{num}输出的:" + result); 29 }
测试结果如下：
只有最后一次操作结果，才是有效的。其它9次操作，一次是无效的，8次操作被取消不执行。
版权声明：本文为kybs0原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。