在基于事件的异步模式上使用任务并行库

Question

我正在编写联网应用程序。

消息通过传输发送，例如：

Network.SendMessage (new FirstMessage()); 

当此消息类型到达，像这样我就可以注册一个事件处理程序被称为：

Network.RegisterMessageHandler<FirstMessage> (OnFirstMessageReceived); 

而且会触发该事件：

public void OnFirstMessageReceived(EventArgs<FirstMessageEventArgs> e) 
{ 
} 

我正在为我的联网应用程序编写自定义身份验证过程，这需要大约五条消息才能完成。

不使用任务并行库，我将被迫在前面的事件处理程序的每个过程的下一步代码，就像这样：

public void OnFirstMessageReceived(EventArgs<FirstMessageEventArgs> e) 
{ 
    Network.SendMessage(new SecondMessage()); 
} 

public void OnSecondMessageReceived(EventArgs<SecondMessageEventArgs> e) 
{ 
    Network.SendMessage(new ThirdMessage()); 
} 

public void OnThirdMessageReceived(EventArgs<ThirdMessageEventArgs> e) 
{ 
    Network.SendMessage(new FourthMessage()); 
} 

public void OnFourthMessageReceived(EventArgs<FourthMessageEventArgs> e) 
{ 
    // Authentication is complete 
} 

我不喜欢周围跳跃的想法源代码来编码这部分和其中的一部分。这很难理解和编辑。

我听说任务并行库大大简化了这个解决方案。

但是，我使用任务并行库读取的许多示例都与启动活动的任务链有关。我的意思是“活跃”什么，是明确提出，当每个任务可以开始，像这样：

public void Drink() {} 
public void Eat() {} 
public void Sleep() {} 

Task.Factory.StartNew( () => Drink()) 
      .ContinueWith(() => Eat() ) 
      .ContinueWith(() => Sleep()); 

这是从我的基于事件的异步模式，其中每个事件处理方法仅称为相反当收到消息时。

换句话说，我不能做这样的事情（但我想）：

Task.Factory.StartNew( () => OnFirstMessageReceived() ) 
      .ContinueWith(() => OnSecondMessageReceived()) 
      .ContinueWith(() => OnThirdMessageReceived() ) 
      .ContinueWith(() => OnFourthMessageReceived()); 

我读过this article，但我不太明白。这似乎是我需要与TaskCompletionSource有关。如果我想从基于事件的异步模式（如上面的代码块）中完成任务，它会是什么样子？

Answer 1

对于TaskCompletionSource而言，它是将EAP（基于事件的异步模式）转换为TPL任务的关键。

这是记录在这里：https://docs.microsoft.com/en-us/dotnet/standard/parallel-programming/tpl-and-traditional-async-programming#exposing-complex-eap-operations-as-tasks

下面是简化代码：

public static class Extensions 
{ 
    public static Task<XDocument> GetRssDownloadTask(
     this WebClient client, Uri rssFeedUri) 
    { 
     // task completion source is an object, which has some state. 
     // it gives out the task, which completes, when state turns "completed" 
     // or else it could be canceled or throw an exception 
     var tcs = new TaskCompletionSource<XDocument>(); 

     // now we subscribe to completed event. depending on event result 
     // we set TaskCompletionSource state completed, canceled, or error 
     client.DownloadStringCompleted += (sender, e) => 
     { 
        if(e.Cancelled) 
        { 
         tcs.SetCanceled(); 
        } 
        else if(null != e.Error) 
        { 
         tcs.SetException(e.Error); 
        } 
        else 
        { 
         tcs.SetResult(XDocument.Parse(e.Result)); 
        } 
     }; 

     // now we start asyncronous operation 
     client.DownloadStringAsync(rssFeedUri); 
     // and return the underlying task immediately 
     return tcs.Task; 
    } 
} 

现在，所有你需要做的，使这些业务链，就是设置你的延续（这是不是现在很舒服，和C＃5的await和异步将很多帮助吧）

所以，这个代码可以使用这样的：

public static void Main() 
{ 
    var client = new WebClient(); 

    client.GetRssDownloadTask(
     new Uri("http://blogs.msdn.com/b/ericlippert/rss.aspx")) 
     .ContinueWith(t => { 
      ShowXmlInMyUI(t.Result); // show first result somewhere 
      // start a new task here if you want a chain sequence 
     }); 

    // or start it here if you want to get some rss feeds simultaneously 

    // if we had await now, we would add 
    // async keyword to Main method defenition and then 

    XDocument feedEric = await client.GetRssDownloadTask(
     new Uri("http://blogs.msdn.com/b/ericlippert/rss.aspx")); 
    XDocument feedJon = await client.GetRssDownloadTask(
     new Uri("http://feeds.feedburner.com/JonSkeetCodingBlog?format=xml")); 
    // it's chaining - one task starts executing after 
    // another, but it is still asynchronous 
} 

Answer 2

Jeremy Likness的博客文章标题Coroutines for Asynchronous Sequential Workflows using Reactive Extensions (Rx)可能会让你感兴趣。下面是他试图回答的问题：

这个概念很简单：我们经常需要一组异步操作来按顺序执行。也许你必须从服务中加载一个列表，然后加载选定的项目，然后触发一个动画。这可以通过链接已完成的事件或嵌套lambda表达式来完成，但有没有更清晰的方法？