Привет, Хабр! При создании фоновых работ, например, через Hangfire, может быть актуально учитывать разделяемые ресурсы (например, базы данных, внешнюю API или файловую систему). Поскольку такие ресурсы являются ограниченными, возникает потребность управления количеством параллельно исполняемых задач без написания сложной логики. Интересующимся особенностями распределения ресурсов в Hangfire при помощи очередей — пожаловать под кат :)

Часто есть задачи, которые обращаются к одному и тому же ресурсу (например, к базе данных), и известно, что ресурс поддерживает не более N параллельных операций, например,

• база данных, которая способна обслужить максимум 2 тяжёлых запроса одновременно;

• ограниченные лицензии на внешние API (например, только 5 параллельных вызовов);

• работа с файловыми хранилищами или сетевыми дисками.

Вместо ручного контроля параллелизма (например, для C# это может быть SemaphoreSlim, блокировки и т. д.), можно использовать очереди Hangfire для распределения задач.

Текущая версия Hangfire (даже версия Hangfire GitHub, не Hangfire Pro) позволяет балансировать нагрузку через очереди Hangfire при наличии разделяемых ресурсов, что позволяет эффективно использовать возможности Hangfire и не писать новый код.

В Hangfire использование очередей в сочетании с созданием нескольких Hangfire серверов позволяет, например:

• изолировать типы задач друг от друга;

• сконфигурировать выполнение задач в рамках очередей;

• ограничить количество параллельных задач в пределах одной очереди через настройку WorkerCount.

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

Допустим, есть кластер из 3 ClickHouse нод, и причем первый ClickHouse ch-node-1 выдерживает максимум 2 параллельных запроса, второй ch-node-2 — максимум 1 параллельный запрос, а третий ClickHouse ch-node-3 — максимум 3 запроса.

Фактически, каждый ClickHouse — это разделяемый ресурс, поэтому имеет смысл не имплементировать вручную логику разделения таких ресурсов, а использовать встроенные средства Hangfire. Стоит отметить, что это частный кейс работы с Hangfire, и рассматриваются базовые возможности Hangfire, не дополнительные возможности.

Создадим в Hangfire 3 очереди — ch-node-1, ch-node-2,ch-node-3, и на каждую добавляем по 2 воркера. Также создадим 3 сервера Hangfire — по одному серверу на каждый ClickHouse, чтобы указать параметры паралельной обработки запросов.

app.AddHangfireServer(new BackgroundJobServerOptions
{
    Queues = [ "ch-node-1" ],
    WorkerCount = 2
});
app.AddHangfireServer(new BackgroundJobServerOptions
{
    Queues = [ "ch-node-2" ],
    WorkerCount = 1
});
app.AddHangfireServer(new BackgroundJobServerOptions
{
    Queues = [ "ch-node-3" ],
    WorkerCount = 3
});

Это можно представить в виде следующей схемы.

Видно, что максимум 6 воркеров, причем условия по максимальному количеству воркеров для каждого ClickHouse выполняются (ch-node-1 — максимум 2 воркера, ch-node-2 — максимум 1 воркер, ch-node-3 — максимум 3 воркера), т.е. корректно разделяются ресурсы.

Для добавления работы в очередь достаточно указать имя очереди в Enqueue:

    BackgroundJob.Enqueue<Services>("ch-node-1", x => x.DoWork());

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

Этот функционал можно имплементировать и вручную на C#, например, через редомендованный Microsoft семафор SemaphoreSlim для ограничения числа задач, очереди на основе BlockingCollection или TaskScheduler с кастомным пулом. Но это достаточно трудозатратно, сложнее поддерживать, не масштабируется по нескольким приложениям/нодам, а Hangfire решает такого вида вопросы из коробки.

Рассмотрим в качестве иллюстрации пример реализации через SemaphoreSlim. Для решения задачи, в которой требуется ограничить количество одновременных вызовов метода Services.DoWork() с использованием SemaphoreSlim, можно реализовать три отдельных семафора для трех ClickHouse (ch-node-1, ch-node-2, ch-node-3). Каждый семафор будет отвечать за управление числом разрешенных ресурсов для соответствующего ClickHouse.

static class Program
{
    private static readonly SemaphoreSlim SemaphoreClickHouseNode1 = new(2); // Для CH1 разрешены 2 ресурса
    private static readonly SemaphoreSlim SemaphoreClickHouseNode2 = new(1); // Для CH2 разрешен 1 ресурс
    private static readonly SemaphoreSlim SemaphoreClickHouseNode3 = new(3); // Для CH3 разрешены 3 ресурса

    public static async Task Main(string[] args)
    {
        // Пример запуска работы для каждого ClickHouse
        var tasks = new[]
        {
            RunClickHouseWork("ch-node-1"),
            RunClickHouseWork("ch-node-2"),
            RunClickHouseWork("ch-node-3")
        };

        await Task.WhenAll(tasks);
    }

    static async Task RunClickHouseWork(string clickHouseNodeName)
    {
        SemaphoreSlim semaphore = null;

        switch (clickHouseNodeName)
        {
            case "ch-node-1":
                semaphore = SemaphoreClickHouseNode1;
                break;
            case "ch-node-2":
                semaphore = SemaphoreClickHouseNode2;
                break;
            case "ch-node-3":
                semaphore = SemaphoreClickHouseNode3;
                break;
            default:
                throw new ArgumentException("Invalid ClickHouse node");
        }

        for (int i = 0; i < 5; i++) // Пример 5 операций на один ClickHouse 
        {
            await semaphore.WaitAsync();
            
            try
            {
                Console.WriteLine($"Starting work on {clickHouseNodeName} - Task {i}");
                await Services.DoWork(clickHouseNodeName, i); // Пример вызова работы
                Console.WriteLine($"Completed work on {clickHouseNodeName} - Task {i}");
            }
            finally
            {
                semaphore.Release();
            }
        }
    }
}

public static class Services
{
    // Пример асинхронной работы
    public static async Task DoWork(string clickHouseNodeName, int taskNumber)
    {
        // Эмуляция асинхронной работы
        await Task.Delay(1000);
    }
}

В этой реализации создаются три семафора SemaphoreClickHouseNode1, SemaphoreClickHouseNode2, и SemaphoreClickHouseNode3, каждый из которых контролирует доступ к ресурсу для соответствующих ClickHouse. Функция RunClickHouseWork запускает симуляцию работы по ClickHouse, соблюдая ограничения семафоров. Services.DoWork эмулирует заданную работу, здесь был использован Task.Delay в качестве примера асинхронной задачи. Также важно не забывать оборачивать операции с семафором в блок try-finally, чтобы обеспечить корректный возврат ресурсов семафору даже в случае возникновения исключений.

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

Однако нужно иметь в виду ограничения бесплатной версии Hangfire по сравнению с Hangfire Pro, например, батчи и продолжения (Batches & Continuations).

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