Очереди задач для C# .NET: библиотеки и примеры

Очереди задач для C# (.NET): от базовых примитивов до промышленных решений

Организация асинхронной обработки задач — одна из ключевых задач при разработке высоконагруженных приложений на C#. В экосистеме .NET существует несколько подходов: от встроенных примитивов синхронизации до мощных внешних библиотек. В этой статье мы рассмотрим основные инструменты для работы с очередями задач, их сильные стороны и примеры использования.

1. System.Threading.Channels — встроенный инструмент .NET

Начиная с .NET Core 2.1, в стандартной библиотеке появился класс System.Threading.Channels.Channel<T>. Это потокобезопасная очередь, оптимизированная для сценариев producer-consumer (производитель-потребитель).

Основные возможности:

Асинхронное чтение и запись (WriteAsync, ReadAsync)
Поддержка ограниченных и неограниченных очередей
Возможность завершить очередь для сигнала потребителям
Высокая производительность без блокировок

Пример использования:

using System.Threading.Channels;

var channel = Channel.CreateBounded<string>(100);

// Производитель
var producer = Task.Run(async () =>
{
    for (int i = 0; i < 10; i++)
    {
        await channel.Writer.WriteAsync($"Task {i}");
        await Task.Delay(100);
    }
    channel.Writer.Complete();
});

// Потребитель
var consumer = Task.Run(async () =>
{
    await foreach (var item in channel.Reader.ReadAllAsync())
    {
        Console.WriteLine($"Processing: {item}");
        await Task.Delay(200);
    }
});

await Task.WhenAll(producer, consumer);

Когда использовать: для внутренних очередей в одном процессе, когда не требуется персистентность или распределённая обработка.

2. Hangfire — продвинутая библиотека фоновых задач

Hangfire — одна из самых популярных библиотек для управления фоновыми задачами в .NET. Она поддерживает очереди, планирование, повторные попытки и мониторинг.

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

Персистентность (SQL Server, Redis, PostgreSQL и др.)
Веб-интерфейс для мониторинга
Поддержка отложенных и повторяющихся задач
Автоматические повторные попытки при сбоях

Пример настройки и использования:

// Установка: Install-Package Hangfire
// В Startup.cs
public void ConfigureServices(IServiceCollection services)
{
    services.AddHangfire(config =>
        config.UseSqlServerStorage(connectionString));
    services.AddHangfireServer();
}

// Использование
public class EmailService
{
    public void SendWelcomeEmail(int userId)
    {
        // Логика отправки email
        Console.WriteLine($"Sending email to user {userId}");
    }
}

// Постановка задачи в очередь
BackgroundJob.Enqueue<EmailService>(x => x.SendWelcomeEmail(42));

// Отложенная задача
BackgroundJob.Schedule<EmailService>(
    x => x.SendWelcomeEmail(43), 
    TimeSpan.FromMinutes(5)
);

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

3. RabbitMQ.Client — распределённая очередь сообщений

Для распределённых систем, где несколько сервисов должны обмениваться задачами, используется RabbitMQ. Библиотека RabbitMQ.Client предоставляет полный доступ к протоколу AMQP.

Базовый пример producer-consumer:

using RabbitMQ.Client;
using System.Text;

// Producer
var factory = new ConnectionFactory() { HostName = "localhost" };
using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();

channel.QueueDeclare(queue: "task_queue",
                     durable: true,
                     exclusive: false,
                     autoDelete: false);

string message = "Process order #123";
var body = Encoding.UTF8.GetBytes(message);

channel.BasicPublish(exchange: "",
                     routingKey: "task_queue",
                     basicProperties: null,
                     body: body);

// Consumer
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
    var body = ea.Body.ToArray();
    var msg = Encoding.UTF8.GetString(body);
    Console.WriteLine($"Received: {msg}");
    channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
};

channel.BasicConsume(queue: "task_queue",
                     autoAck: false,
                     consumer: consumer);

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