使用非同步操作時的注意要點(翻譯)
非同步操作時應注意的要點
使用場景
非同步操作時需要注意的要點
1.使用非同步方法返回值應當避免使用void
在使用非同步方法中最好不要使用void當做返回值,無返回值也應使用Task作為返回值,因為使用void作為返回值具有以下缺點
- 無法得知非同步函式的狀態機在什麼時候執行完畢
- 如果非同步函式中出現異常,則會導致程序崩潰
❌非同步函式不應該返回void
static void Main(string[] args) { try { // 如果Run方法無異常正常執行,那麼程式無法得知其狀態機什麼時候執行完畢 Run(); } catch (Exception ex) { Console.WriteLine(ex.Message); } Console.Read(); } static async void Run() { // 由於方法返回的為void,所以在呼叫此方法時無法捕捉異常,使得程序崩潰 throw new Exception("異常了"); await Task.Run(() => { }); }
☑️應該將非同步函式返回Task
static async Task Main(string[] args) { try { // 因為在此進行await,所以主程式知道什麼時候狀態機執行完成 await RunAsync(); Console.Read(); } catch (Exception ex) { Console.WriteLine(ex.Message); } } static async Task RunAsync() { // 因為此非同步方法返回的為Task,所以此異常可以被捕捉 throw new Exception("異常了"); await Task.Run(() => { }); }
注:事件是一個例外,非同步事件也是返回void
2.對於預計算或者簡單計算的函式建議使用Task.FromResult代替Task.Run
對於一些預先知道的結果或者只是一個簡單的計算函式,使用Task,FromResult要比Task.Run效能要好,因為Task.FromResult只是建立了一個包裝已計算任務的任務,而Task.Run會將一個工作項線上程池進行排隊,計算,返回.並且使用Task.FromResult在具有SynchronizationContext 程式中(例如WinForm)呼叫Result或wait()並不會死鎖(雖然並不建議這麼幹)
❌對於預計算或普通計算的函式不應該這麼寫
public async Task<int> RunAsync()
{
return await Task.Run(()=>1+1);
}☑️而應該使用Task.FromResult代替
public async Task<int> RunAsync()
{
return await Task.FromResult(1 + 1);
}還有另外一種代替方法,那就是使用ValueTask型別,ValueTask是一個可被等待非同步結構,所以並不會在堆中分配記憶體和任務分配,從而效能更優化.
☑️使用ValueTask
static async Task Main(string[] args)
{
await AddAsync(1, 1);
}
static ValueTask<int> AddAsync(int a, int b)
{
// 返回一個可被等待的ValueTask型別
return new ValueTask<int>(a + b);
}注: ValueTask結構是C#7.0加入的,存在於Sysntem,Threading.Task.Extensions包中
3.避免使用Task.Run()方法執行長時間堵塞執行緒的工作
長時間執行的工作是指在應用程式生命週期執行後臺工作的執行緒,如:執行processing queue items,執行sleeping,執行waiting或者處理某些資料,此類執行緒不建議使用Task.Run方法執行,因為Task.Run方法是將任務線上程池內進行排隊執行,如果執行緒池執行緒進行長時間堵塞,會導致執行緒池增長,進而浪費效能,所以如果想要執行長時間的工作建議直接建立一個新執行緒進行工作
❌下面這個例子就利用了執行緒池執行長時間的阻塞工作
public class QueueProcessor
{
private readonly BlockingCollection<Message> _messageQueue = new BlockingCollection<Message>();
public void StartProcessing()
{
Task.Run(ProcessQueue);
}
public void Enqueue(Message message)
{
_messageQueue.Add(message);
}
private void ProcessQueue()
{
foreach (var item in _messageQueue.GetConsumingEnumerable())
{
ProcessItem(item);
}
}
private void ProcessItem(Message message) { }
}☑️所以應該改成這樣
public class QueueProcessor
{
private readonly BlockingCollection<Message> _messageQueue = new BlockingCollection<Message>();
public void StartProcessing()
{
var thread = new Thread(ProcessQueue)
{
// 設定執行緒為背後執行緒,使得在主執行緒結束時此執行緒也會自動結束
IsBackground = true
};
thread.Start();
}
public void Enqueue(Message message)
{
_messageQueue.Add(message);
}
private void ProcessQueue()
{
foreach (var item in _messageQueue.GetConsumingEnumerable())
{
ProcessItem(item);
}
}
private void ProcessItem(Message message) { }
}?執行緒池內執行緒增加會導致在執行時大量的進行上下文切換,從而浪費程式的整體效能, 執行緒池詳細資訊請參考CLR第27章
?Task.Factory.StartNew方法中有一個TaskCreationOptions引數過載,如果設定為LongRunning,則會建立一個新執行緒執行
// 此方法會建立一個新執行緒進行執行
Task.Factory.StartNew(() => { }, TaskCreationOptions.LongRunning);4.避免使用Task.Result和Task.Wait()來堵塞執行緒
使用Task.Result和Task.Wait()兩個方法進行阻塞非同步同步化比直接同步方法阻塞還要MUCH worse(更糟),這種方式被稱為Sync over async 此方式操作步驟如下
1.非同步執行緒啟動
2.呼叫執行緒呼叫Result或者Wait()進行阻塞
3.非同步完成時,將一個延續程式碼排程到執行緒池,恢復等待該操作的程式碼
雖然看起來並沒有什麼關係,但是其實這裡卻是使用了兩個執行緒來完成同步操作,這樣通常會導致執行緒飢餓和死鎖
?執行緒飢餓(starvation):指等待時間已經影響到程序執行,如果等待時間過長,導致程序使命沒有意義時,稱之為餓死
?死鎖(deadlock):指兩個或兩個以上的執行緒相互爭奪資源,導致程序永久堵塞,
?使用Task.Result和Task.Wait()會在winform和ASP.NET中會死鎖,因為它們SynchronizationContext具有物件,兩個執行緒在SynchronizationContext爭奪導致死鎖,而ASP.NET Core則不會產生死鎖,因為ASP.NET Core本質是一個控制檯應用程式,並沒有上下文
❌下面的例子,雖然都不會產生死鎖,但是依然具有很多問題
async Task<string> RunAsync()
{
// 此執行緒ID輸出與UI執行緒ID不一致
Debug.WriteLine("UI執行緒:"+Thread.CurrentThread.ManagedThreadId);
return await Task.Run(() => "Run");
}
string DoOperationBlocking()
{
// 這種方法雖然擺脫了死鎖的問題,但是也導致了上下文問題,RunAsync不在以UI執行緒呼叫
// Result和Wait()方法如果出現異常,異常將被包裝為AggregateException進行丟擲,
return Task.Run(() => RunAsync()).Result;
}
}
private async void button1_Click(object sender, EventArgs e)
{
Debug.WriteLine("RunAsync:" + Thread.CurrentThread.ManagedThreadId);
Debug.WriteLine(DoOperationBlocking());
}public string DoOperationBlocking2()
{
// 此方法也是會導致上下文問題,
// GetAwaiter()方法對異常不會包裝
return Task.Run(() => RunAsync()).GetAwaiter().GetResult();
}5.建議使用await來代替continueWith任務
在async和await,當時可以使用continueWith來延遲執行一些方法,但是continueWith並不會捕捉`SynchronizationContext `,所以建議使用await代替continueWith
❌下面例子就是使用continueWith
private void button1_Click(object sender, EventArgs e)
{
Debug.WriteLine("UI執行緒:" + Thread.CurrentThread.ManagedThreadId);
RunAsync().ContinueWith(task =>
{
Console.WriteLine("RunAsync returned:"+task.Result);
// 因為是使用的continueWith,所以執行緒ID與UI執行緒並不一致
Debug.WriteLine("ContinueWith:" + Thread.CurrentThread.ManagedThreadId);
});
}
public async Task<int> RunAsync()
{
return await Task.FromResult(1 + 1);
}☑️應該使用await來代替continueWith
private async void button1_Click(object sender, EventArgs e)
{
Debug.WriteLine("UI執行緒:" + Thread.CurrentThread.ManagedThreadId);
Debug.WriteLine("RunAsync returned:"+ await RunAsync());
Debug.WriteLine("UI執行緒:" + Thread.CurrentThread.ManagedThreadId);
}
public async Task<int> RunAsync()
{
return await Task.FromResult(1 + 1);
}6.建立TaskCompletionSource
對於編寫類庫的人來說TaskCompletionSource<T>是一個具有非常重要的作用,預設情況下任務延續可能會在呼叫try/set(Result/Exception/Cancel)的執行緒上進行執行,這也就是說作為編寫類庫的人來說必須需要考慮上下文,這通常是非常危險,可能就會導致死鎖' 執行緒池飢餓 *資料結構損壞(如果程式碼異常執行)
所以在建立TaskCompletionSourece<T>時,應該使用TaskCreationOption.RunContinuationAsyncchronously引數將後續任務交給執行緒池進行處理
❌下面例子就沒有使用TaskCreationOptions.RunComtinuationsAsynchronously,
static void Main(string[] args)
{
ThreadPool.SetMinThreads(100, 100);
Console.WriteLine("Main CurrentManagedThreadId:" + Environment.CurrentManagedThreadId);
var tcs = new TaskCompletionSource<bool>();
// 使用TaskContinuationOptions.ExecuteSynchronously來測試延續任務
ContinueWith(1, tcs.Task);
// 測試await延續任務
ContinueAsync(2, tcs.Task);
Task.Run(() =>
{
Console.WriteLine("Task Run CurrentManagedThreadId:" + Environment.CurrentManagedThreadId );
tcs.TrySetResult(true);
});
Console.ReadLine();
}
static void print(int id) => Console.WriteLine($"continuation:{id}\tCurrentManagedThread:{Environment.CurrentManagedThreadId}");
static async Task ContinueAsync(int id, Task task)
{
await task.ConfigureAwait(false);
print(id);
}
static Task ContinueWith(int id, Task task)
{
return task.ContinueWith(
t => print(id),
CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
}☑️所以應該改為使用TaskCreationOptions.RunComtinuationsAsynchronously引數進行設定TaskCompletionSoure
static void Main(string[] args)
{
ThreadPool.SetMinThreads(100, 100);
Console.WriteLine("Main CurrentManagedThreadId:" + Environment.CurrentManagedThreadId);
var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
// 使用TaskContinuationOptions.ExecuteSynchronously來測試延續任務
ContinueWith(1, tcs.Task);
// 測試await延續任務
ContinueAsync(2, tcs.Task);
Task.Run(() =>
{
Console.WriteLine("Task Run CurrentManagedThreadId:" + Environment.CurrentManagedThreadId);
tcs.TrySetResult(true);
});
Console.ReadLine();
}
static void print(int id) => Console.WriteLine($"continuation:{id}\tCurrentManagedThread:{Environment.CurrentManagedThreadId}");
static async Task ContinueAsync(int id, Task task)
{
await task.ConfigureAwait(false);
print(id);
}
static Task ContinueWith(int id, Task task)
{
return task.ContinueWith(
t => print(id),
CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
}?TaskCreationOptions.RunContinuationsAsynchronously屬性和TaskContinuationOptions.RunContinuationsAsynchronously很相似,但請注意它們的使用方式
7.建議使用CancellationTokenSource(s)進行超時管理時總是釋放(dispose)
用於進行超時的CancellationTokenSources,如果不釋放,則會增加timer queue(計時器佇列)的壓力
❌下面例子因為沒有釋放,所以在每次請求發出之後,計時器在佇列中停留10秒鐘
public async Task<Stream> HttpClientAsyncWithCancellationBad()
{
var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
using (var client = _httpClientFactory.CreateClient())
{
var response = await client.GetAsync("http://backend/api/1", cts.Token);
return await response.Content.ReadAsStreamAsync();
}
}☑️所以應該及時的釋放CancellationSoure,使得正確的從佇列中刪除計時器
public async Task<Stream> HttpClientAsyncWithCancellationGood()
{
using (var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)))
{
using (var client = _httpClientFactory.CreateClient())
{
var response = await client.GetAsync("http://backend/api/1", cts.Token);
return await response.Content.ReadAsStreamAsync();
}
}
}?設定延遲時間具有兩種方式
1.構造器引數
public CancellationTokenSource(TimeSpan delay); public CancellationTokenSource(int millisecondsDelay);
2.呼叫例項物件CancelAfter() public void CancelAfter(TimeSpan delay); public void CancelAfter(int millisecondsDelay);
8.建議將協作式取消物件(CancellationToken)傳遞給所有使用到的API
由於在.NET中取消操作必須顯示的傳遞CancellationToken,所以如果想取消所有呼叫的非同步函式,那麼應該將CancllationToken傳遞給此呼叫鏈中的所有函式
❌下面例子在呼叫ReadAsync時並沒有傳遞CancellationToken,所以不能有效的取消
public async Task<string> DoAsyncThing(CancellationToken cancellationToken = default)
{
byte[] buffer = new byte[1024];
// 使用FileOptions.Asynchronous引數指定非同步通訊
using(Stream stream = new FileStream(
@"d:\資料\Blogs\Task\TaskTest",
FileMode.OpenOrCreate,
FileAccess.ReadWrite,
FileShare.None,
1024,
options:FileOptions.Asynchronous))
{
// 由於並沒有將cancellationToken傳遞給ReadAsync,所以無法進行有效的取消
int read = await stream.ReadAsync(buffer, 0, buffer.Length);
return Encoding.UTF8.GetString(buffer, 0, read);
}
}☑️所以應該將CancellationToken傳遞給ReadAsync(),以達到有效的取消
public async Task<string> DoAsyncThing(CancellationToken cancellationToken = default)
{
byte[] buffer = new byte[1024];
// 使用FileOptions.Asynchronous引數指定非同步通訊
using(Stream stream = new FileStream(
@"d:\資料\Blogs\Task\TaskTest",
FileMode.OpenOrCreate,
FileAccess.ReadWrite,
FileShare.None,
1024,
options:FileOptions.Asynchronous))
{
// 由於並沒有將cancellationToken傳遞給ReadAsync,所以無法進行有效的取消
int read = await stream.ReadAsync(buffer, 0, buffer.Length,cancellationToken);
return Encoding.UTF8.GetString(buffer, 0, read);
}
}?在使用非同步IO時,應該將options引數設定為FileOptions.Asynchronous,否則會產生額外的執行緒浪費,詳細資訊請參考CLR中28.12節
9.建議取消那些不會自動取消的操作(CancellationTokenRegistry,timer)
在非同步程式設計時出現了一種模式cancelling an uncancellable operation,這個用於取消像CancellationTokenRegistry和timer這樣的東西,通常是在被取消或超時時建立另外一個執行緒進行操作,然後使用Task.WhenAny進行判斷是完成還是被取消了
使用CancellationToken
:x: 下面例子使用了Task.delay(-1,token)建立在觸發CancellationToken時觸發的任務,但是如果CancellationToken不觸發,則沒有辦法釋放CancellationTokenRegistry,就有可能會導致記憶體洩露 public static async Task<T> WithCancellation<T>(this Task<T> task, CancellationToken cancellationToken)
{
// 沒有方法釋放cancellationToken註冊
var delayTask = Task.Delay(-1, cancellationToken);
var resultTask = await Task.WhenAny(task, delayTask);
if (resultTask == delayTask)
{
// 取消非同步操作
throw new OperationCanceledException();
}
return await task;
}:ballot_box_with_check:所以應該改成下面這樣,在任務一完成,就釋放CancellationTokenRegistrypublic static async Task<T> WithCancellation<T>(this Task<T> task, CancellationToken cancellationToken)
{
var tcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
using (cancellationToken.Register(state =>
{
// 這樣將在其中一個任務觸發時立即釋放CancellationTokenRegistry
((TaskCompletionSource<object>)state).TrySetResult(null);
},
tcs))
{
var resultTask = await Task.WhenAny(task, tcs.Task);
if (resultTask == tcs.Task)
{
// 取消非同步操作
throw new OperationCanceledException(cancellationToken);
}
return await task;
}
}
使用超時任務
:x:下面這個例子即使在操作完成之後,也不會取消定時器,這也就是說最終會在計時器佇列中產生大量的計時器,從而浪費效能public static async Task<T> TimeoutAfter<T>(this Task<T> task, TimeSpan timeout)
{
var delayTask = Task.Delay(timeout);
var resultTask = await Task.WhenAny(task, delayTask);
if (resultTask == delayTask)
{
// 取消非同步操作
throw new OperationCanceledException();
}
return await task;
}:ballot_box_with_check:應改成下面這樣,這樣將在任務完成之後,取消計時器的操作public static async Task<T> TimeoutAfter<T>(this Task<T> task, TimeSpan timeout)
{
using (var cts = new CancellationTokenSource())
{
var delayTask = Task.Delay(timeout, cts.Token);
var resultTask = await Task.WhenAny(task, delayTask);
if (resultTask == delayTask)
{
// 取消非同步操作
throw new OperationCanceledException();
}
else
{
// 取消計時器任務
cts.Cancel();
}
return await task;
}
}10.使用StreamWriter(s)或Stream(s)時在Dispose之前建議先呼叫FlushAsync
當使用Stream和StreamWriter進行非同步寫入時,底層資料也有可能被緩衝,當資料被緩衝時,Stream和StreamWriter將使用同步的方式進行write/flush,這將會導致執行緒阻塞,並且有可能導致執行緒池內執行緒不足(執行緒池飢餓)
❌下面例子由於沒有呼叫FlushAsync(),所以最後是以同步方式進行write/flush的
public async static Task RunAsync()
{
using (var streamWriter = new StreamWriter(@"C:\資料\Blogs\Task"))
{
// 由於沒有呼叫FlushAsync,所以最後是以同步方式進行write/flush的
await streamWriter.WriteAsync("Hello World");
}
}☑️所以應該改為下面這樣,在Dispose之前呼叫FlushAsync()
public async static Task RunAsync()
{
using (var streamWriter = new StreamWriter(@"C:\資料\Blogs\Task"))
{
await streamWriter.WriteAsync("Hello World");
// 呼叫FlushAsync() 使其使用非同步write/flush
await streamWriter.FlushAsync();
}
}11.建議使用 async/await而不是直接返回Task
使用async/await 代替直接返回Task具有以上好處
- 非同步和同步的異常都被始終被規範為了非同步
- 程式碼更容易修改(例如:增加一個
using) - 非同步的方法診斷起來更加容易(例如:除錯,掛起)
- 丟擲的異常將自動包裝在返回的任務之中,而不是丟擲實際異常
❌下面這個錯誤的例子是將Task直接返回給了呼叫者
public Task<int> RunAsync()
{
return Task.FromResult(1 + 1);
}☑️所以應該使用async/await來代替返回Task
public async Task<int> RunAsync()
{
return await Task.FromResult(1 + 1);
}?使用async/await來代替返回Task時,還有效能上的考慮,雖然直接Task會更快,但是最終卻改變了非同步的行為,失去了非同步狀態機的一些好處
使用場景
1. 使用定時器回撥函式
❌下面例子使用一個返回值為void的非同步,將其傳遞給Timer進行,因此,如果其中任務丟擲異常,則整個程序將退出
public class Pinger
{
private readonly Timer _timer;
private readonly HttpClient _client;
public Pinger(HttpClient client)
{
_client = new HttpClient();
_timer = new Timer(Heartbeat, null, 1000, 1000);
}
public async void Heartbeat(object state)
{
await httpClient.GetAsync("http://mybackend/api/ping");
}
}❌下面例子將阻止計時器回撥,這有可能導致執行緒池中執行緒耗盡,這也是一個非同步差於同步的例子
public class Pinger
{
private readonly Timer _timer;
private readonly HttpClient _client;
public Pinger(HttpClient client)
{
_client = new HttpClient();
_timer = new Timer(Heartbeat, null, 1000, 1000);
}
public void Heartbeat(object state)
{
httpClient.GetAsync("http://mybackend/api/ping").GetAwaiter().GetResult();
}
}☑️下面例子是使用基於的非同步的方法,並在定時器回撥函式中丟棄該任務,並且如果此方法丟擲異常,則也不會關閉程序,而是會觸發TaskScheduler.UnobservedTaskException事件
2.建立回撥函式引數時注意避免 async void
假如有BackgroudQueue類中有一個接收回調函式的FireAndForget方法,該方法在某個時候執行呼叫
❌下面這個錯誤例子將強制呼叫者要麼阻塞要麼使用async void非同步方法
public class BackgroundQueue
{
public static void FireAndForget(Action action) { }
}static async Task Main(string[] args)
{
var httpClient = new HttpClient();
// 因為方法型別是Action,所以只能使用async void
BackgroundQueue.FireAndForget(async () =>
{
await httpClient.GetAsync("http://pinger/api/1");
});
}☑️所以應該構建一個回撥非同步方法的過載
public class BackgroundQueue
{
public static void FireAndForget(Action action) { }
public static void FireAndForget(Func<Task> action) { }
}3.使用ConcurrentDictionary.GetOrAdd注意場景
快取非同步結果是一種很常見的做法,ConcurrentDictionary是一個很好的集合,而GetOrAdd也是一個很方便的方法,它用於嘗試獲取已經存在的項,如果沒有則新增項.因為回撥是同步的,所以很容易編寫Task.Result的程式碼,從而生成非同步的結果值,但是這樣很容易導致執行緒池飢餓
❌下面這個例子就有可能導致執行緒池飢餓,因為當如果沒有快取人員資料時,將阻塞請求執行緒
public class PersonController : Controller
{
private AppDbContext _db;
private static ConcurrentDictionary<int, Person> _cache = new ConcurrentDictionary<int, Person>();
public PersonController(AppDbContext db)
{
_db = db;
}
public IActionResult Get(int id)
{
// 如果不存在快取資料,則會進入堵塞狀態
var person = _cache.GetOrAdd(id, (key) => db.People.FindAsync(key).Result);
return Ok(person);
}
}☑️可以改成快取執行緒本身,而不是結果,這樣將不會導致執行緒池飢餓
public class PersonController : Controller
{
private AppDbContext _db;
private static ConcurrentDictionary<int, Task<Person>> _cache = new ConcurrentDictionary<int, Task<Person>>();
public PersonController(AppDbContext db)
{
_db = db;
}
public async Task<IActionResult> Get(int id)
{
// 因為快取的是執行緒本身,所以沒有進行堵塞,也就不會產生執行緒池飢餓
var person = await _cache.GetOrAdd(id, (key) => db.People.FindAsync(key));
return Ok(person);
}
}?這種方法,在最後,GetOrAdd()可能並行多次來執行快取回撥,這可能導致啟動多次昂貴的計算
☑️可以使用async lazy模式來取代多次執行回撥問題
public class PersonController : Controller
{
private AppDbContext _db;
private static ConcurrentDictionary<int, AsyncLazy<Person>> _cache = new ConcurrentDictionary<int, AsyncLazy<Person>>();
public PersonController(AppDbContext db)
{
_db = db;
}
public async Task<IActionResult> Get(int id)
{
// 使用Lazy進行了延遲載入(使用時呼叫),解決了多次執行回撥問題
var person = await _cache.GetOrAdd(id, (key) => new AsyncLazy<Person>(() => db.People.FindAsync(key)));
return Ok(person);
}
private class AsyncLazy<T> : Lazy<Task<T>>
{
public AsyncLazy(Func<Task<T>> valueFactory) : base(valueFactory)
{
}
}
4.建構函式對於非同步的問題
建構函式是同步,下面看看在建構函式中處理非同步情況
下面是使用客戶端API的例子,當然,在使用API之前需要非同步進行連線
public interface IRemoteConnectionFactory
{
Task<IRemoteConnection> ConnectAsync();
}
public interface IRemoteConnection
{
Task PublishAsync(string channel, string message);
Task DisposeAsync();
}❌下面例子使用Task.Result在建構函式中進行連線,這有可能導致執行緒池飢餓和死鎖現象
public class Service : IService
{
private readonly IRemoteConnection _connection;
public Service(IRemoteConnectionFactory connectionFactory)
{
_connection = connectionFactory.ConnectAsync().Result;
}
}☑️正確的方式應該使用靜態工廠模式進行非同步連線
public class Service : IService
{
private readonly IRemoteConnection _connection;
private Service(IRemoteConnection connection)
{
_connection = connection;
}
public static async Task<Service> CreateAsync(IRemoteConnectionFactory connectionFactory)
{
return new Service(await connectionFactory.ConnectAsync());
}
}