Dispatch Concurrency, ConsumerDispatchConcurrency, Backpressure e Controle de Carga em Consumers | RabbitMQ

Dispatch Concurrency, ConsumerDispatchConcurrency, Backpressure e Controle de Carga em Consumers | RabbitMQ

1. Visão Geral

RabbitMQ é amplamente utilizado como buffer, desacoplador e controlador de fluxo entre produtores e consumidores. Em sistemas que lidam com picos de tráfego, especialmente quando há operações custosas (ex.: gravações em banco de dados), o controle correto da concorrência de consumo é crítico.

Esta documentação descreve:

Como o RabbitMQ despacha mensagens
Como o C# realmente processa essas mensagens
Onde normalmente ocorrem erros de concorrência
Como projetar consumers seguros para o banco
Padrões recomendados para ambientes de produção

2. Conceito Fundamental: Dispatch Concurrency

Dispatch Concurrency é a combinação de três fatores independentes:

Prefetch (BasicQos) — controle do broker
Modelo de consumer — como o RabbitMQ entrega mensagens
Modelo de execução do código — paralelismo real no C#

Controlar apenas um desses fatores não garante controle de carga.

3. Nível 1 — Prefetch (`BasicQos`)

channel.BasicQos(
    prefetchSize: 0,
    prefetchCount: N,
    global: false);

O que o Prefetch controla

Número máximo de mensagens não confirmadas (unacked) por consumidor
Quantas mensagens o RabbitMQ pode entregar antecipadamente

O que o Prefetch não controla

Ordem de execução
Paralelismo no código
Quantas threads/processos gravam no banco

Interpretação correta

Prefetch	Efeito real
`1`	Uma mensagem por vez
`5`	Até 5 mensagens “em voo”
`50`	Potencialmente 50 operações simultâneas

Prefetch não garante processamento sequencial, apenas limita entrega.

4. Nível 2 — Modelo de Consumer em C#

`AsyncEventingBasicConsumer`

consumer.ReceivedAsync += HandleMessageAsync;

Comportamento real

Cada mensagem recebida pode disparar uma execução paralela
O paralelismo máximo é limitado pelo prefetchCount
O RabbitMQ não serializa o handler

prefetch = 10
↓
10 mensagens entregues
↓
10 HandleMessageAsync executando em paralelo
↓
10 writes no banco

Este é o ponto mais ignorado em projetos RabbitMQ.

5. Nível 3 — Código do Consumer (C#)

Mesmo com prefetch baixo, o código pode introduzir paralelismo acidental:

Task.Run
Parallel.ForEach
Múltiplos consumers no mesmo channel
Múltiplos channels sem controle
Repositório não preparado para concorrência

6. Backpressure: o verdadeiro papel do RabbitMQ

RabbitMQ não é um processador, ele é um controlador de fluxo.

O backpressure funciona assim:

Consumer não dá ACK
↓
RabbitMQ para de enviar mensagens
↓
Fila cresce
↓
Sistema absorve pico sem derrubar dependências

Esse mecanismo só funciona corretamente quando:

autoAck = false
ACK acontece após sucesso real
Concorrência é controlada no consumer

7. Padrões de Consumo Recomendados

7.1 Consumo Sequencial (Banco Sensível)

Use quando:

Banco sofre com concorrência
Throughput não é prioridade

channel.BasicQos(0, 1, false);

Um consumidor
Um channel
Um write por vez

7.2 Consumo Concorrente Controlado (Recomendado)

Use SemaphoreSlim para limitar concorrência real:

private static readonly SemaphoreSlim _semaphore = new(2);

consumer.ReceivedAsync += async (_, ea) =>
{
    await _semaphore.WaitAsync();

    try
    {
        await ProcessarAsync(ea);
        await channel.BasicAckAsync(ea.DeliveryTag, false);
    }
    catch
    {
        await channel.BasicNackAsync(ea.DeliveryTag, false, true);
    }
    finally
    {
        _semaphore.Release();
    }
};

RabbitMQ despacha
Código limita concorrência
Banco protegido

7.3 Processamento em Batch (Alta Eficiência)

Use quando:

Muitas mensagens pequenas
Banco suporta bulk

Estratégia:

Buffer de mensagens
Flush por tempo ou quantidade
BulkWriteAsync (MongoDB)

Resultado:

Menos IO
Menos lock
Maior throughput

8. Estratégias de ACK / NACK

ACK

Apenas após sucesso real
Nunca antecipado

NACK sem requeue

Erro de validação
Payload inválido
Violação de regra de negócio

NACK com requeue

Timeout
Falha transitória
Banco indisponível

catch (MongoException ex) when (IsTransient(ex))
{
    BasicNack(requeue: true);
}

9. Dead Letter Queue (DLQ)

Use DLQ para:

Mensagens inválidas
Erros definitivos
Auditoria e replay manual

Nunca use DLQ como:

Retry infinito
Fallback automático

10. Conexões, Channels e Thread Safety

Conexão (`IConnection`)

Thread-safe
Deve ser Singleton

Channel (`IChannel`)

Não é thread-safe
Um por consumidor
Nunca compartilhar entre handlers concorrentes

11. Ciclo de Vida e Dispose

Regras

Consumers não devem chamar Dispose
O Host / DI gerencia lifecycle
IAsyncDisposable é obrigatório

SIGTERM
↓
CancellationToken cancelado
↓
Worker finaliza
↓
DI chama DisposeAsync
↓
RabbitMQ fecha canal e conexão

12. Anti-padrões Comuns

Prefetch alto “porque aguenta”
Vários consumers sem controle
ACK antecipado
Paralelismo cego
Retry infinito
Misturar lógica de negócio no consumer

13. Checklist de Produção

autoAck = false
Prefetch definido conscientemente
Concorrência limitada no código
ACK após sucesso
DLQ configurada
MongoClient Singleton
Channel não compartilhado
Dispose controlado pelo Host

14. Regra de Ouro

RabbitMQ controla entrega. O consumer controla concorrência. O banco sente a soma dos dois.

15. DispatchConsumersAsync vs ConsumerDispatchConcurrency (Atualização Importante)

⚠️ Atenção: esta seção atualiza conceitos para versões modernas do RabbitMQ.Client (≥ 6.x / 7.x).

O que mudou na biblioteca

Nas versões antigas do RabbitMQ.Client, era necessário configurar explicitamente:

factory.DispatchConsumersAsync = true;

Essa flag indicava que o consumer deveria ser executado em um contexto assíncrono real, evitando bloqueio de threads quando se utilizava async/await.

A partir das versões mais recentes:

DispatchConsumersAsync foi removido da API
O despacho assíncrono passou a ser comportamento padrão
Não existe mais o risco de "async falso" ou deadlock causado por essa configuração

Ou seja:

Se você usa versões novas, não existe mais nada a configurar para habilitar async.

O novo controle: `ConsumerDispatchConcurrency`

factory.ConsumerDispatchConcurrency = 1;

Essa configuração define:

Quantos handlers de consumer podem ser executados em paralelo por conexão.

Ela controla diretamente o paralelismo real de execução, e substitui conceitualmente a preocupação antiga com DispatchConsumersAsync.

Relação entre Prefetch e Dispatch Concurrency

É fundamental entender que:

Concorrência máxima ≈ prefetch × ConsumerDispatchConcurrency

Exemplo:

prefetch = 10
ConsumerDispatchConcurrency = 4

Resultado potencial:

Até 40 handlers executando em paralelo
Até 40 operações simultâneas no banco

📌 Esse é hoje o erro mais comum em produção.

PrefetchCount = 5 + ConsumerDispatchConcurrency = 5

┌──────────────────────────┐
│      RabbitMQ Queue      │
└───────────┬──────────────┘
            │
            │  Prefetch = 5
            │  (5 mensagens em voo)
            ▼
┌──────────────────────────┐
│        Consumer          │
│  ConsumerDispatch = 5    │
└───────────┬──────────────┘
            │
            │  Dispatch paralelo
            ▼
┌───────┬───────┬───────┬───────┬───────┐
│Task 1 │Task 2 │Task 3 │Task 4 │Task 5 │
└───┬───┴───┬───┴───┬───┴───┬───┴───┬───┘
    │       │       │       │       │
    ▼       ▼       ▼       ▼       ▼
 DB Write DB Write DB Write DB Write DB Write

PrefetchCount = 4 + ConsumerDispatchConcurrency = 5

┌──────────────────────────┐
│      RabbitMQ Queue      │
└───────────┬──────────────┘
            │
            │  Prefetch = 4
            │  (4 mensagens em voo)
            ▼
┌──────────────────────────┐
│        Consumer          │
│  ConsumerDispatch = 5    │
└───────────┬──────────────┘
            │
            │  Dispatch paralelo
            ▼
┌───────┬───────┬───────┬───────┬─────────┐
│Task 1 │Task 2 │Task 3 │Task 4 │ (idle)  │
└───┬───┴───┬───┴───┬───┴───┬───┴─────────┘
    │       │       │       │
    ▼       ▼       ▼       ▼
 DB Write DB Write DB Write DB Write

PrefetchCount = 10 + ConsumerDispatchConcurrency = 1

┌──────────────────────────┐
│      RabbitMQ Queue      │
└───────────┬──────────────┘
            │
            │  Prefetch = 10
            │  (10 mensagens em voo)
            ▼
┌──────────────────────────┐
│        Consumer          │
│  ConsumerDispatch = 1    │
└───────────┬──────────────┘
            │
            │  Execução sequencial
            ▼
┌───────────┐
│  Task 1   │
└────┬──────┘
     │
     ▼
  DB Write
     │
     ▼
   ACK
     │
     ▼
  Próxima mensagem

PrefetchCount = 10 + ConsumerDispatchConcurrency = 2

┌──────────────────────────┐
│      RabbitMQ Queue      │
└───────────┬──────────────┘
            │
            │  Prefetch = 10
            │  (10 mensagens em voo)
            ▼
┌──────────────────────────┐
│        Consumer          │
│  ConsumerDispatch = 2    │
└───────────┬──────────────┘
            │
            │  Dispatch paralelo (máx. 2)
            ▼
┌───────────┬───────────┐
│  Task 1   │  Task 2   │
└────┬──────┴────┬──────┘
     │             │
     ▼             ▼
  DB Write       DB Write
     │             │
     ▼             ▼
   ACK           ACK
     │             │
     ▼             ▼
 Próxima msg   Próxima msg

Valores recomendados (práticos)

Cenário	ConsumerDispatchConcurrency
Banco sensível	`1`
Escrita moderada	`2`
Leitura / cache	`4`
CPU bound	`Environment.ProcessorCount`

Sempre validar com teste de carga real.

Exemplo atualizado (API moderna)

var factory = new ConnectionFactory
{
    HostName = "localhost",
    ConsumerDispatchConcurrency = 1 // Protege o banco
};

using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();

channel.BasicQos(0, 1, false);

var consumer = new AsyncEventingBasicConsumer(channel);

consumer.ReceivedAsync += async (_, ea) =>
{
    await ProcessarMensagemAsync(ea);
    await channel.BasicAckAsync(ea.DeliveryTag, false);
};

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

Regra de ouro (atualizada)

❌ DispatchConsumersAsync → legado / removido ✅ ConsumerDispatchConcurrency → controla paralelismo real ✅ Prefetch + concorrência devem ser pensados juntos

1. Visão Geral​

2. Conceito Fundamental: Dispatch Concurrency​

3. Nível 1 — Prefetch (BasicQos)​

O que o Prefetch controla​

O que o Prefetch não controla​

Interpretação correta​

4. Nível 2 — Modelo de Consumer em C#​

AsyncEventingBasicConsumer​

Comportamento real​

5. Nível 3 — Código do Consumer (C#)​

6. Backpressure: o verdadeiro papel do RabbitMQ​

7. Padrões de Consumo Recomendados​

7.1 Consumo Sequencial (Banco Sensível)​

7.2 Consumo Concorrente Controlado (Recomendado)​

7.3 Processamento em Batch (Alta Eficiência)​

8. Estratégias de ACK / NACK​

ACK​

NACK sem requeue​

NACK com requeue​

9. Dead Letter Queue (DLQ)​

10. Conexões, Channels e Thread Safety​

Conexão (IConnection)​

Channel (IChannel)​

11. Ciclo de Vida e Dispose​

Regras​

12. Anti-padrões Comuns​

13. Checklist de Produção​

14. Regra de Ouro​

15. DispatchConsumersAsync vs ConsumerDispatchConcurrency (Atualização Importante)​

O que mudou na biblioteca​

O novo controle: ConsumerDispatchConcurrency​

Relação entre Prefetch e Dispatch Concurrency​

Valores recomendados (práticos)​

Exemplo atualizado (API moderna)​

Regra de ouro (atualizada)​