Работа с очередями задач в C#: практическое руководство

Введение: зачем нужны очереди задач в C#?

В современной разработке на C# (.NET) часто возникает необходимость организовать асинхронную обработку данных. Классический пример — веб-сервер, который принимает запросы и должен обрабатывать их в фоне, не блокируя основной поток. Или приложение, которое загружает файлы из интернета и параллельно обрабатывает уже скачанные. Для таких сценариев идеально подходит паттерн «Producer-Consumer» (производитель-потребитель), а его основа — очередь задач.

Очередь задач позволяет безопасно передавать работу от одного потока (или асинхронной операции) к другому, синхронизируя доступ к общим данным. В этой статье мы рассмотрим три основных способа реализации очередей в C#: ConcurrentQueue, BlockingCollection и современный System.Threading.Channels. Каждый из них подходит для разных сценариев, и мы разберем их на практических примерах.

1. ConcurrentQueue — базовая потокобезопасная очередь

ConcurrentQueue<T> — это простейшая потокобезопасная очередь из пространства имен System.Collections.Concurrent. Она использует lock-free алгоритмы (сравнение и обмен, CAS) для обеспечения высокой производительности. Однако она не блокирует поток, когда очередь пуста — вам придется самостоятельно реализовать ожидание.

Пример: загрузка данных из нескольких источников

Допустим, у нас есть задача: скачать 10 веб-страниц и обработать их. Один поток (Producer) добавляет URL в очередь, а несколько потоков (Consumers) забирают их и скачивают.

using System;
using System.Collections.Concurrent;
using System.Threading.Tasks;

public class ConcurrentQueueExample
{
    public static async Task RunAsync()
    {
        var queue = new ConcurrentQueue<string>();
        var urls = new[] { "https://example.com", "https://dotnet.microsoft.com" };

        // Producer: добавляем задачи
        foreach (var url in urls)
        {
            queue.Enqueue(url);
        }

        // Consumers: обрабатываем задачи
        var tasks = new Task[3];
        for (int i = 0; i < tasks.Length; i++)
        {
            tasks[i] = Task.Run(() =>
            {
                while (queue.TryDequeue(out var url))
                {
                    Console.WriteLine($"Обработка: {url}");
                    // Симуляция работы
                    Task.Delay(500).Wait();
                }
            });
        }

        await Task.WhenAll(tasks);
        Console.WriteLine("Все задачи выполнены.");
    }
}

Плюсы: простота, высокая производительность при большом количестве операций.
Минусы: нет встроенного механизма ожидания (нужен цикл с проверкой), сложно ограничить размер очереди.

2. BlockingCollection — очередь с блокировкой и лимитом

BlockingCollection<T> — это обертка над потокобезопасной коллекцией, которая добавляет возможность блокировать поток-потребитель, пока очередь пуста, и поток-производитель, пока очередь переполнена. По умолчанию используется ConcurrentQueue, но можно подставить любую коллекцию, реализующую IProducerConsumerCollection<T>.

Пример: обработка заказов с ограничением буфера

Представьте систему обработки заказов. Производитель генерирует заказы, а три потребителя обрабатывают их. Буфер ограничен 10 элементами, чтобы не потреблять слишком много памяти.

using System;
using System.Collections.Concurrent;
using System.Threading.Tasks;

public class BlockingCollectionExample
{
    public static async Task RunAsync()
    {
        using var orders = new BlockingCollection<string>(boundedCapacity: 10);

        // Producer
        var producer = Task.Run(() =>
        {
            for (int i = 1; i <= 20; i++)
            {
                orders.Add($"Заказ #{i}");
                Console.WriteLine($"Добавлен заказ #{i}");
                Task.Delay(200).Wait();
            }
            orders.CompleteAdding(); // Сигнал, что добавление закончено
        });

        // Consumers
        var consumers = new Task[3];
        for (int i = 0; i < consumers.Length; i++)
        {
            int consumerId = i + 1;
            consumers[i] = Task.Run(() =>
            {
                foreach (var order in orders.GetConsumingEnumerable())
                {
                    Console.WriteLine($"Потребитель {consumerId} обрабатывает: {order}");
                    Task.Delay(1000).Wait(); // Симуляция обработки
                }
            });
        }

        await Task.WhenAll(producer, Task.WhenAll(consumers));
        Console.WriteLine("Все заказы обработаны.");
    }
}

Ключевые особенности:

• GetConsumingEnumerable() — блокирует поток, пока не появится новый элемент или не будет вызван CompleteAdding().
• boundedCapacity — ограничивает размер очереди. Если очередь заполнена, Add() заблокирует производителя.
• Поддерживает отмену через CancellationToken.

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

3. System.Threading.Channels — современный и производительный подход

Начиная с .NET Core 3.0 (и .NET 5+), Microsoft представила System.Threading.Channels. Это асинхронная очередь, оптимизированная для работы с async/await. Она поддерживает как однопроизводительные/однопотребительные сценарии (для максимальной производительности), так и многопоточные.

Пример: асинхронная обработка логов

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

using System;
using System.Threading.Channels;
using System.Threading.Tasks;

public class ChannelExample
{
    public static async Task RunAsync()
    {
        // Создаем неограниченный канал
        var channel = Channel.CreateUnbounded<string>(new UnboundedChannelOptions
        {
            SingleWriter = false, // Много производителей
            SingleReader = true   // Один потребитель
        });

        // Producer 1
        var producer1 = Task.Run(async () =>
        {
            for (int i = 1; i <= 10; i++)
            {
                await channel.Writer.WriteAsync($"Лог от Producer1: событие #{i}");
                await Task.Delay(100);
            }
        });

        // Producer 2
        var producer2 = Task.Run(async () =>
        {
            for (int i = 1; i <= 10; i++)
            {
                await channel.Writer.WriteAsync($"Лог от Producer2: событие #{i}");
                await Task.Delay(150);
            }
        });

        //