ASP.NET Core - Queued hosted service

ℹ️ Toto je pokračovanie k článku ASP.NET Core - Periodic Background Task.

Sú situácie kedy spracovanie dlhšie trvajúcej akcie nechcete spracovávať priamo počas spracovania požiadavky, ale chcete ju zaradiť do fronty a nechať ju spracovať neskôr. Toto je veľmi užitočné pri spracovaní veľkého množstva požiadaviek, kedy by ich priamé spracovanie (alebo ich časti) mohlo spomaliť celkový výkon aplikácie.
Pokiaľ používate cloudové služby ako AZURE alebo AWS, tak je možné že siahnete po ich riešeniach. V AZURE by to bola pravedpodobne kombinácia Service bus a AZURE Funkcie.

Ale je dobre vedieť, že podobné riešenie vieme implementovať aj priamo v našej ASP.NET Core službe.

Zadefinujme si rozhranie popisujúce jednoduchú frontu úloh:

// 👇 Simple interface for background task queue
public interface IBackgroundTaskQueue
{
    ValueTask Queue(Func<CancellationToken, ValueTask> workItem);

    ValueTask<Func<CancellationToken, ValueTask>> Dequeue(CancellationToken cancellationToken);
}

Implementácia fronty úloh na základe System.Threading.Channel môže vyzerať napríklad takto:

// 👇 Simple implementation of background task queue based on System.Threading.Channel
public class BackgroundTaskQueue : IBackgroundTaskQueue
{
    private readonly Channel<Func<CancellationToken, ValueTask>> _queue;

    public BackgroundTaskQueue(int capacity)
    {
        var options = new BoundedChannelOptions(capacity)
        {
            // 👇 Wait for the queue to have space
            FullMode = BoundedChannelFullMode.Wait
        };
        _queue = Channel.CreateBounded<Func<CancellationToken, ValueTask>>(options);
    }

    public async ValueTask Queue(Func<CancellationToken, ValueTask> workItem)
    {
        ArgumentNullException.ThrowIfNull(workItem);

        await _queue.Writer.WriteAsync(workItem);
    }

    public async ValueTask<Func<CancellationToken, ValueTask>> Dequeue(CancellationToken cancellationToken)
    {
        var workItem = await _queue.Reader.ReadAsync(cancellationToken);

        return workItem;
    }
}

Nakoniec vytvorme background service, ktorá bude postupne spracovávať úlohy z fronty:

// 👇 Hosted service that processes the queued work items
public class QueuedHostedService(
    IBackgroundTaskQueue taskQueue,
    ILogger<QueuedHostedService> logger) : BackgroundService
{
    private readonly ILogger<QueuedHostedService> _logger = logger;
    private readonly IBackgroundTaskQueue _taskQueue = taskQueue;

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            // 👇 Dequeue the work item
            var workItem =
                await _taskQueue.Dequeue(stoppingToken);

            try
            {
                // 👇 Execute the work item
                await workItem(stoppingToken);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error occurred executing: {WorkItem}.", nameof(workItem));
                throw;
            }
        }
    }
}

Nezabudnime zaregistrovať službu a frontu úloh v DI kontajnery:

// 👇 Register the hosted service and the background task queue in the DI container
builder.Services.AddSingleton<IBackgroundTaskQueue>(new BackgroundTaskQueue(30));
builder.Services.AddHostedService<QueuedHostedService>();

Použitie môže vyzerať napríklad tak, že miesto priameho spracovania objednávky v rámci endpointu ju zaradíme do fronty úloh:

// 👇 Endpoint for processing orders
app.MapPost("/orders", async (
    Order order, 
    [FromServices] IBackgroundTaskQueue backgroundTaskQueue, 
    [FromServices] ILogger<Order> logger) =>
{
    // 👇 Enqueue the work item
    await backgroundTaskQueue.Queue(async token =>
    {
        // 👇 Simulate processing the order
        logger.LogInformation("Processing order: {order}", order);
        await Task.Delay(1000, token);
        logger.LogInformation("Order processed: {order}", order);
    });

    return Results.Created($"/orders/{order.Id}", order);
});

Týmto spôsobom sme vytvorili jednoduchú frontu úloh, ktorá spracováva úlohy na pozadí. Toto riešenie je vhodné pre jednoduché scenáre.
Pri zložitejších scenároch potrebujete napríklad úlohy perzistovať aby v prípade výpadku ste ich vedeli spracovať.
Na ukladanie úloh môžete použiť riešenia ako Redis, AZURE Storage Account Queues alebo databázu.

Ak vám ani toto nestačí, tak vyskúšajte riešenia ako je napríklad Hangfire.