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

Введение в очереди задач в C#

В современном многопоточном программировании очередь задач (task queue) является фундаментальным паттерном для организации асинхронной обработки данных. В C# (.NET) этот паттерн реализуется через несколько мощных механизмов, которые позволяют разработчикам эффективно управлять потоками выполнения и распределять нагрузку.

Очередь задач решает классическую проблему producer-consumer (производитель-потребитель), где одна часть кода создаёт задания, а другая — их обрабатывает. В этой статье мы рассмотрим основные способы реализации очередей задач в C#: от простых коллекций до современных каналов (Channel).

Основные подходы к реализации очередей задач

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

Самая простая и низкоуровневая реализация — ConcurrentQueue<T> из пространства имён System.Collections.Concurrent. Она предоставляет потокобезопасные методы Enqueue и TryDequeue.

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

public class SimpleTaskQueue
{
    private readonly ConcurrentQueue<Action> _tasks = new();
    private readonly Task _worker;

    public SimpleTaskQueue()
    {
        _worker = Task.Run(ProcessTasks);
    }

    public void Enqueue(Action task)
    {
        _tasks.Enqueue(task);
    }

    private void ProcessTasks()
    {
        while (true)
        {
            if (_tasks.TryDequeue(out var task))
            {
                task();
            }
        }
    }
}

Преимущества:

• Простота использования
• Низкое потребление памяти
• Потокобезопасность без блокировок

Недостатки:

• Активное ожидание (busy waiting) в цикле
• Отсутствие сигнализации о новых задачах
• Сложность остановки

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

BlockingCollection<T> решает проблему активного ожидания. Она предоставляет методы Add и Take, которые блокируют поток, пока данные не появятся.

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

public class BlockingTaskQueue : IDisposable
{
    private readonly BlockingCollection<Action> _tasks = new(new ConcurrentQueue<Action>());
    private readonly Task[] _workers;
    private bool _disposed;

    public BlockingTaskQueue(int workerCount = 4)
    {
        _workers = new Task[workerCount];
        for (int i = 0; i < workerCount; i++)
        {
            _workers[i] = Task.Run(ProcessTasks);
        }
    }

    public void Enqueue(Action task)
    {
        _tasks.Add(task);
    }

    private void ProcessTasks()
    {
        foreach (var task in _tasks.GetConsumingEnumerable())
        {
            if (_disposed) break;
            task();
        }
    }

    public void CompleteAdding()
    {
        _tasks.CompleteAdding();
    }

    public void Dispose()
    {
        _disposed = true;
        _tasks.CompleteAdding();
        Task.WaitAll(_workers);
        _tasks.Dispose();
    }
}

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

• Автоматическая блокировка потока при отсутствии задач
• Поддержка отмены через CancellationToken
• Возможность ограничить вместимость очереди
• Безопасное завершение через CompleteAdding

Современный подход: Channel<T>

Начиная с .NET Core 3.0 и .NET 5, в C# появился System.Threading.Channels.Channel<T> — это наиболее производительная и гибкая реализация очереди задач. Она поддерживает асинхронные операции, отмену и ограничение пропускной способности.

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

public class ChannelTaskQueue
{
    private readonly Channel<Func<Task>> _channel;
    private readonly Task _consumer;

    public ChannelTaskQueue(int capacity = 100)
    {
        var options = new BoundedChannelOptions(capacity)
        {
            FullMode = BoundedChannelFullMode.Wait,
            SingleWriter = false,
            SingleReader = false
        };
        
        _channel = Channel.CreateBounded<Func<Task>>(options);
        _consumer = Task.Run(ProcessTasksAsync);
    }

    public async ValueTask EnqueueAsync(Func<Task> taskFactory)
    {
        await _channel.Writer.WriteAsync(taskFactory);
    }

    private async Task ProcessTasksAsync()
    {
        await foreach (var taskFactory in _channel.Reader.ReadAllAsync())
        {
            await taskFactory();
        }
    }

    public void Complete()
    {
        _channel.Writer.Complete();
    }
}

// Пример использования
public class ExampleUsage
{
    public async Task RunAsync()
    {
        var queue = new ChannelTaskQueue(10);
        
        // Производитель
        var producer = Task.Run(async () =>
        {
            for (int i = 0; i < 100; i++)
            {
                int taskId = i;
                await queue.EnqueueAsync(async () =>
                {
                    await Task.Delay(100);
                    Console.WriteLine($"Выполнена задача #{taskId}");
                });
            }
        });

        await producer;
        queue.Complete();
    }
}

Преимущества Channel:

• Полная асинхронность (await/async)
• Поддержка bounded (ограниченной) и unbounded (неограниченной) очередей
• Встроенная поддержка отмены через CancellationToken
• Высокая производительность за счёт lock-free структур
• Гибкие настройки поведения при переполнении

Настройка Channel через ChannelOptions

При создании канала можно тонко настроить его поведение:

var options = new BoundedChannelOptions(100)
{
    // Режим при переполнении
    FullMode = BoundedChannelFullMode.DropWrite,  // или DropOldest, DropNewest, Wait
    
    // Оптимизация для одного производителя/потребителя
    SingleWriter = true,
    SingleReader = false,
    
    // Разрешить запись после завершения
    AllowSynchronousContinuations = false
};

var channel = Channel.CreateBounded<int>(options);

Продвинутые сценарии использования

Очередь с приоритетами

Иногда требуется обрабатывать задачи в порядке их приоритета. Реализовать это можно с помощью PriorityQueue<TElement, TPriority> (доступна с .NET 6):

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

public class PriorityTaskQueue
{
    private readonly PriorityQueue<Func<Task>, int> _queue = new();
    private readonly SemaphoreSlim _semaphore