Session 20: Azure Service Bus Deep Dive

Mục tiêu: Master Azure Service Bus architecture, advanced features, và tích hợp với .NET/MassTransit.

1. Architecture

┌────────────────────────────────────────────────┐
│              Azure Service Bus                  │
│                                                 │
│  Queues (Point-to-Point):                       │
│  ┌──────┐    ┌────────────┐    ┌──────────┐   │
│  │Sender│───→│   Queue    │───→│ Receiver │   │
│  └──────┘    └────────────┘    └──────────┘   │
│                                                 │
│  Topics/Subscriptions (Pub/Sub):                │
│  ┌──────┐    ┌────────────┐    ┌──────────┐   │
│  │Sender│───→│   Topic    │───→│ Sub A    │   │
│  └──────┘    │            │    ├──────────┤   │
│              │            │───→│ Sub B    │   │
│              │            │    ├──────────┤   │
│              │            │───→│ Sub C    │   │
│              └────────────┘    └──────────┘   │
└────────────────────────────────────────────────┘

Tiers

2. Advanced Features

Sessions - Ordered Processing

csharp
// Sessions: FIFO per session, parallel across sessions
// Use case: process messages per customer in order

// Sender: set SessionId
await sender.SendMessageAsync(new ServiceBusMessage
{
    Body = BinaryData.FromObjectAsJson(orderEvent),
    SessionId = order.CustomerId.ToString(), // All messages for same customer → same session
    ContentType = "application/json"
});

// Receiver: session-aware
var processor = client.CreateSessionProcessor("order-queue",
    new ServiceBusSessionProcessorOptions
    {
        MaxConcurrentSessions = 10,
        SessionIdleTimeout = TimeSpan.FromMinutes(5)
    });

processor.ProcessMessageAsync += async args =>
{
    var sessionId = args.SessionId; // = CustomerId
    var body = args.Message.Body.ToObjectFromJson<OrderEvent>();
    // Process in order per customer
    await ProcessOrderEvent(body);
    await args.CompleteMessageAsync(args.Message);
};

processor.ProcessErrorAsync += async args =>
{
    logger.LogError(args.Exception, "Session {SessionId} error", args.FullyQualifiedNamespace);
};

Dead-Letter Queue (DLQ)

csharp
// Messages go to DLQ when:
// 1. Max delivery count exceeded (default: 10)
// 2. Message TTL expired
// 3. Explicitly dead-lettered by receiver

// Explicitly dead-letter
await args.DeadLetterMessageAsync(args.Message,
    deadLetterReason: "ValidationFailed",
    deadLetterErrorDescription: "Invalid order format");

// Process DLQ messages
var dlqReceiver = client.CreateReceiver(
    "order-queue", new ServiceBusReceiverOptions
    {
        SubQueue = SubQueue.DeadLetter
    });

var dlqMessages = await dlqReceiver.ReceiveMessagesAsync(maxMessages: 50);
foreach (var msg in dlqMessages)
{
    logger.LogWarning("DLQ: {Reason} - {Description}",
        msg.DeadLetterReason, msg.DeadLetterErrorDescription);

    // Analyze, fix, resubmit or archive
    await dlqReceiver.CompleteMessageAsync(msg);
}

Scheduled Messages & Deferred Messages

csharp
// Scheduled: deliver at future time
var message = new ServiceBusMessage("Reminder: order expires soon")
{
    ScheduledEnqueueTime = DateTimeOffset.UtcNow.AddHours(24)
};
long sequenceNumber = await sender.ScheduleMessageAsync(message, message.ScheduledEnqueueTime);

// Cancel scheduled message
await sender.CancelScheduledMessageAsync(sequenceNumber);

// Deferred: park message, retrieve later by sequence number
await args.DeferMessageAsync(args.Message); // Park it

// Later: retrieve deferred message
var deferred = await receiver.ReceiveDeferredMessageAsync(sequenceNumber);
await receiver.CompleteMessageAsync(deferred);

Duplicate Detection

csharp
// Enable on queue/topic creation (cannot change later)
// Message with same MessageId within dedup window → silently ignored

var message = new ServiceBusMessage(body)
{
    MessageId = $"order-{orderId}-submitted", // Deterministic ID!
};

// Dedup window: typically 10 minutes
// → Producer can safely retry without creating duplicates

3. .NET SDK Patterns

csharp
// Registration
builder.Services.AddSingleton(_ =>
    new ServiceBusClient(connectionString));

builder.Services.AddSingleton(sp =>
    sp.GetRequiredService<ServiceBusClient>().CreateSender("order-events"));

// Publisher service
public class EventPublisher(ServiceBusSender sender, ILogger<EventPublisher> logger)
{
    public async Task PublishAsync<T>(T @event, CancellationToken ct = default)
    {
        var message = new ServiceBusMessage
        {
            Body = BinaryData.FromObjectAsJson(@event),
            ContentType = "application/json",
            MessageId = Guid.NewGuid().ToString(),
            Subject = typeof(T).Name,
            ApplicationProperties = { ["EventType"] = typeof(T).FullName }
        };

        await sender.SendMessageAsync(message, ct);
        logger.LogInformation("Published {EventType}", typeof(T).Name);
    }

    // Batch sending
    public async Task PublishBatchAsync<T>(IEnumerable<T> events, CancellationToken ct)
    {
        using var batch = await sender.CreateMessageBatchAsync(ct);

        foreach (var @event in events)
        {
            var message = new ServiceBusMessage(BinaryData.FromObjectAsJson(@event))
            {
                ContentType = "application/json",
                Subject = typeof(T).Name
            };

            if (!batch.TryAddMessage(message))
            {
                await sender.SendMessagesAsync(batch, ct);
                batch.Dispose();
                // Create new batch for remaining...
            }
        }

        if (batch.Count > 0)
            await sender.SendMessagesAsync(batch, ct);
    }
}

MassTransit + Azure Service Bus

csharp
builder.Services.AddMassTransit(x =>
{
    x.AddConsumers(typeof(Program).Assembly);

    x.UsingAzureServiceBus((context, cfg) =>
    {
        cfg.Host(new Uri("sb://myapp.servicebus.windows.net"),
            h => h.TokenCredential = new DefaultAzureCredential());

        // Topic subscription configuration
        cfg.SubscriptionEndpoint<OrderSubmitted>("shipping-service", e =>
        {
            e.ConfigureConsumer<OrderSubmittedConsumer>(context);
            e.MaxConcurrentCalls = 10;
            e.LockDuration = TimeSpan.FromMinutes(5);
            e.MaxAutoLockRenewalDuration = TimeSpan.FromMinutes(30);
        });

        cfg.ConfigureEndpoints(context);
    });
});

4. Ví Dụ: Order Processing Pipeline

Order API                 Service Bus                    Services
   │                         │
   │  OrderSubmitted         │
   ├────────────────────────→│──→ [Inventory Sub] → Reserve stock
   │                         │──→ [Payment Sub]   → Process payment
   │                         │──→ [Notification]  → Email confirmation
   │                         │
   │                    StockReserved
   │                         │──→ [Shipping Sub]  → Create shipment
   │                         │
   │                    PaymentFailed
   │                         │──→ [Inventory Sub] → Release stock
   │                         │──→ [Order Sub]     → Cancel order

Tổng Kết Session 20

Bài tập: Setup Azure Service Bus (Standard tier), tạo Topic "order-events" với 3 Subscriptions. Implement publisher trong Order API, consumers cho Inventory, Payment, Notification.