Channels em C#: como evitar gargalos em APIs de alto volume

Channel limitado com BoundedChannelOptions

Para APIs de alto volume, o modelo limitado costuma ser a escolha mais segura entre os channels em C#. O exemplo abaixo cria um channel de auditoria com capacidade para mil itens.

using System.Threading.Channels;

// Criando um channel limitado (bounded)
var options = new BoundedChannelOptions(capacity: 1000)
{
    // Define o que acontece quando o channel está cheio
    FullMode = BoundedChannelFullMode.DropOldest, // Wait, DropNewest, DropOldest ou DropWrite
    SingleReader = true,  // Otimização quando existe apenas um consumer
    SingleWriter = false  // Vários producers (por exemplo, várias requisições simultâneas)
};

Channel<AuditLog> auditChannel = Channel.CreateBounded<AuditLog>(options);

As propriedades SingleReader e SingleWriter funcionam como promessas sobre o uso do channel. Quando você informa que existe um único leitor, a implementação interna consegue operar de forma mais eficiente.

ValueTask e menos alocações

Primeiro, os channels em C# usam ValueTask na maioria das operações. Como ValueTask é uma struct, ela pode evitar a criação de uma Task quando a operação termina de forma síncrona.

Isso reduz a alocação na heap. Em outras palavras, há ganho de performance, porque a pressão sobre o garbage collector diminui.

Backpressure nos channels em C#

Segundo, você pode limitar o tamanho da fila com os bounded channels, evitando falta de recursos. Quando o produtor escreve mais rápido do que o consumidor lê, o writer sofre backpressure, e você decide como agir.

• Wait: é o padrão, e o WriteAsync aguarda até liberar um espaço, enquanto o TryWrite retorna false na hora.
• DropNewest: remove o item mais novo da fila para abrir espaço para a escrita atual.
• DropOldest: remove o item mais antigo, ou seja, o começo da fila.
• DropWrite: descarta o próprio item que está sendo escrito.

A escolha depende do negócio. Para auditoria obrigatória, Wait protege os dados, enquanto para telemetria descartável um modo Drop mantém a API rápida mesmo sob pico.

Monitorando itens descartados

Descartar itens sem saber que isso aconteceu é perigoso. Por isso, os channels em C# limitados aceitam um callback chamado sempre que um item sai da fila por falta de espaço.

// O callback recebe o item descartado quando o channel está cheio
var channel = Channel.CreateBounded<AuditLog>(
    new BoundedChannelOptions(1000)
    {
        FullMode = BoundedChannelFullMode.DropOldest
    },
    static void (AuditLog dropped) =>
        Console.WriteLine($"Audit log dropped: {dropped.Path}"));

Na prática, esse callback é um bom lugar para incrementar uma métrica. Assim, um aumento repentino de descartes vira um alerta, e não uma surpresa no dia da investigação.

Definindo a fila de auditoria

Primeiro, crie um tipo para o evento e uma abstração para a fila. O channel fica encapsulado em uma classe registrada como singleton, para que producers e consumer compartilhem a mesma instância.

using System.Threading.Channels;

public sealed record AuditLog(string Method, string Path, int StatusCode, DateTime CreatedAt);

public interface IAuditQueue
{
    ChannelReader<AuditLog> Reader { get; }
    bool TryEnqueue(AuditLog log);
}

public sealed class AuditQueue : IAuditQueue
{
    private readonly Channel<AuditLog> _channel = Channel.CreateBounded<AuditLog>(
        new BoundedChannelOptions(1000)
        {
            FullMode = BoundedChannelFullMode.DropOldest,
            SingleReader = true,
            SingleWriter = false
        });

    public ChannelReader<AuditLog> Reader => _channel.Reader;

    // TryWrite não bloqueia a requisição, mesmo com a fila cheia
    public bool TryEnqueue(AuditLog log) => _channel.Writer.TryWrite(log);
}

Colocando os eventos no pipe

Em seguida, o middleware atua como producer dos channels em C# da aplicação. Ele apenas escreve o evento no channel e devolve a resposta, sem esperar pelo banco.

builder.Services.AddSingleton<IAuditQueue, AuditQueue>();
builder.Services.AddHostedService<AuditWorker>();

var app = builder.Build();

app.Use(async (context, next) =>
{
    await next(context);

    // Producer: registra o evento no channel e segue o fluxo
    var queue = context.RequestServices.GetRequiredService<IAuditQueue>();
    queue.TryEnqueue(new AuditLog(
        context.Request.Method,
        context.Request.Path,
        context.Response.StatusCode,
        DateTime.UtcNow));
});

Consumindo e gravando em lote

Você pode, inclusive, gravar os logs apenas quando a fila alcançar uma quantidade configurada de registros. Dessa forma, o consumer usa um bulk insert e evita idas desnecessárias ao banco de dados.

Contudo, convém configurar um tempo máximo de espera, para evitar que poucos eventos fiquem parados para sempre. O worker abaixo junta até 30 itens ou espera no máximo 10 segundos, o que acontecer primeiro.

public sealed class AuditWorker(
    IAuditQueue queue,
    IServiceScopeFactory scopeFactory,
    ILogger<AuditWorker> logger) : BackgroundService
{
    private const int BatchSize = 30;
    private readonly TimeSpan _maxWaitTime = TimeSpan.FromSeconds(10);

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var buffer = new List<AuditLog>(BatchSize);

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                // Token que expira após _maxWaitTime ou quando a aplicação para
                using var cts = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
                cts.CancelAfter(_maxWaitTime);

                try
                {
                    // WaitToReadAsync aguarda sem bloquear até que exista algo no channel
                    while (buffer.Count < BatchSize && await queue.Reader.WaitToReadAsync(cts.Token))
                    {
                        // TryRead aproveita o que já está em memória, sem esperar
                        while (buffer.Count < BatchSize && queue.Reader.TryRead(out var log))
                            buffer.Add(log);
                    }
                }
                catch (OperationCanceledException) when (!stoppingToken.IsCancellationRequested)
                {
                    // Tempo máximo atingido: grava o que já estiver no buffer
                }

                if (buffer.Count > 0)
                {
                    await FlushBatchAsync(buffer, stoppingToken);
                    buffer.Clear();
                }
            }
            catch (Exception ex) when (!stoppingToken.IsCancellationRequested)
            {
                logger.LogError(ex, "Failed to process audit batch");
                await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
            }
        }
    }

    private async Task FlushBatchAsync(List<AuditLog> logs, CancellationToken cancellationToken)
    {
        // O worker é singleton, então o repositório vem de um escopo próprio
        using var scope = scopeFactory.CreateScope();
        var repository = scope.ServiceProvider.GetRequiredService<IAuditRepository>();
        await repository.BulkInsertAsync(logs, cancellationToken);
    }
}

O IServiceScopeFactory aparece porque o Background Service vive como singleton. Portanto, dependências com escopo, como um DbContext, precisam de um escopo criado manualmente.

Leitura item a item com ReadAllAsync

Para uma leitura simples, item a item, use ReadAllAsync. Ele devolve um IAsyncEnumerable<T>, então cada evento recebido no channel é processado individualmente com await foreach.

protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
    // Processa cada evento assim que ele chega
    await foreach (var log in _reader.ReadAllAsync(stoppingToken))
        await _repository.InsertAsync(log, stoppingToken);
}

Leitura em lote com WaitToReadAsync

Para controlar manualmente a fila, use WaitToReadAsync, ideal para tratamento em lote. Com ele, o consumer aguarda até que exista pelo menos um item disponível antes de iniciar a leitura.

Depois disso, o TryRead retira tudo o que estiver disponível imediatamente, até o limite definido. Assim, cada ida ao banco carrega vários registros de uma só vez.

protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
    while (await _reader.WaitToReadAsync(stoppingToken))
    {
        var batch = new List<AuditLog>();

        // Pega o que estiver disponível imediatamente, até 50 itens
        while (batch.Count < 50 && _reader.TryRead(out var log))
            batch.Add(log);

        if (batch.Count > 0)
            await _repository.BulkInsertAsync(batch, stoppingToken);
    }
}

Esse laço termina sozinho quando o producer chama Complete no writer e a fila esvazia. Por isso, ele é a forma mais segura de consumir um channel que pode ser fechado.

Vários consumers lendo o mesmo channel

Quando um único consumer não dá conta do volume, os channels em C# também permitem vários leitores. Para isso, basta criar o channel com SingleReader = false e iniciar mais de uma tarefa de leitura.

// Três consumers dividem o trabalho do mesmo channel
var consumers = Enumerable.Range(0, 3)
    .Select(_ => ConsumeAsync(channel.Reader, stoppingToken))
    .ToArray();

await Task.WhenAll(consumers);

Cada item chega a apenas um dos consumers, então não há processamento duplicado. Entretanto, a ordem global deixa de ser garantida, já que cada leitor avança no seu próprio ritmo.