Message Bus & Distributed Systems 49 okunma

MassTransit ile .NET Bus Mimarisi

.NET uygulamalarında MassTransit kütüphanesi ile RabbitMQ, Azure Service Bus veya Kafka üzerinde mesajlaşma altyapısı kurulumu, publish/subscribe ve sagas desteği anlatılır.

MassTransit ile .NET Bus Mimarisi

MassTransit ile .NET Üzerinde Bus Mimarisi Kurmak

MassTransit, .NET ekosisteminin en olgun ve en yaygın kullanılan açık kaynaklı distributed application framework'üdür. RabbitMQ, Azure Service Bus, Amazon SQS, ActiveMQ ve Kafka gibi farklı message broker'ları tek bir soyutlama (abstraction) altında birleştirir. Geliştiricilerin alt yapıdaki mesajlaşma detaylarıyla uğraşmasını engelleyerek, iş mantığına odaklanmasını sağlar. Bu yazıda, MassTransit'in temel kavramlarını, RabbitMQ, Azure Service Bus ve Kafka ile kurulumunu, publish/subscribe mekanizmasını ve saga desteğini detaylıca ele alacağız.

1. MassTransit Nedir ve Neden Kullanılır?

MassTransit, uygulamalar arasında güvenilir, esnek ve ölçeklenebilir mesajlaşma sağlayan bir servis bus'tur. Uygulama kodunuzu, altta yatan mesajlaşma teknolojisinden (transport) tamamen izole eder. Aynı kod tabanı, sadece konfigürasyon değiştirerek RabbitMQ'dan Azure Service Bus'a veya Kafka'ya taşınabilir.

MassTransit'in Sağladığı Temel Faydalar:

  • Transport Soyutlaması: Farklı broker'lar arasında geçiş yapmayı kolaylaştırır.

  • Gelişmiş Desen Desteği: Publish/Subscribe, Request/Response, Sagas (State Machine), Outbox Pattern, Routing Slip gibi dağıtık sistem desenlerini native olarak destekler.

  • Kolay Konfigürasyon: .AddMassTransit() extension metodu ile DI container'a entegrasyonu çok basittir.

  • Otomatik Consumer Discovery: Consumer ve Saga sınıflarını otomatik olarak tarayıp kaydeder.

  • Test Edilebilirlik: In-Memory transport ile entegrasyon testleri yazmak çok kolaydır.

2. Temel Kavramlar

  • Message (Mesaj): İletişimin temel birimidir. Basit bir C# class veya record'dur. İmmutable (değişmez) olması önerilir.

  • Consumer (Tüketici): Bir mesaj tipini işleyen sınıftır. IConsumer<T> arayüzünü implemente eder.

  • Producer (Üretici): Mesajı gönderen uygulamadır. IPublishEndpoint veya ISendEndpoint üzerinden mesaj gönderir.

  • Bus (Otobüs): Mesajlaşma altyapısının merkezi nesnesidir. Tüm publish/send işlemleri bus üzerinden yapılır.

  • Endpoint (Uç Nokta): Bir consumer'ın mesajları dinlediği kuyruk veya topic'tir.

  • Saga (Saga): Uzun süreli, state'li (durumlu) iş akışlarını (workflow) yönetmek için kullanılan state machine'dir.

3. Kurulum ve Konfigürasyon

MassTransit, NuGet paketleri ile kurulur. Temel paket MassTransit'tir. Kullanılacak transport'a göre ek paketler yüklenir.

A. RabbitMQ ile Kurulum

bash

dotnet add package MassTransit
dotnet add package MassTransit.RabbitMQ

Program.cs (ASP.NET Core):

csharp

using MassTransit;

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddMassTransit(x =>
{
    // Consumer'ları otomatik olarak bul ve kaydet
    x.AddConsumers(typeof(Program).Assembly);
    
    // RabbitMQ transport'unu yapılandır
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("localhost", "/", h =>
        {
            h.Username("guest");
            h.Password("guest");
        });

        // Tüm consumer'lar için endpoint'leri otomatik yapılandır
        cfg.ConfigureEndpoints(context);
    });
});

var app = builder.Build();
app.Run();

Bu konfigürasyon, AddConsumers ile bulunan tüm consumer'lar için otomatik olarak kuyruk (queue) ve binding oluşturur.

B. Azure Service Bus ile Kurulum

bash

dotnet add package MassTransit
dotnet add package MassTransit.Azure.ServiceBus.Core

csharp

builder.Services.AddMassTransit(x =>
{
    x.AddConsumers(typeof(Program).Assembly);
    
    x.UsingAzureServiceBus((context, cfg) =>
    {
        cfg.Host("your-connection-string"); // veya cfg.Host(serviceBusNamespace, h => { ... })
        cfg.ConfigureEndpoints(context);
    });
});

Azure Service Bus konfigürasyonu, connection string veya namespace + token credential ile yapılabilir.

C. Kafka (Rider) ile Kurulum

Kafka, MassTransit'de diğer transport'lardan farklı olarak bir "Rider" olarak tanımlanır.

bash

dotnet add package MassTransit
dotnet add package MassTransit.Kafka

csharp

builder.Services.AddMassTransit(x =>
{
    x.AddRider(rider =>
    {
        rider.AddConsumer<KafkaMessageConsumer>();
        
        rider.UsingKafka((context, kcfg) =>
        {
            kcfg.Host("localhost:9092");
            
            kcfg.TopicEndpoint<string, string>("my-topic", "consumer-group", e =>
            {
                e.ConfigureConsumer<KafkaMessageConsumer>(context);
            });
        });
    });
});

Kafka rider, topic'leri tüketmek ve üretmek için kullanılır. Producer tanımlamak için ITopicProducer<T> arayüzü DI üzerinden enjekte edilir.

4. Mesaj Tanımlama, Gönderme ve Tüketme

A. Mesaj (Message) Tanımlama

csharp

// Mesaj sınıfı (immutable olması önerilir)
public record OrderSubmitted
{
    public Guid OrderId { get; init; }
    public string CustomerName { get; init; }
    public decimal TotalAmount { get; init; }
}

B. Consumer (Tüketici) Oluşturma

csharp

public class OrderSubmittedConsumer : IConsumer<OrderSubmitted>
{
    private readonly ILogger<OrderSubmittedConsumer> _logger;

    public OrderSubmittedConsumer(ILogger<OrderSubmittedConsumer> logger)
    {
        _logger = logger;
    }

    public async Task Consume(ConsumeContext<OrderSubmitted> context)
    {
        var message = context.Message;
        _logger.LogInformation("Sipariş alındı: {OrderId} - {CustomerName}", 
            message.OrderId, message.CustomerName);
        
        // İş mantığını çalıştır...
        await Task.Delay(100);
    }
}

C. Mesaj Gönderme (Publish / Send)

  • Publish (Yayın): Bir mesajı, o mesajı dinleyen tüm consumer'lara gönderir (pub/sub). Birden fazla consumer aynı mesajı alabilir.

  • Send (Gönder): Belirli bir endpoint'e (kuyruk) mesaj gönderir. Tek bir consumer hedeflenir.

csharp

[ApiController]
[Route("api/[controller]")]
public class OrdersController : ControllerBase
{
    private readonly IPublishEndpoint _publishEndpoint;
    private readonly ISendEndpointProvider _sendEndpointProvider;

    public OrdersController(IPublishEndpoint publishEndpoint, ISendEndpointProvider sendEndpointProvider)
    {
        _publishEndpoint = publishEndpoint;
        _sendEndpointProvider = sendEndpointProvider;
    }

    [HttpPost]
    public async Task<IActionResult> CreateOrder(OrderRequest request)
    {
        // Tüm subscriber'lara yayınla
        await _publishEndpoint.Publish(new OrderSubmitted
        {
            OrderId = Guid.NewGuid(),
            CustomerName = request.CustomerName,
            TotalAmount = request.TotalAmount
        });

        // Belirli bir kuyruğa gönder
        var endpoint = await _sendEndpointProvider.GetSendEndpoint(new Uri("queue:order-queue"));
        await endpoint.Send(new OrderSubmitted { ... });

        return Ok();
    }
}

5. Saga (State Machine) ile Uzun Süreli İş Akışları

Saga'lar, birden fazla servisi kapsayan, state'li (durumlu) ve uzun süren iş akışlarını (workflow) yönetmek için kullanılır. MassTransit, saga'ları Automatonymous state machine ile implemente eder.

Örnek: Sipariş Süreci Saga'sı

Bir siparişin ödeme onayı ve stok rezervasyonu adımlarını içeren bir saga tasarlayalım.

Adım 1: State (Durum) ve Events (Olaylar) Tanımlama

csharp

// Saga State (Durum)
public class OrderState : SagaStateMachineInstance
{
    public Guid CorrelationId { get; set; }
    public string CurrentState { get; set; }
    public Guid OrderId { get; set; }
    public decimal TotalAmount { get; set; }
    public bool PaymentConfirmed { get; set; }
    public bool StockReserved { get; set; }
}

// Events (Olaylar)
public record OrderSubmittedEvent(Guid OrderId, decimal TotalAmount);
public record PaymentConfirmedEvent(Guid OrderId);
public record StockReservedEvent(Guid OrderId);
public record OrderCompletedEvent(Guid OrderId);
public record OrderFailedEvent(Guid OrderId, string Reason);

Adım 2: State Machine (Saga) Tanımlama

csharp

public class OrderStateMachine : MassTransitStateMachine<OrderState>
{
    public State Submitted { get; set; }
    public State PaymentProcessing { get; set; }
    public State StockProcessing { get; set; }
    public State Completed { get; set; }
    public State Failed { get; set; }

    public Event<OrderSubmittedEvent> OrderSubmitted { get; set; }
    public Event<PaymentConfirmedEvent> PaymentConfirmed { get; set; }
    public Event<StockReservedEvent> StockReserved { get; set; }

    public OrderStateMachine()
    {
        // Başlangıç durumu
        Initially(
            When(OrderSubmitted)
                .Then(context =>
                {
                    context.Instance.OrderId = context.Message.OrderId;
                    context.Instance.TotalAmount = context.Message.TotalAmount;
                })
                .Publish(context => new ProcessPaymentCommand(context.Instance.OrderId))
                .TransitionTo(Submitted)
        );

        // Ödeme onaylandı
        During(Submitted,
            When(PaymentConfirmed)
                .Then(context => context.Instance.PaymentConfirmed = true)
                .Publish(context => new ReserveStockCommand(context.Instance.OrderId))
                .TransitionTo(PaymentProcessing)
        );

        // Stok rezerve edildi
        During(PaymentProcessing,
            When(StockReserved)
                .Then(context => context.Instance.StockReserved = true)
                .Publish(context => new OrderCompletedEvent(context.Instance.OrderId))
                .Finalize() // Completed durumuna geç
        );

        // Hata durumu (örnek: ödeme başarısız)
        During(Submitted,
            When(PaymentFailed)
                .Publish(context => new OrderFailedEvent(context.Instance.OrderId, "Ödeme başarısız"))
                .TransitionTo(Failed)
        );
    }
}

Adım 3: Saga'yı Kaydetme ve Repository Konfigürasyonu

csharp

builder.Services.AddMassTransit(x =>
{
    x.AddConsumers(typeof(Program).Assembly);
    
    // Saga state machine'i kaydet
    x.AddSagaStateMachine<OrderStateMachine, OrderState>()
        .EntityFrameworkRepository(r =>
        {
            r.ExistingDbContext<AppDbContext>(); // EF Core ile
            r.ConcurrencyMode = ConcurrencyMode.Pessimistic;
        });
    
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("localhost", "/", h => { ... });
        cfg.ConfigureEndpoints(context);
    });
});

Saga repository'si olarak Entity Framework Core, NHibernate, MongoDB veya In-Memory kullanılabilir.

6. Request/Response (İstek/Yanıt) Deseni

MassTransit, senkron iletişim gerektiğinde Request/Response desenini de destekler.

csharp

// Request mesajı
public record CheckStockRequest(int ProductId, int Quantity);
// Response mesajı
public record CheckStockResponse(int ProductId, bool InStock);

// Consumer (yanıtlayan)
public class StockChecker : IConsumer<CheckStockRequest>
{
    public async Task Consume(ConsumeContext<CheckStockRequest> context)
    {
        var inStock = await _stockService.CheckStockAsync(context.Message.ProductId, context.Message.Quantity);
        await context.RespondAsync(new CheckStockResponse(context.Message.ProductId, inStock));
    }
}

// İstek gönderen (Client)
var client = _requestClient.GetRequestClient<CheckStockRequest>();
var response = await client.GetResponse<CheckStockResponse>(new CheckStockRequest(1, 10));

7. En İyi Pratikler ve İpuçları

  1. Mesajları İmmutable (Değişmez) Yapın: Mesaj sınıflarını record veya init property'ler ile tanımlayın. Bu, mesaj bütünlüğünü korur.

  2. Consumer'ları State'siz (Stateless) Tasarlayın: Consumer'lar içinde field (değişken) tutmayın. Tüm bağımlılıkları constructor üzerinden alın.

  3. Hata Yönetimi ve Retry: MassTransit'in UseRetry() ve UseCircuitBreaker() middleware'lerini kullanarak dayanıklılığı artırın.

  4. Outbox Pattern: Veritabanı işlemi ile mesaj göndermeyi atomik hale getirmek için MassTransit'in Outbox desteğini kullanın.

  5. Saga Repository'sini Durable (Kalıcı) Yapın: Production ortamında In-Memory repository kullanmayın. EF Core veya MongoDB gibi kalıcı bir repository tercih edin.

  6. Endpoint Konfigürasyonu: ConfigureEndpoints(context) kullanarak consumer'larınızı otomatik olarak endpoint'lere bağlayın.

  7. Mesaj Tiplerini Paylaşın: Producer ve Consumer arasında mesaj tiplerini paylaşmak için ayrı bir Contracts (sözleşme) projesi oluşturun.

Sonuç:

MassTransit, .NET uygulamalarında mesajlaşma altyapısı kurmanın en etkili ve en esnek yoludur. RabbitMQ, Azure Service Bus veya Kafka gibi farklı transport'ları tek bir API altında birleştirerek, geliştiricilerin iş mantığına odaklanmasını sağlar. Publish/Subscribe, Request/Response ve Saga (State Machine) gibi güçlü desenleri native olarak destekler.

MassTransit'i bir projeye entegre etmek, sadece bir NuGet paketi eklemek ve birkaç satır konfigürasyon yapmaktan ibarettir. Doğru yapılandırıldığında, uygulamalar arası iletişimi güvenilir, ölçeklenebilir ve yönetilebilir hale getirir. Dağıtık sistemler veya mikroservis mimarileri için MassTransit, .NET ekosistemindeki en olgun ve en güçlü çözümlerden biridir.

Tüm yazılar