Service Bus Queues and Topics

Queues and topics

Queue   one message, one consumer. Competing consumers share the load.
Topic   one message, many subscriptions. Each subscriber gets a copy.

az servicebus namespace create \
  --name sb-shop-prod --resource-group rg-shop-prod-uks \
  --location uksouth --sku Standard

az servicebus queue create \
  --name orders --namespace-name sb-shop-prod --resource-group rg-shop-prod-uks \
  --max-delivery-count 5 \
  --lock-duration PT5M \
  --default-message-time-to-live P14D \
  --enable-dead-lettering-on-message-expiration true

az servicebus topic create \
  --name order-events --namespace-name sb-shop-prod --resource-group rg-shop-prod-uks

az servicebus topic subscription create \
  --name fulfilment --topic-name order-events \
  --namespace-name sb-shop-prod --resource-group rg-shop-prod-uks

Basic     queues only, no topics. Rarely the right choice.
Standard  topics, sessions, shared throughput. Most workloads.
Premium   dedicated resources, predictable latency, vnet, larger messages, geo DR

Sending and receiving

builder.Services.AddSingleton(sp => new ServiceBusClient(
    "sb-shop-prod.servicebus.windows.net",
    new DefaultAzureCredential()));

var sender = client.CreateSender("orders");
await sender.SendMessageAsync(new ServiceBusMessage(JsonSerializer.Serialize(order))
{
    MessageId = order.Id,               // enables duplicate detection
    ContentType = "application/json",
    Subject = "OrderPlaced",
    CorrelationId = correlationId,
    ApplicationProperties = { ["tenantId"] = order.TenantId }
});

var processor = client.CreateProcessor("orders", new ServiceBusProcessorOptions
{
    MaxConcurrentCalls = 10,
    AutoCompleteMessages = false,
    PrefetchCount = 20
});

processor.ProcessMessageAsync += async args =>
{
    try
    {
        var order = args.Message.Body.ToObjectFromJson<Order>();
        await _handler.HandleAsync(order, args.CancellationToken);
        await args.CompleteMessageAsync(args.Message);
    }
    catch (ValidationException ex)
    {
        // never going to succeed, so do not retry it
        await args.DeadLetterMessageAsync(args.Message, "ValidationFailed", ex.Message);
    }
    catch (Exception)
    {
        // transient: abandon so it is retried
        await args.AbandonMessageAsync(args.Message);
    }
};

processor.ProcessErrorAsync += args =>
{
    _logger.LogError(args.Exception, "Service Bus error in {Source}", args.ErrorSource);
    return Task.CompletedTask;
};

await processor.StartProcessingAsync();

AutoCompleteMessages = false is important. With it true, a message is completed even when your handler throws, which silently loses work.

Peek lock

The default receive mode, and understanding it prevents most problems.

1. The message is locked for the lock duration, invisible to others
2. Your handler runs
3. Complete    the message is removed
   Abandon     the lock is released and it is redelivered immediately
   Dead letter it moves aside for investigation
   Defer       it stays, retrievable only by sequence number
4. If the lock expires first, it is redelivered automatically

Set the lock duration longer than the handler takes. A five minute lock and a six minute handler means the message is redelivered while still being processed, and completing it then fails with a lock lost error.

For genuinely long work, renew the lock:

processor = client.CreateProcessor("orders", new ServiceBusProcessorOptions
{
    MaxAutoLockRenewalDuration = TimeSpan.FromMinutes(30)
});

Dead letter queue

Every queue and subscription has one, automatically.

Messages arrive there when
  delivery count exceeds max-delivery-count
  the handler dead letters explicitly
  the message expires, if enabled
  a subscription filter evaluation fails

var dlqReceiver = client.CreateReceiver("orders",
    new ServiceBusReceiverOptions { SubQueue = SubQueue.DeadLetter });

var messages = await dlqReceiver.ReceiveMessagesAsync(maxMessages: 50);
foreach (var message in messages)
{
    _logger.LogWarning("Dead lettered: {Reason} {Description}",
        message.DeadLetterReason, message.DeadLetterErrorDescription);
}

Alert on dead letter queue depth. A dead letter queue nobody watches is lost work that nobody knows about, and this is the most common operational gap with Service Bus.

az monitor metrics alert create \
  --name dlq-not-empty --resource-group rg-shop-prod-uks \
  --scopes /subscriptions/.../namespaces/sb-shop-prod \
  --condition "total DeadletteredMessages > 0" \
  --window-size 15m --evaluation-frequency 5m \
  --action /subscriptions/.../actionGroups/ag-oncall

Topic filters

# only high value orders
az servicebus topic subscription rule create \
  --name high-value --subscription-name fraud-check \
  --topic-name order-events --namespace-name sb-shop-prod \
  --resource-group rg-shop-prod-uks \
  --filter-sql-expression "total > 1000"

# by a custom property
az servicebus topic subscription rule create \
  --name uk-only --subscription-name uk-fulfilment \
  --topic-name order-events --namespace-name sb-shop-prod \
  --resource-group rg-shop-prod-uks \
  --filter-sql-expression "country = 'GB' AND messageType = 'OrderPlaced'"

Filters operate on application properties and system properties, not the body. Anything you want to filter on must be set as a property when sending.

Every new subscription starts with a default rule matching everything. Delete it when adding your own, or the subscription receives every message regardless of your filter.

Correlation filters are cheaper than SQL filters and cover most cases:

az servicebus topic subscription rule create \
  --name order-placed --subscription-name fulfilment \
  --topic-name order-events --namespace-name sb-shop-prod \
  --resource-group rg-shop-prod-uks \
  --correlation-id OrderPlaced

Sessions for ordering

az servicebus queue create \
  --name orders --namespace-name sb-shop-prod --resource-group rg-shop-prod-uks \
  --enable-session true

await sender.SendMessageAsync(new ServiceBusMessage(body)
{
    SessionId = order.CustomerId    // all messages for one customer are ordered
});

var sessionProcessor = client.CreateSessionProcessor("orders",
    new ServiceBusSessionProcessorOptions
    {
        MaxConcurrentSessions = 10,
        MaxConcurrentCallsPerSession = 1
    });

Sessions give ordered, exclusive processing per session ID. One consumer holds a session at a time, so messages within it are handled in order.

The cost is parallelism: a session is processed by one consumer, so a single hot session becomes a bottleneck. Choose the session ID with the same care as a partition key.

Duplicate detection

az servicebus queue create \
  --name orders --namespace-name sb-shop-prod --resource-group rg-shop-prod-uks \
  --enable-duplicate-detection true \
  --duplicate-detection-history-time-window PT10M

Messages with the same MessageId within the window are discarded. It protects against a sender retrying after a timeout when the first send actually succeeded.

It is not a substitute for idempotent handlers, since the window is finite and delivery is still at least once.

Scheduled and deferred messages

// send later
var sequenceNumber = await sender.ScheduleMessageAsync(
    new ServiceBusMessage(body),
    DateTimeOffset.UtcNow.AddHours(24));

await sender.CancelScheduledMessageAsync(sequenceNumber);

// defer: keep it, but come back to it
await args.DeferMessageAsync(args.Message);
var deferred = await receiver.ReceiveDeferredMessageAsync(sequenceNumber);

Scheduled messages replace a timer table for delayed work like reminders. Deferral is for out of order arrivals, where you must wait for a prerequisite before processing.

Store the sequence number for a deferred message, since it is the only way to retrieve it.

Service Bus or Event Grid or Event Hubs

Service Bus  commands and work items. Ordering, transactions, dead letters.
Event Grid   reactive notifications. Something happened, tell whoever cares.
Event Hubs   high volume telemetry streams. Millions per second, replayable.

Service Bus is for work that must be done, with delivery guarantees and a dead letter path. It is the right default for business processes.

Quick recap

• A queue delivers to one consumer, a topic delivers a copy to each subscription.
• Set AutoCompleteMessages = false, or a failing handler still completes the message.
• Make the lock duration longer than the handler takes, or renew it explicitly.
• Alert on dead letter depth, since an unwatched dead letter queue is invisible lost work.
• Topic filters see properties, not the body, and the default catch all rule must be deleted.
• Sessions give ordering per session ID at the cost of parallelism.
• Duplicate detection has a finite window and does not replace idempotent handlers.