Параллельная библиотека задач Wainany Design

Question

Я только начал исследовать PTL и иметь проектный вопрос.

Мой сценарий: у меня есть список URL-адресов, которые каждый обращается к изображению. Я хочу, чтобы каждое изображение было загружено параллельно. Как только хотя бы один Изображение загружено, я хочу выполнить метод, который делает что-то с загруженным изображением. Этот метод не следует параллелизован - оно должно быть сериалом.

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

//From the Director class
Task<Image>[] downloadTasks = collector.RetrieveImages(listOfURLs);

for (int i = 0; i < listOfURLs.Count; i++)
{
    //Wait for any of the remaining downloads to complete
    int completedIndex = Task<Image>.WaitAny(downloadTasks);
    Image completedImage = downloadTasks[completedIndex].Result;

    //Now do something with the image (this "something" must happen serially)
    //Uses the "Formatter" class to accomplish this let's say
}

///////////////////////////////////////////////////

//From the Collector class
public Task<Image>[] RetrieveImages(List<string> urls)
{
    Task<Image>[] tasks = new Task<Image>[urls.Count];

    int index = 0;
    foreach (string url in urls)
    {
        string lambdaVar = url;  //Required... Bleh
        tasks[index] = Task<Image>.Factory.StartNew(() =>
            {
                using (WebClient client = new WebClient())
                {
                    //TODO: Replace with live image locations
                    string fileName = String.Format("{0}.png", i);
                    client.DownloadFile(lambdaVar, Path.Combine(Application.StartupPath, fileName));
                }

                return Image.FromFile(Path.Combine(Application.StartupPath, fileName));
            },
            TaskCreationOptions.LongRunning | TaskCreationOptions.AttachedToParent);

        index++;
    }

    return tasks;
}

Answer 1

Обычно вы используете Waitany ждать, пока одна задача, когда вы не заботитесь о результатах любого из других. Например, если вы только что заботитесь о первом изображении, который произошел, чтобы вернуться.

Как насчет этого вместо этого.

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

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Drawing;
using System.IO;
using System.Net;
using System.Threading.Tasks;

namespace ClassLibrary1
{
    public class Class1
    {
        readonly string _path = Directory.GetCurrentDirectory();

        public void Demo()
        {
            IList<string> listOfUrls = new List<string>();
            listOfUrls.Add("http://i3.codeplex.com/Images/v16821/editicon.gif");
            listOfUrls.Add("http://i3.codeplex.com/Images/v16821/favorite-star-on.gif");
            listOfUrls.Add("http://i3.codeplex.com/Images/v16821/arrow_dsc_green.gif");
            listOfUrls.Add("http://i3.codeplex.com/Images/v16821/editicon.gif");
            listOfUrls.Add("http://i3.codeplex.com/Images/v16821/favorite-star-on.gif");
            listOfUrls.Add("http://i3.codeplex.com/Images/v16821/arrow_dsc_green.gif");
            listOfUrls.Add("http://i3.codeplex.com/Images/v16821/editicon.gif");
            listOfUrls.Add("http://i3.codeplex.com/Images/v16821/favorite-star-on.gif");
            listOfUrls.Add("http://i3.codeplex.com/Images/v16821/arrow_dsc_green.gif");

            BlockingCollection<Image> images = new BlockingCollection<Image>();

            Parallel.Invoke(
                () =>                   // Task 1: load the images
                {
                    Parallel.For(0, listOfUrls.Count, (i) =>
                        {
                            Image img = RetrieveImages(listOfUrls[i], i);
                            img.Tag = i;
                            images.Add(img);    // Add each image to the queue
                        });
                    images.CompleteAdding();    // Done with images.
                },
                () =>                   // Task 2: Process images serially
                {
                    foreach (var img in images.GetConsumingEnumerable())
                    {
                        string newPath = Path.Combine(_path, String.Format("{0}_rot.png", img.Tag));
                        Console.WriteLine("Rotating image {0}", img.Tag);
                        img.RotateFlip(RotateFlipType.RotateNoneFlipXY);

                        img.Save(newPath);
                    }
                });
        }

        public Image RetrieveImages(string url, int i)
        {
            using (WebClient client = new WebClient())
            {
                string fileName = Path.Combine(_path, String.Format("{0}.png", i));
                Console.WriteLine("Downloading {0}...", url);
                client.DownloadFile(url, Path.Combine(_path, fileName));
                Console.WriteLine("Saving {0} as {1}.", url, fileName);
                return Image.FromFile(Path.Combine(_path, fileName));
            }
        } 
    }
}

Предупреждение: код не имеет никакой проверки ошибок или отмена. Поздно, и вам нужно что-то, что нужно сделать правильно? :)

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

Наша книга ... Вы можете прочитать больше об этом и других моделях для параллельного программирования в http://parallelpatterns.codeplex.com/Глава 7 охватывает трубопроводы и сопроводительные примеры показывают трубопроводы с обработкой ошибок и отмену.

Answer 2

TPL уже предоставляет функцию Storewith для выполнения одной задачи, когда другая отделка. Цепочка задач является одним из основных моделей, используемых в TPL для асинхронных операций.

Следующий метод загружает набор изображений и продолжает переименовать каждый из файлов

static void DownloadInParallel(string[] urls)
{
   var tempFolder = Path.GetTempPath();

   var downloads = from url in urls
                   select Task.Factory.StartNew<string>(() =>{
                       using (var client = new WebClient())
                       {
                           var uri = new Uri(url);
                           string file = Path.Combine(tempFolder,uri.Segments.Last());
                           client.DownloadFile(uri, file);
                           return file;
                       }
                   },TaskCreationOptions.LongRunning|TaskCreationOptions.AttachedToParent)
                  .ContinueWith(t=>{
                       var filePath = t.Result;
                       File.Move(filePath, filePath + ".test");
                  },TaskContinuationOptions.ExecuteSynchronously);

    var results = downloads.ToArray();
    Task.WaitAll(results);
}

Вы также должны проверить WebClient async задачи из образцов параллелетоксинсиозныхxtras. Методы расширения Downloadxxxtask обрабатывают как создание задач, так и асинхронной загрузки файлов.

Следующий метод использует расширение DownloadDataSask, чтобы получить данные изображения и повернуть его перед сохранением его на диск

static void DownloadInParallel2(string[] urls)
{
    var tempFolder = Path.GetTempPath();

    var downloads = from url in urls
         let uri=new Uri(url)
         let filePath=Path.Combine(tempFolder,uri.Segments.Last())
         select new WebClient().DownloadDataTask(uri)                                                        
         .ContinueWith(t=>{
            var img = Image.FromStream(new MemoryStream(t.Result));
            img.RotateFlip(RotateFlipType.RotateNoneFlipY);
            img.Save(filePath);
         },TaskContinuationOptions.ExecuteSynchronously);

    var results = downloads.ToArray();
    Task.WaitAll(results);
}

Answer 3

Лучший способ сделать это, вероятно, будет реализовать шаблон наблюдателя: у вас есть RetreiveImages Функция реализована Iobsibable, поместите свое «завершенное действие изображения» в Iobserver. объекты OnNext метод и подписаться RetreiveImages.

Я еще не пробовал этому (все еще должен играть больше с библиотекой задач), но я думаю, что это «правильный» способ сделать это.

Answer 4

// скачать все изображения

private async void GetAllImages ()
{
    var downloadTasks = listOfURLs.Where(url =>   !string.IsNullOrEmpty(url)).Select(async url =>
            {
                var ret = await RetrieveImage(url);
                return ret;
        }).ToArray();

        var counts = await Task.WhenAll(downloadTasks);
}

//From the Collector class
public async Task<Image> RetrieveImage(string url)
{
    var lambdaVar = url;  //Required... Bleh
    using (WebClient client = new WebClient())
    {
        //TODO: Replace with live image locations
        var fileName = String.Format("{0}.png", i);
        await client.DownloadFile(lambdaVar, Path.Combine(Application.StartupPath, fileName));
    }
    return Image.FromFile(Path.Combine(Application.StartupPath, fileName));
}