沒有按照正確順序阻塞並接收事件的上升事件

Question

這首先需要一些解釋。另外有一個上升的一些事件工作線程：

Task.Run(() => 
{ 
    for(int i = 0; i < 123456789; i++) 
    { 
     ... // some job 
     OnSomeEvent(i); 
    } 
}); 

瑞星事件同步將阻止工作，直到所有的事件處理程序已經完成：

void OnSomeEvent(int i) => SomeEvent?.Invoke(this, new SomeEventArgs(i)); 

的異步事件上升將不再阻止作業（耶）

void OnSomeEvent(int i) => Task.Run(() => SomeEvent?.Invoke(this, new SomeEventArgs(i))); 

，但現在還有另一個問題：在正確的順序沒有收到事件：

OnSomeEvent(1); 
OnSomeEvent(2); 
OnSomeEvent(3); 
... 

// event handler 
SomeEvent += (s, e) => Console.WriteLine(e.I); 

// possible output 
1 
3 
2 

問題：如何實現以正確的順序出現的異步事件上升？

最近我學會了什麼Dispatcher.InvokeAsyncuses queue。看起來我必須做類似的事情。如果我必須這樣做：1）是否應該是來電者的工作？2）我是否應該同步上升事件，並且接收者必須組織生產者/消費者以防止工作被阻止？或者，也許有另一種方式？

P.S .:這與ContinueWhith無關..除非存儲任務列表是一個適當的解決方案。我關心的是如何在以下情況下實施「即發即停」事件：a）呼叫者未被阻止2）以相同的順序收到事件。

P.P.S .:我不知道如何讓MCVE重現這個問題。它出現在真實的項目重UI，大量的線程等

Answer 1

您可以使用以異步操作添加到隊列以下TaskQueue所以當在隊列前面的項目完成，每一個開始：

public class TaskQueue 
{ 
    private Task previous = Task.FromResult(false); 
    private object key = new object(); 

    public Task<T> Enqueue<T>(Func<Task<T>> taskGenerator) 
    { 
     lock (key) 
     { 
      var next = previous.ContinueWith(t => taskGenerator()).Unwrap(); 
      previous = next; 
      return next; 
     } 
    } 
    public Task Enqueue(Func<Task> taskGenerator) 
    { 
     lock (key) 
     { 
      var next = previous.ContinueWith(t => taskGenerator()).Unwrap(); 
      previous = next; 
      return next; 
     } 
    } 
} 

這允許你寫：

private TaskQueue taskQueue = new TaskQueue(); 
private void OnSomeEvent(int i) => 
    taskQueue.Enqueue(() => Task.Run(() => SomeEvent?.Invoke(this, new SomeEventArgs(i)))); 

Answer 2

您可以使用ActionBlock隊列從TPL Dataflow保持事件的隊列。

你會如下創建隊列：

queue = new ActionBlock<SomeEventArgs>(item => SomeEvent?.Invoke(item)); 

那麼你就事件添加到隊列中，像這樣：

queue.Post(new SomeEventArgs(value)); 

當不再需要排隊你這樣做：

queue.Complete(); 

之後，如果您需要等待處理隊列中的任何項目，您可以這樣做：

queue.Completion.Wait(); 

但是，請注意queue.Completion事實上是一個Task所以你會經常與await使用它。

這裏是展示一種方法（不保持線程活着，只是爲了處理事件隊列）一個完整的例子：

using System; 
using System.Threading; 
using System.Threading.Tasks.Dataflow; 

namespace Demo 
{ 
    public class SomeEventArgs : EventArgs 
    { 
     public SomeEventArgs(int value) 
     { 
      Value = value; 
     } 

     public int Value { get; } 
    } 

    internal class Program 
    { 
     public delegate void SomeEventHandler(SomeEventArgs e); 

     public event SomeEventHandler SomeEvent; 

     ActionBlock<SomeEventArgs> queue; 

     private void run() 
     { 
      queue = new ActionBlock<SomeEventArgs>(item => SomeEvent?.Invoke(item)); 

      // Subscribe to my own event (this just for demonstration purposes!) 

      this.SomeEvent += Program_SomeEvent; 

      // Raise 100 events. 

      for (int i = 0; i < 100; ++i) 
      { 
       OnSomeEvent(i); 
       Console.WriteLine("Raised event " + i); 
      } 

      Console.WriteLine("Signalling that queue is complete."); 
      queue.Complete(); 

      Console.WriteLine("Waiting for queue to be processed."); 
      queue.Completion.Wait(); 

      Console.WriteLine("Done."); 
     } 

     private void Program_SomeEvent(SomeEventArgs e) 
     { 
      Console.WriteLine("Handled " + e.Value); 
      Thread.Sleep(1); // Simulate load. 
     } 

     private void OnSomeEvent(int value) 
     { 
      queue.Post(new SomeEventArgs(value)); 
     } 

     private static void Main() 
     { 
      new Program().run(); 
     } 
    } 
}