如何避免使用異步無效事件處理程序重入？

Question

在WPF應用程序中，我有一個通過網絡接收消息的類。每當該類的對象收到完整的消息時，就會引發一個事件。在應用程序的MainWindow中，我有一個訂閱該事件的事件處理程序。事件處理程序保證在應用程序的GUI線程上調用。

每當調用事件處理程序時，都需要將消息的內容應用於模型。這樣做可能會非常昂貴（在當前硬件上> 200ms）。這就是爲什麼使用Task.Run將消息應用於線程池的原因。

現在，可以非常接近地接收消息，因此可以調用事件處理程序，而前一個更改仍在處理中。確保消息只應用於一個最簡單的方法是什麼？到目前爲止，我已經想出了以下情況：

using System; 
using System.Threading.Tasks; 
using System.Windows; 

public partial class MainWindow : Window 
{ 
    private Model model = new Model(); 
    private Task pending = Task.FromResult<bool>(false); 

    // Assume e carries a message received over the network. 
    private void OnMessageReceived(object sender, EventArgs e) 
    { 
     this.pending = ApplyToModel(e); 
    } 

    private async Task ApplyToModel(EventArgs e) 
    { 
     await this.pending; 
     await Task.Run(() => this.model.Apply(e)); // Assume this is an expensive call. 
    } 
} 

這似乎按預期方式工作，但它也出現，這將不可避免地產生「內存泄漏」，因爲任務應用的消息總是會第一等待應用上一條消息的任務。如果是的話，那麼下面的變化應該避免泄漏：

private async Task ApplyToModel(EventArgs e) 
{ 
    if (!this.pending.IsCompleted) 
    { 
     await this.pending; 
    } 

    await Task.Run(() => this.model.Apply(e)); 
} 

這是一個明智的方式，以避免與異步無效的事件處理程序重入？

編輯：刪除OnMessageReceived中不必要的await this.pending;聲明。

編輯2：消息必須以接收它們的相同順序應用於模型。

Answer 1

我們需要感謝Stephen Toub，因爲他在博客系列中展示了一些非常有用的異步鎖定結構，包括async lock塊。

下面是從文章（包括從以前的文章中的一系列一些代碼）代碼：

public class AsyncLock 
{ 
    private readonly AsyncSemaphore m_semaphore; 
    private readonly Task<Releaser> m_releaser; 

    public AsyncLock() 
    { 
     m_semaphore = new AsyncSemaphore(1); 
     m_releaser = Task.FromResult(new Releaser(this)); 
    } 

    public Task<Releaser> LockAsync() 
    { 
     var wait = m_semaphore.WaitAsync(); 
     return wait.IsCompleted ? 
      m_releaser : 
      wait.ContinueWith((_, state) => new Releaser((AsyncLock)state), 
       this, CancellationToken.None, 
       TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); 
    } 

    public struct Releaser : IDisposable 
    { 
     private readonly AsyncLock m_toRelease; 

     internal Releaser(AsyncLock toRelease) { m_toRelease = toRelease; } 

     public void Dispose() 
     { 
      if (m_toRelease != null) 
       m_toRelease.m_semaphore.Release(); 
     } 
    } 
} 

public class AsyncSemaphore 
{ 
    private readonly static Task s_completed = Task.FromResult(true); 
    private readonly Queue<TaskCompletionSource<bool>> m_waiters = new Queue<TaskCompletionSource<bool>>(); 
    private int m_currentCount; 

    public AsyncSemaphore(int initialCount) 
    { 
     if (initialCount < 0) throw new ArgumentOutOfRangeException("initialCount"); 
     m_currentCount = initialCount; 
    } 
    public Task WaitAsync() 
    { 
     lock (m_waiters) 
     { 
      if (m_currentCount > 0) 
      { 
       --m_currentCount; 
       return s_completed; 
      } 
      else 
      { 
       var waiter = new TaskCompletionSource<bool>(); 
       m_waiters.Enqueue(waiter); 
       return waiter.Task; 
      } 
     } 
    } 
    public void Release() 
    { 
     TaskCompletionSource<bool> toRelease = null; 
     lock (m_waiters) 
     { 
      if (m_waiters.Count > 0) 
       toRelease = m_waiters.Dequeue(); 
      else 
       ++m_currentCount; 
     } 
     if (toRelease != null) 
      toRelease.SetResult(true); 
    } 
} 

現在把它應用到你的情況：

private readonly AsyncLock m_lock = new AsyncLock(); 

private async void OnMessageReceived(object sender, EventArgs e) 
{ 
    using(var releaser = await m_lock.LockAsync()) 
    { 
     await Task.Run(() => this.model.Apply(e)); 
    } 
} 

Answer 2

給定一個事件處理程序，它使用異步等待我們不能在任務 之外使用鎖，因爲每個事件調用的調用線程都是相同的，所以鎖總是讓它通過。

var object m_LockObject = new Object(); 

private async void OnMessageReceived(object sender, EventArgs e) 
{ 
    // Does not work 
    Monitor.Enter(m_LockObject); 

    await Task.Run(() => this.model.Apply(e)); 

    Monitor.Exit(m_LockObject); 
} 

但是，我們可以將任務內鎖，因爲Task.Run總是產生一個相同的並行線程，當一個事件調用OnMessageReceived它不能運行

var object m_LockObject = new Object(); 

private async void OnMessageReceived(object sender, EventArgs e) 
{ 
    await Task.Run(() => 
    { 
     // Does work 
     lock(m_LockObject) 
     { 
      this.model.Apply(e); 
     } 
    }); 
} 

所以一個新的任務返回immidiatly和型號。申請只能輸入一個接一個。