Использование библиотеки параллельных задач в асинхронном шаблоне на основе событий - PullRequest
Новая
       23

Использование библиотеки параллельных задач в асинхронном шаблоне на основе событий

12 голосов
/ 08 сентября 2011

Я пишу сетевое приложение.

Сообщения отправляются по транспорту как таковые: 

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
}

Мне не нравится идея прыгать по исходному коду, чтобы кодировать часть этого и часть этого. Это трудно понять и отредактировать.

Я слышал, что библиотека параллельных задач существенно упрощает это решение.

Однако многие из примеров, которые я прочитал с помощью библиотеки параллельных задач, были связаны с запуском цепочки активных задач. Под active я подразумеваю, что каждая задача может запускаться при явном вызове, например, так: 

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() );

Я прочитал эту статью , но я не совсем понимаю. Похоже, что мне нужно сделать с TaskCompletionSource. Если бы я хотел создать задачу из моего асинхронного шаблона, основанного на событиях, как приведенный выше блок кода, как бы он выглядел?

Ответы [ 2 ]

21 голосов
/ 08 сентября 2011

Вы правы насчет 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 и async очень помогут с этим)

Итак, этот код можно использовать так: 

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
}
3 голосов
/ 08 сентября 2011

Джереми Ликнесс имеет заголовок в блоге Сопрограммы для асинхронных последовательных рабочих процессов с использованием Reactive Extensions (Rx) , которые могут вас заинтересовать. Вот вопрос, на который он пытается ответить:

Концепция проста: часто мы хотим, чтобы асинхронный набор операций выполнялся последовательно. Возможно, вы должны загрузить список из службы, затем загрузить выбранный элемент, а затем запустить анимацию. Это можно сделать, связав завершенные события или вложив лямбда-выражения, но есть ли более чистый способ?

...