Queues and Job Processing

Hand slow work to a queue so the request returns at once, start with a channel in process, and move to a broker when loss matters.

SyntaxChannel<T> queue = Channel.CreateBounded<T>(capacity);

Some requests ask for slow work. Resizing an upload, sending a welcome email, rebuilding a report. Doing it before you answer means the caller waits for all of it, and a timeout loses the whole thing. A queue breaks that apart. The handler writes a small item describing the work and returns 202 Accepted, which means accepted, not finished. Something else does the work.

A channel is the in-process queue

A Channel<T> is a queue that is safe to use from several threads at once, with an asynchronous reader and an asynchronous writer. Channel.CreateBounded<T>(100) gives you one that holds at most 100 unread items. Channel.CreateUnbounded<T>() has no limit. Prefer bounded. A full bounded channel makes WriteAsync wait, which is backpressure: the producer slows down to the consumer's speed instead of filling memory until the process dies.

Register the channel as a singleton, write to it from the endpoint, and read it from a BackgroundService. There are two ways to read, and they do the same thing. The short form is await foreach (string item in reader.ReadAllAsync(token)). The long form is while (await reader.WaitToReadAsync(token)) wrapped around while (reader.TryRead(out string? item)). You will meet both in other people's code.

When in-process is not enough

A channel lives in one process's memory. If that process restarts, is evicted or crashes, every queued item is gone, and the caller already got its 202. A channel also cannot spread work across replicas. When losing an item costs money, the queue must be durable and outside the process: a database table you poll, or a broker such as RabbitMQ, Amazon SQS or Azure Service Bus.

Example

Example · csharp
using System.Threading.Channels;

// A bounded channel is an in-process queue. Capacity 2 means a third unread
// item makes the writer wait, instead of letting memory grow without a limit.
Channel<string> queue = Channel.CreateBounded<string>(2);

// The consumer. In a real service this loop lives in a BackgroundService.
Task consumer = Task.Run(async () =>
{
    await foreach (string email in queue.Reader.ReadAllAsync())
    {
        Console.WriteLine($"sent welcome mail to {email}");
    }
});

// The producer. In a real service this is the request handler, which returns
// 202 Accepted as soon as the write completes.
string[] signups = ["[email protected]", "[email protected]", "[email protected]"];
foreach (string email in signups)
{
    await queue.Writer.WriteAsync(email);
}

queue.Writer.Complete();   // no more items, so ReadAllAsync ends
await consumer;

Console.WriteLine("queue drained");

// Output:
// sent welcome mail to [email protected]
// sent welcome mail to [email protected]
// sent welcome mail to [email protected]
// queue drained

When to use it

  • Return 202 from an upload endpoint and resize the image afterwards, so the client is not waiting on image work.
  • Send a welcome email after sign-up without letting a slow mail server fail the sign-up.
  • Buffer metric or audit rows in memory and flush them to the database in batches of 500.
  • Spread long PDF generation jobs across several replicas by claiming rows from a jobs table.

More examples

The whole thing: endpoint writes, hosted service reads

POST a JSON body of {"email":"[email protected]"} to /signups and the response is 202 Accepted with an empty body, returned before the email is sent. Reach for this when the work is cheap to redo and the caller only needs to know the request was taken. FullMode.Wait is what turns a full queue into a slow writer rather than a dropped job.

Example · csharp
using System.Threading.Channels;

WebApplicationBuilder builder = WebApplication.CreateBuilder(args);

// One channel shared by the endpoint and the worker, so it is a singleton.
builder.Services.AddSingleton(Channel.CreateBounded<EmailJob>(
    new BoundedChannelOptions(100)
    {
        FullMode = BoundedChannelFullMode.Wait,   // apply backpressure
        SingleReader = true                       // one consumer loop
    }));

builder.Services.AddHostedService<EmailWorker>();

WebApplication app = builder.Build();

app.MapPost("/signups", async (EmailJob job, Channel<EmailJob> queue, CancellationToken ct) =>
{
    await queue.Writer.WriteAsync(job, ct);
    return Results.Accepted();
});

app.Run();

public record EmailJob(string Email);

public sealed class EmailWorker(Channel<EmailJob> queue) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await foreach (EmailJob job in queue.Reader.ReadAllAsync(stoppingToken))
        {
            Console.WriteLine($"emailing {job.Email}");
        }
    }
}

The long form of the read loop, and batching

This is the same loop as await foreach over ReadAllAsync, written out. Use the long form when you want batches rather than single items, because one database round trip for 500 rows costs far less than 500 round trips. TryRead returning false just means nothing is waiting, so the outer loop goes back to waiting.

Example · csharp
using Microsoft.Extensions.Hosting;
using System.Threading.Channels;

namespace Metrics;

public sealed class MetricWriter(Channel<int> queue) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        List<int> batch = [];

        // WaitToReadAsync waits until at least one item is available.
        while (await queue.Reader.WaitToReadAsync(stoppingToken))
        {
            // TryRead never waits, so this drains everything ready right now.
            while (batch.Count < 500 && queue.Reader.TryRead(out int value))
            {
                batch.Add(value);
            }

            Console.WriteLine($"flushing {batch.Count} metrics");
            batch.Clear();
        }
    }
}

Making it survive a restart with a database table

Use a table like this when the work must not be lost and you do not want to run a broker yet, which covers most small services. The endpoint calls EnqueueAsync inside its own transaction, so the job is committed with the rest of the request. A claim with a lease is what lets several replicas poll the same table safely.

Example · csharp
using Microsoft.Extensions.Hosting;

namespace Shop;

public sealed record ThumbnailJob(long Id, string BlobPath);

// Backed by a jobs table. ClaimNextAsync is one UPDATE ... RETURNING that sets
// the row to 'running' with a lease, so two replicas cannot claim the same row.
public interface IJobStore
{
    Task<long> EnqueueAsync(string blobPath, CancellationToken cancellationToken);
    Task<ThumbnailJob?> ClaimNextAsync(CancellationToken cancellationToken);
    Task CompleteAsync(long id, CancellationToken cancellationToken);
    Task ReleaseAsync(long id, CancellationToken cancellationToken);
}

public sealed class ThumbnailWorker(IJobStore store) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        using PeriodicTimer timer = new(TimeSpan.FromSeconds(2));

        while (await timer.WaitForNextTickAsync(CancellationToken.None))
        {
            if (stoppingToken.IsCancellationRequested)
            {
                return;
            }

            ThumbnailJob? job = await store.ClaimNextAsync(stoppingToken);
            if (job is null)
            {
                continue;
            }

            try
            {
                Console.WriteLine($"resizing {job.BlobPath}");
                await store.CompleteAsync(job.Id, stoppingToken);
            }
            catch (Exception)
            {
                // Put it back so another pass picks it up.
                await store.ReleaseAsync(job.Id, stoppingToken);
                throw;
            }
        }
    }
}

Related

dotnet channels basicsdotnet channel backpressuredotnet background servicedotnet producer consumerdotnet cloud queues

Discussion

  • Be the first to comment on this lesson.