First Producer and Consumer with RabbitMQ.Client 7: Async API, Connection/Channel Lifecycle and Manual Ack

Version note: The code in this post was built and run with .NET 10, RabbitMQ 4.4 and RabbitMQ.Client 7.2.2.

Guess First

The Billing consumer listens on the billing.order-placed queue, and every invoice takes 1 second. There are 500 messages waiting in the queue. You start the consumer and, 2.5 seconds later, ask RabbitMQ: "how many messages are ready in the queue, and how many are waiting for an ack at the consumer?" There is no BasicQosAsync call in the code. What is the answer? And if the same consumer runs with autoAck: true and the process crashes on the third message, how many messages are left in a 20-message queue?

Both answers are in the middle of this post, with real rabbitmqctl output. Spoiler: both are worse than most people guess.

Connection and Channel: Who Lives How Long?

RabbitMQ.Client 7, released in November 2024, is the version that removed the synchronous API (Application Programming Interface) entirely. Everything is async now: CreateConnectionAsync, CreateChannelAsync, BasicPublishAsync. Migrating from 6.x is covered in the last post of the series; here we write correct code from scratch. There are two core objects. IConnection represents the TCP (Transmission Control Protocol) connection to the broker; opening one means a TCP handshake, an AMQP (Advanced Message Queuing Protocol) handshake and authentication. IChannel is a lightweight logical session inside that connection; every operation such as publish, consume and ack goes through a channel.

The rule is simple: one long-lived connection for the lifetime of the application, and a separate, unshared channel per unit of work. RabbitMQ's .NET client guide says an IChannel should not be used by more than one thread at a time, and in particular must not be shared between publishing threads; concurrent publishing on the same channel interleaves protocol frames. That is why the Order API's Program.cs looks like this:

using System.Text.Json;
using Bookstore.Contracts;
using Bookstore.OrderApi;
using RabbitMQ.Client;

WebApplicationBuilder builder = WebApplication.CreateBuilder(args);

ConnectionFactory factory = new()
{
    Uri = new Uri(builder.Configuration.GetConnectionString("RabbitMq")!),
    ClientProvidedName = "bookstore-order-api"
};

// One connection for the whole application lifetime: opened once at startup
IConnection connection = await factory.CreateConnectionAsync();
await RabbitTopologySetup.DeclareAsync(connection);

builder.Services.AddSingleton(connection);
builder.Services.AddSingleton<OrderEventPublisher>();

WebApplication app = builder.Build();

app.MapPost("/orders", async (PlaceOrderRequest request, OrderEventPublisher publisher, CancellationToken ct) =>
{
    OrderPlaced message = new(
        OrderId: Guid.CreateVersion7(),
        CustomerId: request.CustomerId,
        Total: request.Items.Sum(i => i.Quantity * i.UnitPrice),
        Currency: "TRY",
        PlacedAt: DateTimeOffset.UtcNow,
        Items: request.Items);

    await publisher.PublishAsync(message, ct);
    return Results.Accepted($"/orders/{message.OrderId}", new { message.OrderId });
});

app.Lifetime.ApplicationStopping.Register(() => app.Logger.LogInformation("Stopping, connection will be closed"));

await app.RunAsync();
await connection.CloseAsync();

public sealed record PlaceOrderRequest(Guid CustomerId, IReadOnlyList<OrderPlacedItem> Items);

We hand the connection to the DI (Dependency Injection) container as a ready instance created with top-level await, not through a factory lambda. Writing CreateConnectionAsync().GetAwaiter().GetResult() inside a lambda means both blocking on async code and the surprise of "the connection opens on the first request". Here, if the broker is unreachable the application does not start at all, which is what we want, because RabbitMQ's automatic recovery does not cover the initial connection attempt. Lifetimes are covered in detail in the ASP.NET Core Web API series post "Dependency Injection: Lifetimes and the Captive Dependency Trap". ClientProvidedName is a small but valuable setting: it shows which application opened the connection in the management UI (user interface) and in rabbitmqctl list_connections.

How expensive is "a new connection per message"? A small experiment sending 200 messages to the same broker on the same machine:

New connection per message: 200 messages in 393 ms
One long-lived connection: 200 messages in 1 ms
Long-lived + confirms, awaited one by one: 200 messages in 38 ms
Long-lived + confirms, batch awaited: 200 messages in 3 ms

Even with the broker on localhost, opening a connection per message is roughly 400 times slower. Over a real network with TLS (Transport Layer Security) the gap grows, and on the broker side every connection is a separate Erlang process and its own memory. The last two lines show the cost of publisher confirmations; we will get to that shortly.

Declaring the Topology at Startup

Instead of creating exchanges and queues by hand in the management UI, we declare them when the application starts:

public static class RabbitTopologySetup
{
    public static async Task DeclareAsync(IConnection connection)
    {
        await using IChannel channel = await connection.CreateChannelAsync();

        await channel.ExchangeDeclareAsync(RabbitTopology.OrderingExchange, ExchangeType.Topic, durable: true);
        await channel.QueueDeclareAsync(RabbitTopology.BillingQueue, durable: true, exclusive: false, autoDelete: false,
            arguments: new Dictionary<string, object?> { ["x-queue-type"] = "quorum" });
        await channel.QueueBindAsync(RabbitTopology.BillingQueue, RabbitTopology.OrderingExchange,
            RabbitTopology.OrderPlacedRoutingKey);
    }
}

Declarations are idempotent: calling them a second time with the same name and arguments changes nothing. bookstore.ordering is a topic exchange, billing.order-placed is a quorum queue, and the binding between them uses the order.placed routing key. Exchange types, quorum queues, and which setting belongs in code versus a broker policy are the subject of the next post.

Publishing: BasicPublishAsync and Publisher Confirmations

The publisher is a small singleton class that uses a single channel behind a SemaphoreSlim:

public sealed class OrderEventPublisher(IConnection connection, ILogger<OrderEventPublisher> logger) : IAsyncDisposable
{
    private readonly SemaphoreSlim _gate = new(1, 1);
    private IChannel? _channel;

    public async Task PublishAsync(OrderPlaced message, CancellationToken ct)
    {
        byte[] body = JsonSerializer.SerializeToUtf8Bytes(message, MessageJson.Options);
        BasicProperties props = new()
        {
            MessageId = Guid.NewGuid().ToString(),
            CorrelationId = message.OrderId.ToString(),
            Type = OrderPlaced.MessageType,
            ContentType = "application/json",
            DeliveryMode = DeliveryModes.Persistent
        };

        // IChannel is not thread-safe: one publish at a time
        await _gate.WaitAsync(ct);
        try
        {
            _channel ??= await connection.CreateChannelAsync(new CreateChannelOptions(
                publisherConfirmationsEnabled: true,
                publisherConfirmationTrackingEnabled: true), ct);

            // with confirm tracking on, this await lasts until the broker says "ack"
            await _channel.BasicPublishAsync(RabbitTopology.OrderingExchange, RabbitTopology.OrderPlacedRoutingKey,
                mandatory: true, basicProperties: props, body: body, cancellationToken: ct);
        }
        finally
        {
            _gate.Release();
        }

        logger.LogInformation("Published {MessageType} {MessageId} for order {OrderId}",
            props.Type, props.MessageId, message.OrderId);
    }

    public async ValueTask DisposeAsync()
    {
        if (_channel is not null)
        {
            await _channel.CloseAsync();
            await _channel.DisposeAsync();
        }
    }
}

There are three decisions here. The first is CreateChannelOptions: in 7.x, publisher confirmations are enabled when the channel is created. publisherConfirmationsEnabled asks the broker to confirm every message, and publisherConfirmationTrackingEnabled makes the client track that confirmation and tie it to the ValueTask returned by BasicPublishAsync. With both on, the await does not complete until the broker confirms it has the message; if the broker rejects it, an exception is thrown. Without confirmations, BasicPublishAsync returning successfully only means "the bytes were written to the socket".

The second is mandatory: true. If the exchange cannot route the message to any queue, for example because you used a routing key with no binding, the broker silently drops it by default. With mandatory and confirm tracking both on, the same failure comes back to the publisher as an exception. Real output when we tried the order.refunded routing key, which has no binding:

PublishReturnException: IsReturn=True, SeqNo=1, 312 NO_ROUTE Exchange: bookstore.ordering Routing Key: order.refunded
Without confirms and mandatory: publish 'succeeded', message silently dropped

The third is the message properties. MessageId is assigned once per message and does not change on redelivery; in the upcoming idempotent consumer post we will use it for deduplication. CorrelationId carries the order, Type carries the contract name and version. DeliveryMode = Persistent makes the broker write the message to disk; quorum queues always write messages to disk anyway, but for a message that lands in a classic queue this setting matters.

Back to the cost of confirmations. In the measurement above, awaiting 200 messages one by one took 38 ms, while collecting the same 200 publish ValueTasks in a list and awaiting them afterwards took 3 ms. Every await waits for a broker round trip. In the Order API each HTTP (Hypertext Transfer Protocol) request produces a single message, so awaiting one by one is right: we make sure the message is at the broker before returning 202 to the client. For scenarios like bulk imports, await in batches.

The POST /orders request and the API log from a real run:

HTTP/1.1 202 Accepted
Location: /orders/01a1b8de-6a5a-70fd-8aab-7f9dd66cdbe7

{"orderId":"01a1b8de-6a5a-70fd-8aab-7f9dd66cdbe7"}

info: Bookstore.OrderApi.OrderEventPublisher[0]
      Published bookstore.order-placed.v1 dfa727ef-8eae-473e-a4c7-5f89a484da6d for order 01a1b8de-6a5a-70fd-8aab-7f9dd66cdbe7

Running the Consumer Inside a BackgroundService

The Billing side is a Worker Service. The consumer opens its own channel inside the BackgroundService's ExecuteAsync:

public sealed class OrderPlacedConsumer(
    IConnection connection, InvoiceService invoices, ILogger<OrderPlacedConsumer> logger) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await using IChannel channel = await connection.CreateChannelAsync(cancellationToken: stoppingToken);

        // at most 10 unacked messages are delivered to this consumer
        await channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 10, global: false, stoppingToken);

        AsyncEventingBasicConsumer consumer = new(channel);
        consumer.ReceivedAsync += async (_, ea) =>
        {
            try
            {
                OrderPlaced order = JsonSerializer.Deserialize<OrderPlaced>(ea.Body.Span, MessageJson.Options)
                    ?? throw new JsonException("Empty body");

                await invoices.CreateInvoiceAsync(order, stoppingToken);
                await channel.BasicAckAsync(ea.DeliveryTag, multiple: false);
            }
            catch (JsonException ex)
            {
                // a malformed message will not be fixed by retrying: do not requeue
                logger.LogError(ex, "Malformed message {MessageId}, rejecting", ea.BasicProperties.MessageId);
                await channel.BasicNackAsync(ea.DeliveryTag, multiple: false, requeue: false);
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                // shutting down: no ack, the broker hands the message to another consumer
                logger.LogWarning("Shutdown during {MessageId}, leaving it unacked", ea.BasicProperties.MessageId);
            }
        };

        string consumerTag = await channel.BasicConsumeAsync(RabbitTopology.BillingQueue, autoAck: false,
            consumer: consumer, cancellationToken: stoppingToken);
        logger.LogInformation("Consuming {Queue} with tag {ConsumerTag}", RabbitTopology.BillingQueue, consumerTag);

        try
        {
            await Task.Delay(Timeout.Infinite, stoppingToken);
        }
        catch (OperationCanceledException)
        {
            // normal shutdown
        }

        await channel.BasicCancelAsync(consumerTag, noWait: false, CancellationToken.None);
        await channel.CloseAsync(CancellationToken.None);
        logger.LogInformation("Consumer {ConsumerTag} cancelled", consumerTag);
    }
}

In 7.x you can think of AsyncEventingBasicConsumer as the consumer type; the event is called ReceivedAsync and the handler returns a Task. The message body arrives as ReadOnlyMemory<byte>, and the client may reuse that memory once the handler finishes. That is why we deserialize inside the handler; stashing ea.Body in a field and reading it later means reading garbage.

By default, messages on a channel are dispatched one at a time, in order: ConsumerDispatchConcurrency is 1. If the handler takes 200 ms, this consumer processes at most 5 messages per second. If you want parallelism, either raise ConsumerDispatchConcurrency or attach several consumers (several instances or channels) to the same queue. The second option is both simpler and makes ordering expectations more visible.

Why ExecuteAsync has the shape "start consuming and wait forever" also matters. Consuming is not a loop but a subscription the broker pushes to; BasicConsumeAsync returns immediately. If the method ends, await using closes the channel and the subscription dies. In .NET 10 the whole of ExecuteAsync runs on a background thread, so the part before the first await no longer blocks host startup; that is why the Consuming billing.order-placed line appears after Application started in the logs. We will cover the BackgroundService lifecycle in detail in the upcoming Background Jobs series.

Manual Ack, Nack and Prefetch

Back to the first question. Without prefetch, the broker sends every message in the queue to the consumer; all of them sit in the client's memory waiting for an ack. 500 messages of 2 KB each, a 1-second handler, 2.5 seconds after the consumer started:

prefetch=none: consumer started, check the queue now
name	messages_ready	messages_unacknowledged	consumers
lab.prefetch	0	498	1
prefetch=10: consumer started, check the queue now
name	messages_ready	messages_unacknowledged	consumers
lab.prefetch	488	10	1

The consumer without prefetch has finished two messages and pulled all of the remaining 498 into its own memory. The queue shows zero "ready" messages. If you started a second Billing instance now, it would not get a single message; they are all locked in the first instance. Had the messages been 200 KB instead of 2 KB, the process's memory would have grown accordingly. The same experiment with BasicQosAsync(0, 10, false): 10 messages at the consumer, 488 waiting in the queue, and a new instance can take its share immediately.

The right prefetch value depends on handler duration. For fast handlers bound by the CPU (Central Processing Unit), a value between 50 and 300 hides network latency; for slow handlers that call out, like Billing, 5–20 is a reasonable start. There is no fixed right answer; measure and tune.

The second question: autoAck: true. In this mode the broker considers a message delivered the moment it writes it to the consumer and removes it from the queue. Whether the consumer finished the work does not matter to it. Prefetch does not apply either, because no message is awaiting an ack. A consumer reading a 20-message queue with autoAck: true and crashing on the third message:

20 messages published to lab.autoack
Processing invoice-1
Processing invoice-2
Processing invoice-3
Process crashed
exit=1
lab.autoack	0	0

Three messages started processing, yet zero messages are left in the queue. The broker pushed all 20 to the client at once; when the process died, 17 of them vanished without ever being processed. Seventeen orders without an invoice, and no error record anywhere. With autoAck: false, every unacked message goes back to the queue after the same crash. This is the code-level counterpart of the difference between at-most-once and at-least-once described in the "Delivery Guarantees" post of the Messaging and Event-Driven Architecture series.

Besides ack there are two negative answers. BasicNackAsync(tag, multiple: false, requeue: true) puts the message back in the queue; requeue: false drops it, or sends it to a DLX (Dead Letter Exchange) if the queue has dead-lettering configured. Malformed JSON (JavaScript Object Notation) will not be fixed by retrying, so the code above uses requeue: false. Retry and dead-letter strategies are the subject of two later posts in the series.

When the Connection Drops: Automatic Recovery

ConnectionFactory defaults in 7.x, from a real run:

AutomaticRecoveryEnabled = True
TopologyRecoveryEnabled  = True
NetworkRecoveryInterval  = 00:00:05
RequestedHeartbeat       = 00:01:00
ConsumerDispatchConcurrency = 1

With automatic recovery on, when the connection drops the client tries to reconnect every 5 seconds; once it succeeds, it restores channels, QoS (Quality of Service) settings, the exchanges, queues and bindings the client declared, and consumers. With Billing and the Order API running, we stopped and restarted the RabbitMQ container. The Billing log:

Connection shutdown: 320 CONNECTION_FORCED - broker forced connection closure with reason 'shutdown'
Connection recovered
info: Bookstore.Billing.InvoiceService[0]
      Invoice created for order 01a1b8de-bc82-73c1-b8cd-ffa17698f366: 450.00 TRY

The consumer came back without any extra code and processed the next order. The publisher side is less lucky. A request that arrived while the broker was down:

fail: Microsoft.AspNetCore.Diagnostics.DeveloperExceptionPageMiddleware[1]
      An unhandled exception has occurred while executing the request.
      RabbitMQ.Client.Exceptions.AlreadyClosedException: Already closed: The AMQP operation was interrupted: AMQP close-reason, initiated by Peer, code=320, text='CONNECTION_FORCED - broker forced connection closure with reason 'shutdown'', classId=0, methodId=0

The client does not buffer messages published during recovery; it throws. That is honest behaviour: the API returned 500 and did not claim the order was "received". But if the order is written to the database and the publish then fails, you are left with an inconsistent state. The real fix is the transactional outbox, and we will build the .NET side in the upcoming outbox post of this series. Another subtle point after recovery: the client does not try to ack delivery tags from the old channel, so messages that were being processed at the moment of recovery are delivered again. Accept from the start that a consumer can see the same message twice.

Graceful Shutdown

During a deploy, Billing receives SIGTERM (the Unix termination signal that asks a process to shut down). The host cancels stoppingToken, Task.Delay ends, BasicCancelAsync tells the broker "stop sending me new messages", and the channel closes. We sent five orders and sent SIGTERM to the process 300 ms later:

info: Microsoft.Hosting.Lifetime[0]
      Application is shutting down...
warn: Bookstore.Billing.OrderPlacedConsumer[0]
      Shutdown during 0a024092-1151-492b-86eb-fc979a82305a, leaving it unacked
warn: Bookstore.Billing.OrderPlacedConsumer[0]
      Shutdown during d44eaba8-d344-4ce3-b79b-320cdb159a89, leaving it unacked
warn: Bookstore.Billing.OrderPlacedConsumer[0]
      Shutdown during b7cfdfb2-014a-4156-91b8-df0f5dd4300e, leaving it unacked
warn: Bookstore.Billing.OrderPlacedConsumer[0]
      Shutdown during 0a2950b8-e18f-4291-aa89-2de6a3d2e610, leaving it unacked
info: Bookstore.Billing.OrderPlacedConsumer[0]
      Consumer amq.ctag-I7Fm7V6xSAGRCy5Tokf_BQ cancelled
Connection shutdown: 200 Goodbye

name	messages_ready	messages_unacknowledged	messages
billing.order-placed	4	0	4

One invoice had finished before shutdown; the other four messages already delivered to the client through prefetch were interrupted by the cancelled token, were not acked, and went back to the queue when the channel closed. Nothing was lost. We made a deliberate choice here: the handler gets stoppingToken, so half-done work is cancelled on shutdown. The alternative is to pass CancellationToken.None and wait for in-flight work to finish. Then it must finish within the host's shutdown timeout (30 seconds by default); with a large prefetch that may not be enough. Carrying cancellation all the way to the database is covered in the ASP.NET Core Web API series post "Async/Await and CancellationToken".

Production Notes

Watch the connection count. In the management UI the number of connections should be proportional to the number of instances and stable; if it grows with request volume, a connection is leaking somewhere. ClientProvidedName turns that hunt into minutes.

Watch channels too. In RabbitMQ 4.x the default upper limit of channels per connection (channel_max) is 2047. Code that opens a channel per request and forgets to close it hits this limit, and no new channel can be opened.

Watch the queue's two numbers separately: messages_ready shows the backlog waiting to be consumed, messages_unacknowledged shows work stuck at consumers. If unacked sits constantly at prefetch × consumer count, the consumers cannot keep up. We will collect these metrics with OpenTelemetry in the observability post of the series.

Put a timeout on the publish path. An await waiting for a confirm may not return for a long time while the broker is in a memory or disk alarm (which blocks publishers). Pass the HTTP request's CancellationToken, or your own timeout, to BasicPublishAsync.

A single-channel publisher's throughput is bounded by one broker round trip. If the Order API does not receive hundreds of orders per second, that is enough; if it does, consider a small pool of a few channels or batched confirms.

Common Pitfalls

A new connection per message. The 400x gap we measured was on localhost; in a real environment it also wears down the broker. Connection as a singleton, channel per unit of work.

Sharing a channel across threads. Using a single IChannel without a lock inside a singleton publisher leads to occasional protocol errors under load; they rarely show up in a test environment.

Saying "nothing will happen" with autoAck: true. As we saw, when the process crashes, every message pushed to the client without a prefetch limit is lost.

A consumer without prefetch. The whole queue flows into a single instance's memory, and horizontal scaling stops working.

Trusting publish without confirmations. BasicPublishAsync not throwing does not mean the message reached a queue; a message sent with the wrong routing key silently disappears.

Carrying ea.Body outside the handler. The memory belongs to the client; copy it or deserialize inside the handler.

A Small Experiment

★☆☆ Run Billing as two instances and send 20 orders. With prefetch 10, how are the messages distributed between the two instances? Lower prefetch to 1 and try again.

★★☆ Set ConsumerDispatchConcurrency to 4 and log which thread the handler runs on. Does the processing order of the 20 orders change? Write one paragraph on the risk of this setting in a scenario that needs ordered processing.

★★★ Rewrite OrderEventPublisher with a pool of 4 channels instead of a single channel. Send 1,000 concurrent requests with a load testing tool and compare p95 latency with the single-channel version.

So far we have worked with a single topic exchange, a single binding and a single quorum queue. In the next post we take topology in code seriously: direct, topic, fanout and headers exchanges, routing key design, the difference between classic, quorum and stream queues, and which settings belong in a broker policy rather than in the client.