MassTransit ile Mesajlaşma: Consumer, Publish/Send ve Broker Topology'si — v9 Lisansı ve Alternatifler

Sürüm notu: Bu yazıdaki kodlar .NET 10, MassTransit 8.5.12 (RabbitMQ transport'u, RabbitMQ 4.4) ile derlenip denenmiştir; lisans davranışı MassTransit 9.2.5 ile denenmiştir.

Önce Tahmin Et

Serinin ikinci yazısında Billing, bookstore.ordering topic exchange'ine order.placed routing key'i ile bağlı bir queue'yu dinliyordu. Şimdi Order API'yi (Application Programming Interface) MassTransit'e geçiriyoruz ve siparişi IPublishEndpoint.Publish(new OrderPlaced(...)) ile yayınlıyoruz. Eski, ham RabbitMQ.Client ile yazılmış Billing bu mesajı alır mı? Almazsa mesaj hangi exchange'e gitti?

İkinci soru: MassTransit'in son sürümü 9.2.5'i projeye ekleyip hiçbir ek ayar yapmadan uygulamayı başlatırsan ne olur?

Neden Bir Abstraction?

Önceki beş yazıda broker'ları doğrudan kullandık ve her seferinde aynı işleri elle yaptık: bağlantıyı singleton tutmak, consumer'ı bir BackgroundService içinde yaşatmak, mesajı deserialize etmek, ack/complete/offset kararını vermek, graceful shutdown, topology'yi declare etmek, MessageId ve CorrelationId'yi taşımak. Bunlar her consumer'da tekrar eden altyapı kodu. Retry, dead-letter, outbox ve test eklenince bu kod Billing'in kendisinden büyük olur.

MassTransit, Wolverine, Rebus, NServiceBus ve Brighter gibi kütüphaneler bu altyapıyı üstlenir. Sen IConsumer<OrderPlaced> yazarsın; bağlantıyı, consumer yaşam döngüsünü, serialization'ı, topology'yi, hata yönetimini kütüphane yapar. Bedeli de var: broker'ın inceliklerini bir katman arkasında saklar, kendi mesaj formatını ve isimlendirme convention'larını getirir, ve bir bağımlılık (son dönemde bir de lisans) ekler. Bu yazıda MassTransit 8 ile Billing'i taşıyıp neyi kazanıp neyi kaybettiğimize bakacağız.

Billing'i MassTransit'e Taşımak

İkinci yazıdaki yaklaşık 60 satırlık OrderPlacedConsumer şuna dönüşüyor:

public sealed class OrderPlacedConsumer(ILogger<OrderPlacedConsumer> logger) : IConsumer<OrderPlaced>
{
    public async Task Consume(ConsumeContext<OrderPlaced> context)
    {
        OrderPlaced order = context.Message;
        await Task.Delay(TimeSpan.FromMilliseconds(200), context.CancellationToken);
        logger.LogInformation("Invoice created for order {OrderId}: {Total} {Currency} (MessageId {MessageId}, CorrelationId {CorrelationId})",
            order.OrderId, order.Total, order.Currency, context.MessageId, context.CorrelationId);
    }
}

Ack yok, channel yok, deserialize yok. Consume exception fırlatmadan biterse MassTransit mesajı ack'ler; exception fırlatırsa mesaj billing-order-placed_error queue'suna taşınır. Bunun ayrıntılarını ve retry'ı serinin ilerideki iki yazısında konuşacağız. Kayıt:

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<OrderPlacedConsumer>()
        .Endpoint(e => e.Name = "billing-order-placed");
    x.AddConsumer<CreateInvoiceConsumer>()
        .Endpoint(e => e.Name = "billing-create-invoice");

    // MassTransit RabbitMQ'da varsayılan olarak classic queue açar; quorum'u biz istiyoruz
    x.AddConfigureEndpointsCallback((_, _, cfg) =>
    {
        if (cfg is IRabbitMqReceiveEndpointConfigurator rabbit)
        {
            rabbit.SetQuorumQueue();
        }
    });

    // transport tek satırda değişiyor, consumer kodu aynı kalıyor
    if (builder.Configuration["Transport"] == "AzureServiceBus")
    {
        x.UsingAzureServiceBus((context, cfg) =>
        {
            cfg.Host(builder.Configuration.GetConnectionString("ServiceBus"));
            cfg.ConfigureEndpoints(context);
        });
    }
    else
    {
        x.UsingRabbitMq((context, cfg) =>
        {
            cfg.Host(new Uri(builder.Configuration.GetConnectionString("RabbitMq")!));
            cfg.PrefetchCount = 10;
            cfg.ConfigureEndpoints(context);
        });
    }
});

Consumer'lar DI (Dependency Injection) container'ına scoped olarak kaydedilir; her mesaj için yeni bir scope açılır. Bu, consumer içinde bir DbContext kullanmayı doğal kılar: ham client'ta BackgroundService singleton olduğu için scope'u kendimiz açmak zorundaydık. ConfigureEndpoints, kayıtlı her consumer için bir receive endpoint (queue) oluşturur. Endpoint adını açıkça verdik; vermeseydik MassTransit consumer'ın adından türetecekti.

Queue tipiyle ilgili bir not. İlk denemede MassTransit 8.5.12 endpoint'leri classic queue olarak açtı. Serinin üçüncü yazısında kalıcı iş mesajları için quorum queue'yu varsayılan seçim yapmıştık; AddConfigureEndpointsCallback ile bunu bütün endpoint'lere uyguluyoruz. Var olan bir classic queue'nun tipi değişmeyeceği için queue'ları silip yeniden oluşturduk:

billing-order-placed	quorum	0
billing-create-invoice	quorum	0

Publish ve Send

MassTransit, event ile command arasındaki farkı API'ye taşır. Event yayınlanır (publish): yayıncı kimin dinlediğini bilmez, sıfır ya da daha fazla consumer alabilir. Command gönderilir (send): tek ve belli bir alıcısı vardır. Order API:

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

    // event: kimin dinlediğini bilmeden yayınla
    await publisher.Publish(message, context => context.CorrelationId = message.OrderId, ct);
    return Results.Accepted($"/orders/{message.OrderId}", new { message.OrderId });
});

app.MapPost("/orders/{orderId:guid}/invoice", async (Guid orderId, decimal amount, ISendEndpointProvider sendEndpoints, CancellationToken ct) =>
{
    // command: tek ve belli bir alıcıya gönder
    ISendEndpoint endpoint = await sendEndpoints.GetSendEndpoint(new Uri("queue:billing-create-invoice"));
    await endpoint.Send(new CreateInvoice(orderId, amount, "TRY"), ct);
    return Results.Accepted();
});

İki isteğin ardından Billing'in logu:

info: MassTransit[0]
      Configured endpoint billing-order-placed, Consumer: Bookstore.Billing.MassTransit.OrderPlacedConsumer
info: MassTransit[0]
      Configured endpoint billing-create-invoice, Consumer: Bookstore.Billing.MassTransit.CreateInvoiceConsumer
info: MassTransit[0]
      Bus started: rabbitmq://localhost:17201/
info: Bookstore.Billing.MassTransit.CreateInvoiceConsumer[0]
      Manual invoice requested for order 01a1b8de-6a5a-70fd-8aab-7f9dd66cdbe7: 99.90 TRY
info: Bookstore.Billing.MassTransit.OrderPlacedConsumer[0]
      Invoice created for order 01a1b907-2d51-7348-88d5-65e386490dea: 1220.50 TRY (MessageId dd880000-0000-0000-a768-08df3d41a011, CorrelationId 01a1b907-2d51-7348-88d5-65e386490dea)

MessageId'yi MassTransit kendisi üretti; CorrelationId'yi publish sırasında sipariş id'si olarak verdik. Order API'de publisher confirmation ayarı yazmadık: MassTransit'in RabbitMQ transport'u Publish'i broker onayı gelene kadar bekletiyor. Mesaj tiplerinin command ve event olarak ayrımı Messaging ve Event-Driven Architecture serisinin "Message Türleri: Command, Event ve Query" yazısında.

Topology Convention'ları ve Mesaj Zarfı

Yazının başındaki ilk sorunun cevabı RabbitMQ'da. rabbitmqctl list_exchanges ve list_bindings çıktısından MassTransit'in kurduğu kısım:

Bookstore.Contracts:CreateInvoice	fanout
Bookstore.Contracts:OrderPlaced	fanout
billing-create-invoice	fanout
billing-order-placed	fanout

Bookstore.Contracts:CreateInvoice	billing-create-invoice	exchange
billing-create-invoice	billing-create-invoice	queue
Bookstore.Contracts:OrderPlaced	billing-order-placed	exchange
billing-order-placed	billing-order-placed	queue

MassTransit, routing key'li bir topic exchange kullanmıyor. Her mesaj tipi için tipin tam adıyla bir fanout exchange açıyor (Bookstore.Contracts:OrderPlaced), her receive endpoint için queue ile aynı adda bir fanout exchange daha açıyor ve consumer'ın dinlediği mesaj tiplerinin exchange'lerini buna bağlıyor. Publish, mesaj tipinin exchange'ine gider. Yani cevap hayır: ham Billing bookstore.ordering'i dinliyor ve MassTransit oraya hiçbir şey yazmıyor. Bu routing, mesaj tipini contract'ın kendisi yapar: yeni bir consumer yalnızca IConsumer<OrderPlaced> yazarak abone olur, routing key tasarımı gerekmez. Ama topology'yi elle tasarlanmış bir broker'a uymaz; ikisini bir arada kullanacaksan exchange'leri birbirine bağlaman gerekir.

İkinci fark mesajın kendisinde. Bookstore.Contracts:OrderPlaced exchange'ine bağladığımız bir test queue'sundan okunan mesaj (host bölümü kısaltıldı):

content_type: application/vnd.masstransit+json
{
  "messageId": "dd880000-0000-0000-e34e-08df3d41b4fd",
  "requestId": null,
  "correlationId": "01a1b907-b67c-7a10-ac4c-a2820aa6dbc0",
  "conversationId": "dd880000-0000-0000-e47a-08df3d41b4fd",
  "initiatorId": null,
  "sourceAddress": "rabbitmq://localhost:17201/Mac_BookstoreOrderApiMassTransit_bus_5sryyyyyyyyyya9qbdxu4oc9g6?temporary=true",
  "destinationAddress": "rabbitmq://localhost:17201/Bookstore.Contracts:OrderPlaced",
  "responseAddress": null,
  "faultAddress": null,
  "messageType": [
    "urn:message:Bookstore.Contracts:OrderPlaced"
  ],
  "message": {
    "orderId": "01a1b907-b67c-7a10-ac4c-a2820aa6dbc0",
    "customerId": "3f2b8c1e-7d4a-4e9b-9c55-1a2b3c4d5e6f",
    "total": "450.00",
    "currency": "TRY",
    "placedAt": "2026-11-08T02:40:52.860523+00:00",
    "items": [
      {
        "bookId": "0b9d6a4e-1c2f-4a8b-b3d7-5e6f7a8b9c0d",
        "quantity": 1,
        "unitPrice": "450.00"
      }
    ]
  },
  "expirationTime": null,
  "sentTime": "2026-11-08T02:40:52.860603Z",
  "headers": {}
}

Gövde, MassTransit'in zarfı (envelope) içinde. OrderPlaced message alanında; id'ler, adresler ve mesaj tipi zarfın üst seviyesinde. Bir ayrıntıya dikkat: decimal alanlar "450.00" şeklinde string olarak yazılmış. Serinin önceki yazılarındaki ham consumer'lar bu mesajı doğrudan okuyamaz: hem zarfı bilmiyorlar hem de total'ı sayı olarak bekliyorlar. MassTransit dışı sistemlerle konuşacaksan raw JSON (JavaScript Object Notation) serializer'ını (UseRawJsonSerializer) bu endpoint'ler için açman gerekir; o zaman zarfın taşıdığı bilgileri broker header'larından okursun.

Aynı Kod, Başka Transport

MassTransit'in vaadi, consumer kodunun transport'tan bağımsız olması: yukarıdaki Program.cs'te Transport ayarı AzureServiceBus olunca UsingAzureServiceBus devreye giriyor ve OrderPlacedConsumer tek satır değişmeden çalışıyor. Azure Service Bus'ta MassTransit, RabbitMQ'daki exchange'lerin yerine mesaj tipinin adından türetilen topic'ler, her endpoint için bir queue ve topic'ten queue'ya yönlendiren subscription'lar oluşturur.

Bunu bu yazının ortamında doğrulayamadık ve dürüstçe söyleyelim: MassTransit topology'yi kurmak için Service Bus'ın administration client'ını kullanıyor; emulator bu client'ı ayrı bir HTTP (Hypertext Transfer Protocol) yönetim portundan (container içinde 5300) destekliyor. Bizim ortamımızda AMQP (Advanced Message Queuing Protocol) için 17220, yönetim için 17221 portunu eşledik ve tek bir connection string iki portu birden ifade edemedi. Billing şu hatayla yeniden denemeye girdi:

info: MassTransit[0]
      Retrying 00:00:06.9222259: ReceiveTransport faulted: sb://localhost:17220/billing-create-invoice
      MassTransit.ServiceBusConnectionException: ReceiveTransport faulted: sb://localhost:17220/billing-create-invoice
       ---> System.AggregateException: Retry failed after 4 tries. Retry settings can be adjusted in ClientOptions.Retry or by configuring a custom retry policy in ClientOptions.RetryPolicy. (An error occurred while sending the request.) (An error occurred while sending the request.) (An error occurred while sending the request.) (An error occurred while sending the request.)
       ---> Azure.RequestFailedException: An error occurred while sending the request.
       ---> System.Net.Http.HttpRequestException: An error occurred while sending the request.
       ---> System.Net.Http.HttpIOException: The response ended prematurely. (ResponseEnded)

Emulator'ü varsayılan portlarıyla (5672 ve 5300) çalıştırdığında bu sorun çıkmamalı; ama bunu biz denemedik. Buradan çıkan genel ders şu: "aynı kod her transport'ta çalışır" vaadi consumer kodu için doğru, ama topology kurulumu, yetkiler (MassTransit'in entity oluşturabilmesi için Service Bus'ta Manage yetkisi gerekir) ve yerel geliştirme ortamı transport'a göre değişir. Production'da entity'leri altyapı kodu ile oluşturup MassTransit'in topology kurulumunu kapatmak da yaygın bir seçenek.

Kazanılan ve Kaybedilen

Billing'in iki sürümünü yan yana koyunca tablo şöyle:

Kafka bu tabloda ayrı bir yerde duruyor. MassTransit Kafka'yı bir "rider" olarak destekliyor, yani Kafka bus'ın ana transport'u değil, yanında çalışan bir eklenti. Kafka ağırlıklı bir sistemde serinin beşinci yazısındaki gibi Confluent.Kafka'yı doğrudan kullanmak genellikle daha az sürprizli.

Lisans Gerçeği: MassTransit v9 ve Alternatifler

İkinci sorunun cevabı. MassTransit 9.2.5 ile, lisans ayarı olmadan:

fail: Microsoft.Extensions.Hosting.Internal.Host[11]
      Hosting failed to start
      MassTransit.ConfigurationException: The bus configuration is invalid:
      [Failure] License must be specified with SetLicense/SetLicenseLocation or by setting the MT_LICENSE/MT_LICENSE_PATH environment variables. See https://masstransit.massient.com/configuration/license for details.

Uygulama hiç başlamadı. MassTransit v9, Ocak 2026'da (NuGet'te 9.0.0, 6 Ocak 2026) Massient şirketinin ticari ürünü olarak çıktı. Yazım günü itibarıyla massient.com'daki bilgiler şöyle: v9 ve sonrası lisans gerektiriyor; mesaj ya da endpoint başına ücret olmayan, yıllık sabit fiyatlı abonelikler var; yıllık brüt geliri 1 milyon ABD dolarının altındaki şirketler ve yıllık gideri bu sınırın altındaki kâr amacı gütmeyen kuruluşlar yüzde 100 lisans indirimine başvurabiliyor (ticari destek hariç). Lisans anahtarı MT_LICENSE ya da MT_LICENSE_PATH ortam değişkeniyle veya bus configurator'da SetLicense/SetLicenseLocation ile veriliyor. v8 ve öncesi açık kaynak lisanslarıyla kalıyor (NuGet'te 8.5.12 için Apache-2.0), ama massient.com'a göre v8 desteği 31 Aralık 2026'da bitiyor. Fiyat ve koşullar değişebilir; karar vermeden önce siteyi kendin kontrol et.

Yeni bir proje için seçenekler, NuGet'teki lisans bilgileriyle. Tablodaki MIT (Massachusetts Institute of Technology) lisansı, Apache-2.0 gibi ticari kullanıma izin veren serbest bir açık kaynak lisansı:

Bu tablo bir sıralama değil. Kararı belirleyen sorular: ekibin MassTransit'e ne kadar yatırımı var, lisans bütçesi ve gelir eşiği ne durumda, ve gerçekten ihtiyaç duyduğun özellikler (saga, outbox, scheduling) hangi seçenekte olgun. Serinin geri kalanında retry, dead-letter, inbox ve outbox konularını önce ham client'la anlatıp ardından MassTransit ve Wolverine karşılıklarını göstereceğiz; böylece hangi soyutlamayı seçersen seç, altında ne olduğunu bileceksin. MassTransit 8'den 9'a geçişin ayrıntıları serinin son yazısında.

Üretimde Dikkat

Convention'ları erken sabitle. Endpoint adları ve exchange adları broker'da kalıcı object'lere dönüşür; sonradan değiştirmek, eski queue'ları boşaltmayı ve taşımayı gerektirir. Endpoint adını açıkça ver, consumer class'ının adına bağlı bırakma.

Queue tipini kontrol et. Gördüğümüz gibi MassTransit 8.5.12 RabbitMQ'da classic queue açtı; quorum istiyorsan bunu açıkça yapılandır ve deploy'dan önce broker'da doğrula.

Lisansı deploy pipeline'ının parçası yap. v9'da lisans anahtarı eksikse uygulama hiç başlamıyor; anahtarı secret store'dan MT_LICENSE olarak ver ve abonelik bitiş tarihini takvime koy. Lisans süresi dolduğunda dokümantasyona göre MassTransit uyarı logluyor ama çalışan servisleri durdurmuyor.

Zarfı ve serialization'ı dışarıya açılan her sınırda düşün. Başka dilde yazılmış bir consumer'ın okuyacağı mesajlar için raw JSON ve açık bir contract kullan.

Sık Düşülen Tuzaklar

Abstraction'ın broker'ı tamamen gizleyeceğini sanmak. Prefetch, lock süresi, queue tipi ve partition kararları hâlâ senin; sadece başka bir yerde yapılandırılıyor.

MassTransit ile ham client'ı aynı mesaj akışında habersizce karıştırmak. Farklı exchange'ler ve zarf formatı yüzünden mesajlar ya hiç ulaşmaz ya da okunamaz.

Paket sürümünü * ya da açık uçlu bırakmak. v8'den v9'a otomatik yükselen bir build, lisans anahtarı olmadığı için startup'ta düşer.

Consumer içinde exception yutmak. Consume başarılı dönerse mesaj ack'lenir; hatayı loglayıp yutarsan MassTransit'in retry ve error queue mekanizması hiç devreye girmez.

Lisans kararını erteleyip v8 ile yeni projeye başlamak. Destek bitiş tarihi yakın; geçiş maliyetini bugünden hesaba kat.

Küçük Bir Deneme

★☆☆ Notification için ikinci bir IConsumer<OrderPlaced> yaz ve notification-order-placed endpoint'ine kaydet. Bir sipariş gönder ve rabbitmqctl list_bindings çıktısında yeni bağlantıyı bul.

★★☆ billing-order-placed endpoint'ine UseRawJsonSerializer uygula ve ikinci yazıdaki ham OrderEventPublisher'ın yayınladığı bir mesajı MassTransit consumer'ına ulaştır. Hangi exchange'i hangisine bağlaman gerekti?

★★★ Aynı Billing'i Wolverine ile yaz: bir handler, RabbitMQ transport'u ve aynı OrderPlaced contract'ı. İki uygulamanın kod satırlarını, oluşturdukları topology'yi ve mesaj formatlarını karşılaştır.

Billing artık tek bir interface ile mesaj tüketiyor. Peki fatura servisi bir anlığına erişilemez olduğunda ne olacak? Bir sonraki yazıda consumer'da retry ve redelivery konusunu ele alıyoruz: geçici ve kalıcı hata ayrımı, immediate, delayed ve exponential backoff, ve bunların RabbitMQ, Azure Service Bus, Kafka ve MassTransit'teki karşılıkları.