重构《手搓》TaskFactory带你更安全的起飞

一、彻底的重构和优化

• 本次《手搓》TaskFactory的重构是直接把依赖的《手搓》线程池一起重构了
• 可以参看上一篇异步"伪线程"重构《手搓》线程池,支持任务清退
• 也就是说安全系数非常高,直接在线程里面处理
• 大家可以更安全更放心的起飞

二、先看一下同步任务取消的Case

1. 《手搓》TaskFactory手动取消同步任务

• 执行同步方法Hello
• 等待100毫秒后调用Cancel
• 等待117毫秒,触发异常
• 之后在另一个线程Hello还是执行了
• 如果方法已经执行很可能无法真实的取消(除非增加token参数来控制)
• 但Task已经反馈给调用方本次取消,不会继续等待
• 当前“线程”也回收了,避免由此可能导致的线程池堵塞
• 有人可能说,这有什么稀奇的,系统API就支持

var options = new TaskFactoryOptions { ConcurrencyLevel = 1 };
var factory = new ConcurrentTaskFactory(options);
var tokenSource = new CancellationTokenSource();
var sw = Stopwatch.StartNew();
var task = factory.StartNew(() => Hello("张三", 1000), tokenSource.Token);
Assert.NotNull(task);
await Task.Delay(100);
tokenSource.Cancel();
try
{await task;sw.Stop();
}
catch (Exception ex)
{sw.Stop();_output.WriteLine(($"Thread{Environment.CurrentManagedThreadId} {ex}"));
}
_output.WriteLine($"Thread{Environment.CurrentManagedThreadId} Total Span :{sw.Elapsed.TotalMilliseconds}");
await Task.Delay(1000);void Hello(string name, int time = 10)
{Thread.Sleep(time);_output.WriteLine($"Thread{Environment.CurrentManagedThreadId} Hello {name},{DateTime.Now:HH:mm:ss.fff}");
}// Thread19 System.Threading.Tasks.TaskCanceledException: A task was canceled.
//    at TaskTests.Tasks.ConcurrentTaskFactoryTests.ActionCancel() in D:\projects\HandCore.net\UnitTests\TaskTests\Tasks\ConcurrentTaskFactoryTests.cs:line 90
// Thread19 Total Span :121.4448
// Thread11 Hello 张三,11:41:00.641

2. 系统TaskFactory手动取消任务

• 系统TaskFactory的流程和前面一样
• 也是执行同步方法Hello
• 等待100毫秒后调用Cancel
• 执行成功,共耗时1秒
• 是不是很魔幻,居然没取消
• 系统TaskFactory有Bug?

var factory = Task.Factory;
var tokenSource = new CancellationTokenSource();
var sw = Stopwatch.StartNew();
var task = factory.StartNew(() => Hello("张三", 1000), tokenSource.Token);
Assert.NotNull(task);
await Task.Delay(100);
tokenSource.Cancel();
try
{await task;sw.Stop();
}
catch (Exception ex)
{sw.Stop();_output.WriteLine(($"Thread{Environment.CurrentManagedThreadId} {ex}"));
}
_output.WriteLine($"Thread{Environment.CurrentManagedThreadId} Total Span :{sw.Elapsed.TotalMilliseconds}");// Thread10 Hello 张三,11:43:06.118
// Thread19 Total Span :1006.9077

3. 《手搓》TaskFactory设置超时Case

• 还是同步方法Hello
• 这次通过CancelAfter设置1秒超时
• 等待1秒触发异常
• 之后在另一个线程Hello还是执行了
• 也就是说《手搓》TaskFactory超时清退成功

var options = new TaskFactoryOptions { ConcurrencyLevel = 1 };
var factory = new ConcurrentTaskFactory(options);
var tokenSource = new CancellationTokenSource();
var sw = Stopwatch.StartNew();
tokenSource.CancelAfter(TimeSpan.FromSeconds(1));
var task = factory.StartNew(() => Hello("张三", 2000), tokenSource.Token);
Assert.NotNull(task);
try
{await task;sw.Stop();
}
catch (Exception ex)
{sw.Stop();_output.WriteLine(($"Thread{Environment.CurrentManagedThreadId} {ex}"));
}
_output.WriteLine($"Thread{Environment.CurrentManagedThreadId} Total Span :{sw.Elapsed.TotalMilliseconds}");
await Task.Delay(1000);// Thread16 System.Threading.Tasks.TaskCanceledException: A task was canceled.
//    at TaskTests.Tasks.ConcurrentTaskFactoryTests.ActionTimeout() in D:\projects\HandCore.net\UnitTests\TaskTests\Tasks\ConcurrentTaskFactoryTests.cs:line 114
// Thread16 Total Span :1075.7213
// Thread11 Hello 张三,15:57:04.440

4. 系统TaskFactory设置超时Case

• 执行同步方法Hello
• 通过CancelAfter设置1秒超时
• 任务执行成功,耗时2秒
• 再次魔幻的结果,居然又没取消
• 系统TaskFactory有Bug?

var factory = Task.Factory;
var tokenSource = new CancellationTokenSource();
var sw = Stopwatch.StartNew();
tokenSource.CancelAfter(TimeSpan.FromSeconds(1));
var task = factory.StartNew(() => Hello("张三", 2000), tokenSource.Token);
Assert.NotNull(task);
try
{await task;sw.Stop();
}
catch (Exception ex)
{sw.Stop();_output.WriteLine(($"Thread{Environment.CurrentManagedThreadId} {ex}"));
}
sw.Stop();
_output.WriteLine($"Thread{Environment.CurrentManagedThreadId} Total Span :{sw.Elapsed.TotalMilliseconds}");// Thread10 Hello 张三,10:39:45.556
// Total Span :2017.2007

5. 系统TaskFactory修改再测

• 把Cancel提前到StartNew之前
• 这次耗时106毫秒,任务取消
• 原来系统TaskFactory是要添加任务的时候判断是否取消,而不是执行的时候
• 意外不意外,惊喜不惊喜
• 《手搓》TaskFactory是执行时判断是否取消
• 你满意系统TaskFactory的Cancel逻辑还是《手搓》TaskFactory
• 很长一段时间笔者都以为系统TaskFactory有Bug
• 原来是设计如此,终于破案了

var factory = Task.Factory;
var tokenSource = new CancellationTokenSource();
var sw = Stopwatch.StartNew();
tokenSource.Cancel();
var task = factory.StartNew(() => Hello("张三", 1000), tokenSource.Token);
Assert.NotNull(task);
await Task.Delay(100);
try
{await task;sw.Stop();
}
catch (Exception ex)
{sw.Stop();_output.WriteLine(($"Thread{Environment.CurrentManagedThreadId} {ex}"));
}
_output.WriteLine($"Thread{Environment.CurrentManagedThreadId} Total Span :{sw.Elapsed.TotalMilliseconds}");
// System.Threading.Tasks.TaskCanceledException: A task was canceled.
//    at TaskTests.Tasks.TaskFactoryTests.ActionCancel2() in D:\projects\HandCore.net\UnitTests\TaskTests\Tasks\TaskFactoryTests.cs:line 69
// Total Span :106.1664

三、取消异步任务的Case

1. 《手搓》TaskFactory并发控制的Case

• ConcurrencyLevel设置为1,设置1秒超时
• 这个CancellationToken参数传给实际执行的异步方法同时也传给线程池
• 以便线程池及时处理并及时清理已经取消的任务
• 还起到双保险的作用
• 如果HelloAsync没有对CancellationToken做出正确的处理
• 调用者也会及时触发取消的异常
• 本次执行3个异步任务,每个500毫秒
• 耗时1秒,两个任务执行成功,并触发取消异常
• 第3个并未实际执行,这里要归功于token参数起到了拦截的作用
• 系统TaskFactory没有控制异步和并发的逻辑就不对比

var options = new TaskFactoryOptions { ConcurrencyLevel = 1 };
var factory = new ConcurrentTaskFactory(options);
var tokenSource = new CancellationTokenSource();
var sw = Stopwatch.StartNew();
tokenSource.CancelAfter(TimeSpan.FromSeconds(1));
var token = tokenSource.Token;
var task = factory.StartTask((t) => HelloAsync("张三", 500, t), token);
var task2 = factory.StartTask((t) => HelloAsync("李四", 500, t), token);
var task3 = factory.StartTask((t) => HelloAsync("王二", 500, t), token);
try
{await Task.WhenAll(task, task2, task3);sw.Stop();
}
catch (Exception ex)
{sw.Stop();_output.WriteLine(($"Thread{Environment.CurrentManagedThreadId} {ex}"));
}
_output.WriteLine($"Thread{Environment.CurrentManagedThreadId} Total Span :{sw.Elapsed.TotalMilliseconds}");
await Task.Delay(2000);async Task HelloAsync(string name, int time = 10, CancellationToken token = default)
{await Task.Delay(time, token);_output.WriteLine($"Thread{Environment.CurrentManagedThreadId} HelloAsync {name},{DateTime.Now:HH:mm:ss.fff}");
}// Thread31 HelloAsync 张三,10:30:54.720
// Thread11 HelloAsync 李四,10:30:55.231
// System.Threading.Tasks.TaskCanceledException: A task was canceled.
//    at TaskTests.Tasks.ConcurrentTaskFactoryTests.TaskCancel0() in D:\projects\HandCore.net\UnitTests\TaskTests\Tasks\ConcurrentTaskFactoryTests.cs:line 126
// Total Span :1013.774

四、全局设置超时的Case

1. 同步任务无CancellationToken的Case

• ItemLife配置为1秒
• Hello实际需要2秒
• 调用等待1秒后触发取消异常,与ItemLife配置一致
• 同步任务清退成功

var options = new TaskFactoryOptions { ConcurrencyLevel = 1, ItemLife = TimeSpan.FromSeconds(1) };
var factory = new ConcurrentTaskFactory(options);
var sw = Stopwatch.StartNew();
var task = factory.StartNew(() => Hello("张三", 2000));
Assert.NotNull(task);
try
{await task;sw.Stop();
}
catch (Exception ex)
{sw.Stop();_output.WriteLine(($"Thread{Environment.CurrentManagedThreadId} {ex}"));
}
_output.WriteLine($"Thread{Environment.CurrentManagedThreadId} Total Span :{sw.Elapsed.TotalMilliseconds}");
await Task.Delay(1000);// Thread16 System.Threading.Tasks.TaskCanceledException: A task was canceled.
//    at TaskTests.Tasks.ConcurrentTaskFactoryTests.ActionItemLife() in D:\projects\HandCore.net\UnitTests\TaskTests\Tasks\ConcurrentTaskFactoryTests.cs:line 35
// Thread16 Total Span :1016.3529
// Thread11 Hello 张三,16:09:49.847

2. 异步任务无CancellationToken的Case

• ItemLife配置为1秒
• HelloAsync实际需要2秒
• 调用等待1秒后触发取消异常,与ItemLife配置一致
• 异步任务也清退成功
• 由于本次没有给HelloAsync传token,虽然清退了,HelloAsync还是稍后在其他线程执行了
• 这个Case再次证明token参数的重要性,可以避免不必要的操作

var options = new TaskFactoryOptions { ConcurrencyLevel = 1, ItemLife = TimeSpan.FromSeconds(1) };
var factory = new ConcurrentTaskFactory(options);
var sw = Stopwatch.StartNew();
var task = factory.StartTask(() => HelloAsync("张三", 2000));
Assert.NotNull(task);
try
{await task;sw.Stop();
}
catch (Exception ex)
{sw.Stop();_output.WriteLine(($"Thread{Environment.CurrentManagedThreadId} {ex}"));
}
_output.WriteLine($"Thread{Environment.CurrentManagedThreadId} Total Span :{sw.Elapsed.TotalMilliseconds}");
await Task.Delay(2000);// Thread17 System.Threading.Tasks.TaskCanceledException: A task was canceled.
//    at TaskTests.Tasks.ConcurrentTaskFactoryTests.TaskItemLife() in D:\projects\HandCore.net\UnitTests\TaskTests\Tasks\ConcurrentTaskFactoryTests.cs:line 65
// Thread17 Total Span :1020.1329
// Thread8 HelloAsync 张三,16:14:04.831

五、揭秘重构《手搓》TaskFactory

1. 前1版本《手搓》TaskFactory是基于并发配额控制

• 启动异步扣除并发配额
• 注册ContinueWith返还并发配额
• 参考以前的博文《手搓》TaskFactory带你安全的起飞

2. 重构《手搓》TaskFactory去掉了并发配额控制

• 去掉并发配额控制
• 异步线程直接在线程池里面await
• 线程池原生支持性能更好
• 线程池大小就是最好的并发控制,原生支持后并发配额完全不需要

六、总结

1. 《手搓》TaskFactory的作用

• 《手搓》TaskFactory的主要作用是并发控制
• .net实现异步多并发很容易,但是并发太大很容易失控
• 比如数据库连接过高、上游服务被打挂
• 为此很多人都把.net程序写成“单线程”模式
• 即使写成“单线程”模式依然避免不了上游资源被打挂的事故
• 通过《手搓》TaskFactory可以设置为全局模式,来管控某些某些资源的并发控制
• 同时程序改动极小
• 无论是同步操作还是异步都可以托管给《手搓》TaskFactory
• 也就是实现程序的异步多并发的起飞
• 加上CancellationToken和ItemLife可以保证线程池的不会被堵塞
• 安全高效的线程池助力.net程序安全的起飞

2. 《手搓》线程池的作用

• 其一支持《手搓》TaskFactory的执行处理
• 其二支持事件总线的事件派发
• 拿来做一个毫秒级的任务调度系统应该也没什么问题

3. CancellationToken

• 及时清退任务的关键之一是CancellationToken
• 特别是含await的异步方法
• 当await成功后很可能已经发生了线程切换,耗时更是不确定的因素
• 这时很可能要校验CancellationToken判断是否还有必要继续执行
• 多步骤逻辑耗时,每一步骤前也很可能需要判断是否还有必要继续执行后面的步骤
• 耗时循环,每次循环也很可能需要判断是否还有必要继续执行
• 在某些特殊情况,同步方法也可能有必要增加CancellationToken
• 给实际执行的方法传CancellationToken的同时传给《手搓》TaskFactory或者《手搓》线程池,可以实现双保险

4. ItemLife

• ItemLife是线程池兜底的安全机制
• 可以有效避免线程池被堵塞
• 特别是那些还没来得及加CancellationToken的祖传代码

好了,就介绍到这里,更多信息请查看源码库
源码托管地址: https://github.com/donetsoftwork/HandCore.net ，欢迎大家直接查看源码。
gitee同步更新:https://gitee.com/donetsoftwork/HandCore.net

如果大家喜欢请动动您发财的小手手帮忙点一下Star,谢谢！！！