码迷,mamicode.com
首页 > Windows程序 > 详细

C# 异步并发操作,只保留最后一次操作

时间:2019-12-05 13:32:41      阅读:112      评论:0      收藏:0      [点我收藏+]

标签:write   技术   als   管理   send   hid   name   图片   自定义   

在我们业务操作时,难免会有多次操作,我们期望什么结果呢?

绝大部分情况,应该是只需要最后一次操作的结果,其它操作应该无效。

自定义等待的任务类

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次操作被取消不执行。

 Demo,见Github

C# 异步并发操作,只保留最后一次操作

标签:write   技术   als   管理   send   hid   name   图片   自定义   

原文地址:https://www.cnblogs.com/kybs0/p/11988554.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!