Azure Service Bus (2) — producers and consumers in .NET

The second part of the series about Azure Service Bus. In the first part we chose between Queue and Topic. Now we write producers and consumers in .NET.

Setup: package and authentication

dotnet add package Azure.Messaging.ServiceBus
dotnet add package Azure.Identity

Consistent with the rest of the series: no connection string. ServiceBusClient authenticates via Managed Identity:

# Role for sending
az role assignment create \
  --assignee $PRINCIPAL_ID \
  --role "Azure Service Bus Data Sender" \
  --scope $SERVICEBUS_ID

# Role for consuming
az role assignment create \
  --assignee $PRINCIPAL_ID \
  --role "Azure Service Bus Data Receiver" \
  --scope $SERVICEBUS_ID

// Program.cs -- a single ServiceBusClient, singleton
builder.Services.AddSingleton(_ => new ServiceBusClient(
    "my-servicebus.servicebus.windows.net",  // fully qualified namespace
    new DefaultAzureCredential()));

ServiceBusClient is thread-safe and internally managed with connection pooling — you register it only once as a singleton and create senders/processors from it.

The Producer

Simple publishing

public class OrderEventPublisher
{
    private readonly ServiceBusSender _sender;

    public OrderEventPublisher(ServiceBusClient client)
    {
        // Sender per queue/topic -- it is also thread-safe, keep it
        _sender = client.CreateSender("order-events");
    }

    public async Task PublishOrderPlacedAsync(OrderPlacedEvent evt)
    {
        var message = new ServiceBusMessage(JsonConvert.SerializeObject(evt))
        {
            ContentType = "application/json",
            MessageId = evt.OrderId.ToString(),   // for deduplication
            Subject = "OrderPlaced",              // event type
            CorrelationId = evt.CorrelationId     // for end-to-end tracing
        };

        // Custom properties -- subscription filters apply on these
        message.ApplicationProperties["TotalAmount"] = evt.TotalAmount;
        message.ApplicationProperties["Region"] = evt.Region;

        await _sender.SendMessageAsync(message);
    }
}

Details that matter:

• MessageId — set deterministically (e.g. OrderId), allows automatic deduplication if you enable duplicate detection on the queue/topic.
• Subject — event type; consumers can run different handlers based on it.
• ApplicationProperties — subscription SQL filters from part 1 apply on these (TotalAmount > 1000).

Batching for volume

Sending message-by-message at high volumes is inefficient. ServiceBusMessageBatch packs multiple messages into a single operation, respecting the size limit:

public async Task PublishBatchAsync(IEnumerable<OrderPlacedEvent> events)
{
    using ServiceBusMessageBatch batch = await _sender.CreateMessageBatchAsync();

    foreach (var evt in events)
    {
        var message = new ServiceBusMessage(JsonConvert.SerializeObject(evt));

        if (!batch.TryAddMessage(message))
        {
            // Batch is full -- send and start a new one
            await _sender.SendMessagesAsync(batch);
            batch.Dispose();
            batch = await _sender.CreateMessageBatchAsync();

            if (!batch.TryAddMessage(message))
                throw new InvalidOperationException("Message too large for an empty batch");
        }
    }

    if (batch.Count > 0)
        await _sender.SendMessagesAsync(batch);
}

Scheduled messages

// Deliver the message in 30 minutes (e.g. reminder, delayed retry)
message.ScheduledEnqueueTime = DateTimeOffset.UtcNow.AddMinutes(30);
await _sender.SendMessageAsync(message);

The Consumer: ServiceBusProcessor

For consumption, use ServiceBusProcessor — it automatically manages lock renewal, concurrency, and reconnection. You naturally integrate it into a BackgroundService:

public class OrderProcessorService : BackgroundService
{
    private readonly ServiceBusProcessor _processor;
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly ILogger<OrderProcessorService> _logger;

    public OrderProcessorService(
        ServiceBusClient client,
        IServiceScopeFactory scopeFactory,
        ILogger<OrderProcessorService> logger)
    {
        _scopeFactory = scopeFactory;
        _logger = logger;

        _processor = client.CreateProcessor(
            topicName: "order-events",
            subscriptionName: "inventory",
            new ServiceBusProcessorOptions
            {
                MaxConcurrentCalls = 5,       // 5 messages processed in parallel
                PrefetchCount = 20,            // prefetch 20 for throughput
                AutoCompleteMessages = false,  // complete explicitly after success
                MaxAutoLockRenewalDuration = TimeSpan.FromMinutes(5)
            });
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _processor.ProcessMessageAsync += OnMessageAsync;
        _processor.ProcessErrorAsync += OnErrorAsync;

        await _processor.StartProcessingAsync(stoppingToken);

        // Graceful shutdown -- clean stop on SIGTERM
        stoppingToken.Register(() =>
            _processor.StopProcessingAsync().GetAwaiter().GetResult());
    }

    private async Task OnMessageAsync(ProcessMessageEventArgs args)
    {
        // New scope per message -- scoped repositories work correctly
        using var scope = _scopeFactory.CreateScope();
        var handler = scope.ServiceProvider.GetRequiredService<IOrderHandler>();

        var evt = JsonConvert.DeserializeObject<OrderPlacedEvent>(
            args.Message.Body.ToString())!;

        await handler.HandleAsync(evt, args.CancellationToken);

        // Complete ONLY after success -- otherwise the message returns to the queue
        await args.CompleteMessageAsync(args.Message);
    }

    private Task OnErrorAsync(ProcessErrorEventArgs args)
    {
        _logger.LogError(args.Exception,
            "Service Bus error: {ErrorSource}, entity {EntityPath}",
            args.ErrorSource, args.EntityPath);
        return Task.CompletedTask;
    }
}

Settings that matter

• AutoCompleteMessages = false — the most important. You explicitly complete the message after successful processing. If the handler throws an exception, the message returns to the queue and is redelivered — the basis for natural retry.
• MaxConcurrentCalls — parallelism per instance. 5 means 5 messages simultaneously; correlate with container resources and downstream capacity (don't bombard the database).
• PrefetchCount — how many messages the client preloads. Increases throughput, but prefetched messages have the lock active — too large a prefetch + slow processing = expired locks. Practical rule: 2-4x MaxConcurrentCalls.
• MaxAutoLockRenewalDuration — the processor automatically renews the lock for long processing. Set it above the maximum actual processing duration.

Connection to the series: this BackgroundService respects CancellationToken (graceful shutdown from the first article of the series), and in Container Apps, KEDA scales exactly this consumer based on queue length (part 2 of the ACA mini-series).

What’s next

What happens when processing repeatedly fails? In the third part: delivery count, dead-letter queue, retry policies with exponential backoff, and how to monitor and reprocess dead messages.

Questions? Write me at contact@ludoprogramming.com.