Client Sürüm Geçişleri: RabbitMQ.Client 6→7 (IModel'dan IChannel'a) ve MassTransit 8→9

Sürüm notu: Bu yazıdaki kodlar .NET 10, RabbitMQ 4.4, RabbitMQ.Client 6.8.1 ve 7.2.2, RabbitMQ.AMQP.Client 1.4.1, MassTransit 8.5.12 ve 9.2.5 ile derlenip denenmiştir.

Önce Tahmin Et

RabbitMQ.Client 6.8.1 ile yazılmış, 35 satırlık bir producer/consumer var: exchange ve quorum queue declare ediyor, publisher confirm ile bir mesaj yayınlıyor, async bir consumer ile okuyup ack'liyor. Paketi 7.2.2'ye yükseltip derliyorsun. Kaç derleme hatası alırsın? Hataların hepsini düzeltince işin bittiğini mi düşünürsün? İkinci soru: MassTransit 8.5.12 kullanan bir Billing servisinde paketi kod değiştirmeden 9.2.5'e yükseltiyorsun. Derlenir mi, çalışır mı?

Serinin bu son yazısı sürüm geçişleri üzerine: RabbitMQ.Client 6'dan 7'ye, broker tarafında RabbitMQ 3.13'ten 4.x'e, MassTransit 8'den 9'a ve Azure Service Bus'ın eski SDK'sından (Software Development Kit) yenisine. Hepsinde gerçek derleme hataları ve çalışma zamanı sürprizleri var.

RabbitMQ.Client 7'de Ne Değişti?

RabbitMQ.Client 7.0.0, 6 Kasım 2024'te NuGet'e çıktı ve .NET client'ının en büyük kırıcı sürümü oldu: senkron API (Application Programming Interface) tamamen kaldırıldı, her şey async ve CancellationToken alan method'lara dönüştü. Resmi geçiş rehberinin özeti ve derleyicinin bize gösterdikleri:

Derleme Hataları Dalga Dalga

6.8.1 ile yazılmış, derlenen ve çalışan kod:

using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

var factory = new ConnectionFactory
{
    HostName = "localhost",
    Port = 17301,
    DispatchConsumersAsync = true
};
using IConnection connection = factory.CreateConnection();
using IModel channel = connection.CreateModel();

channel.ExchangeDeclare("bookstore.ordering", ExchangeType.Topic, durable: true);
channel.QueueDeclare("lab14.billing", durable: true, exclusive: false, autoDelete: false,
    arguments: new Dictionary<string, object> { ["x-queue-type"] = "quorum" });
channel.QueueBind("lab14.billing", "bookstore.ordering", "order.placed");
channel.ConfirmSelect();

IBasicProperties props = channel.CreateBasicProperties();
props.MessageId = Guid.NewGuid().ToString();
props.Persistent = true;
channel.BasicPublish("bookstore.ordering", "order.placed", props, Encoding.UTF8.GetBytes("{}"));
channel.WaitForConfirmsOrDie(TimeSpan.FromSeconds(5));

channel.BasicQos(0, 10, false);
var consumer = new AsyncEventingBasicConsumer(channel);
consumer.Received += async (_, ea) =>
{
    Console.WriteLine($"received {ea.BasicProperties.MessageId}");
    channel.BasicAck(ea.DeliveryTag, multiple: false);
    await Task.Yield();
};
channel.BasicConsume("lab14.billing", autoAck: false, consumer);
await Task.Delay(1000);

received 4da49a62-acfb-426a-a56d-66c3496fd3c4

Paketi 7.2.2'ye yükseltince ilk derleme beş hata veriyor:

Program.cs(9,5): error CS0117: 'ConnectionFactory' does not contain a definition for 'DispatchConsumersAsync'
Program.cs(12,7): error CS0246: The type or namespace name 'IModel' could not be found (are you missing a using directive or an assembly reference?)
Program.cs(11,40): error CS1061: 'ConnectionFactory' does not contain a definition for 'CreateConnection' and no accessible extension method 'CreateConnection' accepting a first argument of type 'ConnectionFactory' could be found (are you missing a using directive or an assembly reference?)
Program.cs(12,35): error CS1061: 'IConnection' does not contain a definition for 'CreateModel' and no accessible extension method 'CreateModel' accepting a first argument of type 'IConnection' could be found (are you missing a using directive or an assembly reference?)
Program.cs(28,10): error CS1061: 'AsyncEventingBasicConsumer' does not contain a definition for 'Received' and no accessible extension method 'Received' accepting a first argument of type 'AsyncEventingBasicConsumer' could be found (are you missing a using directive or an assembly reference?)

Beş hata az görünüyor ama yanıltıcı: IModel çözülemediği için derleyici channel üzerindeki çağrıların hiçbirini kontrol edemiyor. Bu beşini mekanik olarak düzeltip (IChannel, CreateConnectionAsync, CreateChannelAsync, ReceivedAsync, DispatchConsumersAsync'i silmek) yeniden derleyince ikinci dalga geliyor (uzun mesajların sonunu kısalttım):

Program.cs(13,9): error CS1061: 'IChannel' does not contain a definition for 'ExchangeDeclare'
Program.cs(14,9): error CS1061: 'IChannel' does not contain a definition for 'QueueDeclare'
Program.cs(16,9): error CS1061: 'IChannel' does not contain a definition for 'QueueBind'
Program.cs(17,9): error CS1061: 'IChannel' does not contain a definition for 'ConfirmSelect'
Program.cs(22,9): error CS1061: 'IChannel' does not contain a definition for 'BasicPublish'
Program.cs(23,9): error CS1061: 'IChannel' does not contain a definition for 'WaitForConfirmsOrDie'
Program.cs(25,9): error CS1061: 'IChannel' does not contain a definition for 'BasicQos'
Program.cs(33,9): error CS1061: 'IChannel' does not contain a definition for 'BasicConsume'
Program.cs(19,34): error CS1061: 'IChannel' does not contain a definition for 'CreateBasicProperties'
Program.cs(20,13): error CS1061: 'IChannel' does not contain a definition for 'BasicAck'

Gerçek bir projede bu ikinci dalga, kodun her köşesinden yüzlerce satır olarak gelir. Planlamayı ilk dalgaya göre yapma. Taşınmış hâli:

using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

var factory = new ConnectionFactory
{
    HostName = "localhost",
    Port = 17301
};
await using IConnection connection = await factory.CreateConnectionAsync();
// confirm'ler artık channel açılırken seçilir; tracking açıkken BasicPublishAsync onayı bekler
await using IChannel channel = await connection.CreateChannelAsync(
    new CreateChannelOptions(publisherConfirmationsEnabled: true, publisherConfirmationTrackingEnabled: true));

await channel.ExchangeDeclareAsync("bookstore.ordering", ExchangeType.Topic, durable: true);
await channel.QueueDeclareAsync("lab14.billing", durable: true, exclusive: false, autoDelete: false,
    arguments: new Dictionary<string, object?> { ["x-queue-type"] = "quorum" });
await channel.QueueBindAsync("lab14.billing", "bookstore.ordering", "order.placed");

var props = new BasicProperties
{
    MessageId = Guid.NewGuid().ToString(),
    Persistent = true
};
await channel.BasicPublishAsync("bookstore.ordering", "order.placed", mandatory: true, props,
    Encoding.UTF8.GetBytes("{}"));

await channel.BasicQosAsync(0, 10, false);
var consumer = new AsyncEventingBasicConsumer(channel);
consumer.ReceivedAsync += async (_, ea) =>
{
    Console.WriteLine($"received {ea.BasicProperties.MessageId}");
    await channel.BasicAckAsync(ea.DeliveryTag, multiple: false);
};
await channel.BasicConsumeAsync("lab14.billing", autoAck: false, consumer);
await Task.Delay(1000);

received 5f4d025e-06e1-4155-9017-03204cfb387e

Satır sayısı neredeyse aynı, ama üç anlamlı fark var. ConfirmSelect + WaitForConfirmsOrDie çifti gitti; confirm'ler channel açılırken seçiliyor ve tracking açıkken her BasicPublishAsync broker'ın onayını bekliyor. BasicProperties artık bir interface değil, doğrudan oluşturulan bir class. Ack çağrısı da artık await ediliyor; handler'daki yapay await Task.Yield()'e gerek kalmadı.

Derleyicinin Söylemedikleri

Derlenen kod her zaman doğru çalışan kod değil. 7.x'te davranış olarak değişen ve derleyicinin yakalamadığı noktalar:

Body'nin ömrü. Geçiş rehberine göre ReceivedAsync'te gelen ea.Body yalnız handler çalışırken geçerli; handler'dan sonra kullanacaksan (ör. bir Channel<T>'a koyup başka bir thread'de işleyeceksen) kopyalaman gerekiyor (ea.Body.ToArray()). 6.x'te bu kodu yazmış bir ekip için en sinsi hata bu: derlenir, testlerde çoğunlukla çalışır, yük altında rastgele bozuk mesaj okur.

Timeout artık iptal. Bir operasyon zaman aşımına uğradığında 7.x TimeoutException yerine iptal edilmiş bir task döndürüyor; catch (TimeoutException) blokları artık çalışmıyor ve hata yukarıya OperationCanceledException olarak çıkıyor. Retry kararını exception tipine bağlayan kod (sekizinci yazıdaki IsTransient gibi) gözden geçirilmeli.

Eşzamanlılık varsayılanı. ConsumerDispatchConcurrency varsayılan olarak 1; async handler'lar bile bir channel üzerinde sırayla çalışır. 6.x'te kendi Task.Run'ıyla paralellik kurmuş bir consumer'ı taşırken bunu bilerek ayarla; onuncu yazıdaki race'i ancak bu değeri 4 yapınca görebilmiştik.

Yeni header'lar. Confirm tracking açıkken client, yayınladığı mesajlara x-dotnet-pub-seq-no header'ını ekliyor; dokuzuncu yazıdaki MassTransit _error mesajında bu header'ı gördük. Header'ları birebir karşılaştıran ya da tüm header'ları bir yere kopyalayan kod varsa fark edersin. 7.x ayrıca ActivitySource ile kendi trace'lerini üretiyor ve dinleyici varken traceparent header'ı ekliyor (on ikinci yazı).

RabbitMQ 3.13'ten 4.x'e: Mirrored Queue'lar Gitti

Client'ı yükseltirken broker da büyük olasılıkla yükselecek. RabbitMQ 4.0, yıllardır deprecated olan classic mirrored queue'ları kaldırdı. 3.13'te çalışan bir mirroring policy'sini 4.4'e uygulamayı denedim:

$ rabbitmqctl set_policy ha-all "^lab14\." '{"ha-mode":"all","ha-sync-mode":"automatic"}' --apply-to queues
Validation failed

[{<<"ha-mode">>,<<"all">>},{<<"ha-sync-mode">>,<<"automatic">>}] are not recognised policy settings

Replikasyon gereken her queue quorum queue'ya taşınmalı ve burada bir tuzak var: bir queue'nun tipi sonradan değiştirilemez. Yukarıdaki lab14.billing quorum olarak oluşturuldu. Aynı adı argümansız (eski koddaki gibi) ve x-queue-type = classic ile declare eden iki çağrının sonucu:

qargs=none: The AMQP operation was interrupted: AMQP close-reason, initiated by Peer, code=406, text='PRECONDITION_FAILED - inequivalent arg 'x-queue-type' for queue 'lab14.billing' in vhost '/': received 'classic' but current is 'quorum'', classId=50, methodId=10
qargs=x-queue-type=classic: The AMQP operation was interrupted: AMQP close-reason, initiated by Peer, code=406, text='PRECONDITION_FAILED - inequivalent arg 'x-queue-type' for queue 'lab14.billing' in vhost '/': received 'classic' but current is 'quorum'', classId=50, methodId=10

İki çağrı da aynı hatayı aldı: argüman verilmediğinde client "classic" istemiş sayılıyor. Yani queue'yu quorum'a taşıyıp declare kodunu güncellemeyi unuttuğun tek bir eski servis, açılışta channel'ını kaybeder. Vhost'un varsayılan queue tipini quorum yapmak argümansız declare'leri yeni queue'lar için çözer, ama zaten var olan bir queue'nun tipini değiştirmez. Taşıma yolu her zaman aynı: yeni adla bir quorum queue oluştur, binding'leri ekle, consumer'ları yeni queue'ya geçir, eski queue boşalınca sil; trafiği kesmeden taşımak için Shovel ya da yeni bir vhost ile blue-green geçiş kullanılır. Quorum queue'lar her zaman durable'dır, exclusive ya da auto-delete olamaz; bu özellikleri kullanan geçici queue'lar classic kalmalı.

4.0'ın ikinci sessiz değişikliği sekizinci yazıda gördüğümüz delivery limit: quorum queue'larda varsayılan 20. 3.13'te sonsuza kadar requeue edilen bir zehirli mesaj, 4.x'te 20 denemeden sonra DLX'e (Dead Letter Exchange) gider; DLX yoksa silinir. Yükseltmeden önce her quorum queue'nun bir DLX'i olduğunu doğrula. Bir de sıralama: 3.13'ten 4.0'a geçmeden önce bütün stable feature flag'lerin açık olması gerekiyor; açık değilse yükseltme yapılamaz.

İyi haber: istemci ve broker sürümleri birbirinden bağımsız. Yukarıdaki 6.8.1 kodu RabbitMQ 4.4'e bağlanıp sorunsuz çalıştı; önce broker'ı, sonra servisleri birer birer yükseltebilirsin.

AMQP 1.0 ve RabbitMQ.AMQP.Client

RabbitMQ 4.0 ile AMQP (Advanced Message Queuing Protocol) 1.0, plugin gerektirmeyen çekirdek bir protokol oldu ve RabbitMQ ekibi onun için yeni bir .NET client'ı yayınladı: RabbitMQ.AMQP.Client. API, 0-9-1 client'ından farklı; channel kavramı yok, topology yönetimi ayrı bir management nesnesinden yapılıyor:

ConnectionSettings settings = ConnectionSettingsBuilder.Create()
    .Host("localhost").Port(17301).ContainerId("bookstore-billing").Build();
IEnvironment environment = AmqpEnvironment.Create(settings);
IConnection connection = await environment.CreateConnectionAsync();

IManagement management = connection.Management();
IQueueInfo queue = await management.Queue("lab14.amqp10").Type(QueueType.QUORUM).DeclareAsync();
Console.WriteLine($"queue {queue.Name()} type={queue.Type()}");

IPublisher publisher = await connection.PublisherBuilder().Queue("lab14.amqp10").BuildAsync();
PublishResult result = await publisher.PublishAsync(new AmqpMessage("{}").MessageId("msg-1"));
Console.WriteLine($"publish outcome: {result.Outcome.State}");

var received = new TaskCompletionSource<string>();
IConsumer consumer = await connection.ConsumerBuilder()
    .Queue("lab14.amqp10")
    .MessageHandler((context, message) =>
    {
        context.Accept();
        received.TrySetResult($"{message.MessageId()}");
        return Task.CompletedTask;
    })
    .BuildAndStartAsync();
Console.WriteLine($"consumed {await received.Task.WaitAsync(TimeSpan.FromSeconds(5))}");

await consumer.CloseAsync();
await publisher.CloseAsync();
await management.Queue("lab14.amqp10").DeleteAsync();
await connection.CloseAsync();

queue lab14.amqp10 type=QUORUM
publish outcome: Accepted
consumed msg-1

Publish'in sonucu bir "outcome": Accepted, broker'ın mesajı kabul ettiği anlamına geliyor; 0-9-1'deki publisher confirm'in karşılığı protokolün kendisinde. Consumer tarafında ack yerine Accept, reject yerine Discard, requeue yerine Requeue var. Bu client'a geçmek için şu an zorunlu bir sebep yok; RabbitMQ.Client 7 desteklenmeye devam ediyor ve bu serinin bütün örnekleri onunla çalıştı. Ama 4.4'ün bazı yenilikleri (ör. consumer'ın mesajı bir token ile erteleyip sonra geri alabildiği deferred messages) yalnız AMQP 1.0 üzerinden kullanılabiliyor; yeni bir RabbitMQ-only projede bu client'ı değerlendirmeye değer.

MassTransit 8'den 9'a

MassTransit 9.0.0, 6 Ocak 2026'da NuGet'e çıktı ve ticari lisansla geldi. Sekizinci yazıdaki Billing örneğini hiçbir kod satırına dokunmadan 8.5.12'den 9.2.5'e yükselttim. Proje hatasız derlendi; çalıştırınca:

Unhandled exception. 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.

Bus başlamıyor ve host ayağa kalkmıyor. Lisans dört yoldan verilebiliyor: MT_LICENSE ortam değişkeni (anahtarın kendisi), MT_LICENSE_PATH (dosya yolu) ya da bus configurator'daki SetLicense ve SetLicenseLocation. Doğrulama yerel yapılıyor, bir lisans sunucusuna gidilmiyor; dokümana göre abonelik bittiğinde çalışan servisler durmuyor, yalnız uyarı log'lanıyor ve abonelik süresince çıkmış sürümler kullanılmaya devam edebiliyor. Massient'in sitesine göre lisans geliştirme, CI (Continuous Integration), test, staging ve production'ı kapsıyor; yıllık geliri 1 milyon doların altındaki şirketler ve yıllık gideri 1 milyon doların altındaki kâr amacı gütmeyen kuruluşlar için %100 indirimli seçenekler var.

Bir gözlem: aynı 9.2.5 paketiyle on üçüncü yazıdaki test harness testleri lisans olmadan geçti; kontrol, in-memory harness'ta tetiklenmedi. Bu teknik bir davranış, lisans izni değil: lisans koşulları CI ve testi de kapsıyor.

Kodunu etkileyebilecek 9.x değişiklikleri (değişiklik kaydından): JSON (JavaScript Object Notation) serialization ayarları global değil, bus ya da endpoint başına yapılıyor; deprecated TransactionalBus 9.2'de kaldırıldı; 9.1'de transactional outbox tabloları değişti ve şema migration'ı gerekiyor; 9.1 ile RabbitMQ için plugin istemeyen UseQueueBasedDelayedRedelivery ve test harness'ı için harness.Act(...) geldi. Bir de paket yönetimi: dotnet add package MassTransit.RabbitMQ sürüm vermeden çalıştırılırsa artık 9.x gelir. v8'de kalacaksan sürümü açıkça sabitle ve Dependabot/Renovate gibi araçlarda major yükseltmeyi kapat.

Karar zamanı: Massient, v8 desteğinin 31 Aralık 2026'da biteceğini duyuruyor. Bu yazının yayınlandığı Kasım 2026 itibarıyla v8'de kalmak, birkaç hafta sonra güvenlik düzeltmesi almayan bir bağımlılıkla yaşamak demek. Seçenekler serinin yedinci yazısındaki tabloda: v9 lisansı, Wolverine (MIT, yani Massachusetts Institute of Technology lisansı), Rebus, NServiceBus (ticari) ya da bu serinin büyük kısmında yaptığımız gibi doğrudan client'lar.

Eski Service Bus SDK'sından Yenisine

Azure tarafında geçiş çoktan zorunlu hâle geldi: NuGet'teki Microsoft.Azure.ServiceBus paketi, 30 Eylül 2026 itibarıyla obsolete olduğunu ve bakımının yapılmadığını, yerine Azure.Messaging.ServiceBus'ın kullanılmasını söylüyor. Eski bir kod tabanını okurken karşılıklar:

Bu bölümü bir okuma notu olarak bıraktım; eski SDK ile kod çalıştırmadım. Geçişte kavramlar aynı kalıyor, değişen nesnelerin yaşam döngüsü: yeni SDK'da tek bir ServiceBusClient uygulama boyunca yaşar ve sender, receiver, processor ondan üretilir. Serinin dördüncü yazısındaki singleton kuralı tam olarak bu.

Üretimde Dikkat

Bir seferde bir şey yükselt. Broker major sürümü, client major sürümü ve abstraction kütüphanesi aynı release'te değişirse, çıkan bir sorunun hangisinden geldiğini bulmak saatler sürer. Önce broker (client'lar geriye uyumlu), sonra servisler tek tek.

Testleri iki sürüme karşı çalıştır. On üçüncü yazıdaki Testcontainers testlerini geçiş süresince hem eski hem yeni broker image'ı ile çalıştırmak, davranış farklarını (delivery limit, queue tipi) production'dan önce gösterir.

Consumer'ı canary olarak taşı. Bir queue'nun consumer'larından yalnız birini yeni client sürümüyle çalıştırıp hata, işlem süresi ve DLQ (Dead-Letter Queue) oranını on ikinci yazıdaki metric'lerle karşılaştırmak, tüm filoyu birden yükseltmekten daha güvenlidir.

Lisansı bir sır gibi yönet. MassTransit v9 lisans anahtarı, bağlantı cümleleri gibi bir secret'tır; kaynak koduna değil, secret store'a ve ortam değişkenine aittir. Süresinin dolması servisleri durdurmasa da yeni sürüm deploy'unu engelleyebilir; yenileme tarihini takvime koy.

Sık Düşülen Tuzaklar

Geçişi ilk derleme dalgasına göre planlamak. IModel çözülmeden channel çağrılarının hataları görünmez; gerçek iş ikinci dalgada.

ea.Body'yi handler dışında kullanmak. 7.x'te body yalnız handler süresince geçerli; kopyalamadan başka bir thread'e vermek rastgele veri bozulması demek.

catch (TimeoutException)'a güvenmek. 7.x'te zaman aşımı iptal olarak gelir; retry ve hata sınıflandırma kodunu buna göre güncelle.

Queue tipini yerinde değiştirmeye çalışmak. Aynı adla farklı tipte declare her zaman PRECONDITION_FAILED verir; taşıma yeni bir queue ile yapılır.

DLX'siz quorum queue ile 4.x'e geçmek. Varsayılan delivery limit, 3.13'te sonsuz döngüye giren zehirli mesajı 4.x'te sessizce siler.

Sürümü sabitlenmemiş paket. CI'da sürüm vermeden dotnet add package ya da geniş bir sürüm aralığı, bir gün MassTransit'i 9'a taşır ve servis lisans hatasıyla açılmaz.

Küçük Bir Deneme

★☆☆ 6.8.1 kodundaki consumer'ı, mesaj body'sini bir Channel<ReadOnlyMemory<byte>>'a koyup ayrı bir task'te işleyecek şekilde değiştir ve 7.2.2'ye taşı. Kopyalamadan ve ToArray() ile kopyalayarak 1000 mesajla çalıştır, sonuçları karşılaştır.

★★☆ billing.order-placed'i trafiği kesmeden classic'ten quorum'a taşı: yeni adla bir quorum queue oluştur, aynı binding'leri ekle, consumer'ı yeni queue'ya çevir ve eski queue boşaldığında sil. Taşıma sırasında hiçbir mesajın kaybolmadığını on üçüncü yazıdaki gibi bir testle göster.

★★★ MassTransit kullanan bir servisi, aynı consumer'ları koruyarak ham RabbitMQ.Client 7'ye taşı: sekizinci yazıdaki retry topology'si, dokuzuncu yazıdaki DLQ, onuncu yazıdaki inbox ve on birinci yazıdaki outbox. Hangi parçayı kendin yazmak zorunda kaldın ve bunun bakım maliyeti v9 lisansıyla karşılaştırıldığında nasıl görünüyor?

Bu yazıyla Message Queue'larla Çalışmak serisi bitiyor. On dört yazı boyunca aynı OrderPlaced mesajını RabbitMQ, Azure Service Bus ve Kafka üzerinden gönderdik; retry'ı, DLQ'yu, idempotency'yi, outbox'ı, observability'yi ve testleri kodla kurduk ve her birinde gerçek çıktılara baktık. Buradan devam etmek istersen kavramsal tarafı derinleştiren Messaging ve Event-Driven Architecture serisine, consumer'ları barındıran arka plan işleri için ilerideki Background Job'lar serisine, trace ve metric'leri toplayıp görselleştirmek için de ilerideki Observability serisine bakabilirsin.