Как многопоточно обрабатывать данные из списка?

Question

Как на C# многопоточно обрабатывать данные из списка(очереди)?
Есть список url, с которых надо загрузить данные и их обработать.
Как правильно организовать код,что б создалось например только 10 потоков и каждый поток брал ссылку из списка,удалял её в списке или очереди,загружал данные и как выполнит действия-забирал следующий элемент списка. Ну и так,что б элемент списка блокировался на время доступа другого потока.

Как организовать это? Читаю семафоры,мьютексы,но пока не вник в это.

---

подкорректировал :
Я хотел многопоточность изначально, но лучше совместить с асинхронностью. То-есть нет смысла ждать ответ от удалённого сервера. Если в главном потоке делать, то он подвиснет. По-этому сдлав асинхронно- сам поток не будет ждать, когда будет ответ от сервера. Но на одном потоке выигрыша не будет. Если сразу запускаем например 10 потоков, то мы посылаем 10 запросов и ждём(внутри потока асинхронность используем и даже каждый десятый поток не напрягаем в момент ожидания,но асинхронность я могу сделать,а вот многопоточность в очереди пока что не могу),потом каждый поток,получив данные,выполняет действия и затем переходит к следующей порции. То-есть потоки не создаются миллион раз. Они создались сразу и выполняют задачи из очереди.

Создать 10 потоков и сказать каждому -> async GetMyData_and_Save_Data()

,где метод GetMyData_and_Save_Data() берёт ссылку из очереди , качает асинхронно(ну скачивание то я сделаю,это не сложно), ну и потом обрабатывает асинхронно, а потом повторяет, то-еть находится в цикле while, а как очередь будет пуста, то вылетает с метода и поток уничтожается.

Спасибо

Answer 1

Предлагаю данное решение, довольно таки простое, вместо List для _linkQueue можно использовать потокобезопасный ConcurrentQueue.

    private List<string> _linkQueue = new List<string>();
    private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(10);

    public async Task ProcessData()
    {
        var taskPool = new List<Task>(10);

        foreach (var link in _linkQueue)
        {
            await _semaphore.WaitAsync().ConfigureAwait(false);

            taskPool.Add(Task.Run(() =>
            {
                ProcessItem(link);
                _semaphore.Release();
            }));
        }

        await Task.WhenAll(taskPool.ToArray()).ConfigureAwait(false);
    }

UPD

А можно вообще с данным примером не использовать потокобезопасные коллекции.

Answer 2

На тему многопоточности и асинхронности есть много вопросов на ru.so, с очень хорошими и подробными ответами, но всё же, судя по ответам и комментариям в этой теме, нужно ещё раз всё объяснить.

Допустим имеется следующий код, выполняющий получение данных последовательно:

var urls = new List<string>
{
    @"http:\\ru.stackoverflow.com", @"http:\\yandex.ru", @"http:\\google.ru"
};

foreach (var url in urls)
{
    var client = new WebClient();
    var text = client.DownloadString(url);
    rtb.AppendText(url + " : " + text.Length + Environment.NewLine);
}

Как мы видим, есть список урлов из которых мы в цикле последовательно получаем текст и выводим в RichTextBox.

Если поместить этот код в баттон-клик формы WinForms и выполнить, можно легко заметить, что форма на время его выполнения замерзает.

---

Попробуем сделать многопоточность с использованием Parallel LINQ (в комментариях приведён один из вариантов):

Task.Run(() =>
{
    urls.AsParallel()
        .WithDegreeOfParallelism(10)
        .ForAll(url =>
        {
            var client = new WebClient();
            var text = client.DownloadString(url);
            rtb.Invoke(new MethodInvoker(() =>
                rtb.AppendText(url + " : " + text.Length + Environment.NewLine)));
        });
});

Task.Run здесь нужен по двум причинам: во-первых, чтобы не замерзала форма, во-вторых, чтобы не попасть в дедлок (нужно обращение к текстбоксу).

С одной стороны, это очень простой код. Всё распараллеливание берёт на себя система. С другой стороны, долгая операция в нашем код - именно запрос к серверам. Это IO-bound код. Сам по себе он выполняется асинхронно на IOCP. А мы насоздавали потоков, которые по сути не нужны. (Вообще, операционная система довольно умная и заметит долгое ожидание IO-операций и грамотно использует потоки, отдав их другим процессам, но не будем углубляться...)

Этот код станет иметь больше смысла, если, например, после получения данных они будут долго обрабатываться (CPU-bound - нагрузка на процессор):

var text = client.DownloadString(url);
VeryLongProcessData(text);

Тут потоки будут использоваться более эффективно, но первая часть кода пока далека от совершенства.

---

Сделаем получение данных асинхронно, но последовательно (метод пометить async).

foreach (var url in urls)
{
    var client = new WebClient();
    var text = await client.DownloadStringTaskAsync(url);
    rtb.AppendText(url + " : " + text.Length + Environment.NewLine);
}

Тут пока ещё идёт последовательная работа с серверами, но форма уже не замерзает.

Сделаем запросы асинхронно и одновременно:

var tasks = new List<Task<string>>();

foreach (var url in urls)
{
    var client = new WebClient();
    var task = client.DownloadStringTaskAsync(url);
    tasks.Add(task);
}

await Task.WhenAll(tasks);

foreach (var task in tasks)
{
    rtb.AppendText(task.Result.Length + Environment.NewLine);
}

В данном коде задачи создаются и помещаются в список. Обратите внимание, что в цикле нет await! Благодаря этому все запросы уходят на выполнение параллельно. А ожидание их завершения выполняется в методе WhenAll.

В этом примере не создаётся ни одного потока! Круто, правда?

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

---

Давайте теперь совместим асинхронное выполнение запросов и многопоточную обработку данных.

В данном конкретном случае очень удобно применить продолжение:

//var tasks = new List<Task>();
foreach (var url in urls)
{
    var client = new WebClient();

    var task = client.DownloadStringTaskAsync(url)
        .ContinueWith(
            t => rtb.AppendText(url + " : " + t.Result.Length + Environment.NewLine),
            CancellationToken.None,
            TaskContinuationOptions.LongRunning,
            TaskScheduler.FromCurrentSynchronizationContext());

    //tasks.Add(task);
}

//await Task.WhenAll(tasks);

Что мы имеем: сперва на каждый урл выполняется асинхронный запрос (без создания потока), а затем, после завершения запроса выполняется обработка данных в потоке.

Это весьма эффективно (и просто). Единственно, тут не сделано ограничение максимального количества задач/потоков.

В методе ContinueWith можно выполнить любые действия после завершения получения данных. Если это действительно долгие вычисления, добавляем опцию LongRunning. Так как у нас есть обращение к гуевому контролу, передаём шедулер с контекстом синхронизации (соответственно, если работы с гуем нет, он не нужен).

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

---

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

Answer 3

Непонятно, каким именно образом можно одновременно и удалить элемент из очереди, и заблокировать его от других потоков. Предлагаю такое решение

class Program
{
    private static ConcurrentQueue<string> _urlQueue = new ConcurrentQueue<string>();

    static void Main(string[] args)
    {
        Parallel.For(0,
            _urlQueue.Count,
            new ParallelOptions() {MaxDegreeOfParallelism = 10},
            _ =>
            {
                _urlQueue.TryDequeue(out var link);
                // your code
            });
    }
}

Answer 4

Используйте потокобезопасный список Queue.

        public const int _taskPoolSize = 10;
        private void MainThread() {
            var myData = GetData();
            var taskFact = new TaskFactory();
            var taskPool = new Task[_taskPoolSize];
            for(var i = 0; i < _taskPoolSize; i++) {
                taskPool[i] = taskFact.StartNew(() => {
                    while (myData.Count > 0) {
                        var data = myData.Dequeue();
                        //other code
                    }
                });
            }
            Task.WaitAll(taskPool);
        }

        public Queue<string> GetData() {
            return new Queue<string>();
        }