Merhaba, bu yazıda message broker kullandığımız sistemlerde asenkron iletişimi nasıl daha güvenilir hale getirebileceğimizden bahsedeceğim.
Özellikle bir mesajı publish ettiğimizde, bu mesajın gerçekten broker tarafından alınıp alınmadığını nasıl anlayabileceğimizi ve olası bağlantı problemlerinde mesaj kaybını nasıl önleyebileceğimizi inceleyeceğiz.
1. Bir Mesajı Publish Etmek Gerçekten Yeterli mi?
RabbitMQ ile çalışmaya ilk başladığımızda akış oldukça basit görünür. Publisher kendi içerisinde yapması gereken işlemi yapar, ardından yaptığı işlem ile ilgili bir mesaj publish eder. Publish edilen mesaj broker’a ulaşır ve ilgili consumerlar bu mesajları tüketir.
Peki ilk başladığımızda bu kadar basit görünen şey, neden karmaşıklaşır?
Bu sorunun cevabını doğrudan vermek yerine, bir sistem kurgulayıp o kurgu üzerine düşünürken bulalım istiyorum. Bu sebeple şu örnek ile devam edelim.
Bir e-ticaret platformunda sipariş oluşturuldukça, ilgili siparişleri oluşturan müşterilere siparişlerinin başarıyla oluşturulduğuna dair bildirim gönderilir. Biz de bu yapıyı basitçe tasarlayalım, sonrasında başımıza gelebilecek problemleri deneyimleyip bu problemleri çözmek adına neler yapabileceğimize bakalım.
Kurguladığımız yapı şuanda bu şekilde. İsterseniz direkt bir mesaj publish ederek akışın nasıl olduğuna bakalım.
İlgili kod ile event’ımızı broker’a publish ettik. Alttaki görselde de mesajın başarıyla broker’a iletildiğini görebiliriz.
Şuanda istediğimiz şeyi kolaylıkla yapabildik çünkü her şey istediğimiz gibiydi. Peki her şey her zaman istediğimiz gibi olabilir mi?
Message broker’lar çoğu zaman bir sunucuda çalışır ve ağ üzerinden iletişime geçilebilir haldedirler. Dolayısıyla hem producer, hem de consumer’lar ağ üzerinden broker’a yazabilir ve tüketebilir. Az önceki senaryomuzda ağ gayet sağlıklıydı ve biz istediğimizi yapabildik. İsterseniz şimdi de broker’ın çalışmayı durdurduğu senaryoda neler olacağını gözlemleyelim. Çünkü gerçek hayatta her şey istediğimiz gibi olmayabilir.
Kodumuzu şu şekilde güncelledim:
Uygulama mesajı publish etmeden önce broker process’ini durduracağım ve mesajı publish edeceğim. Mesaj publish ettikten sonra da broker’ı tamamen kapatacağım. Bakalım mesajımız gerçekten broker tarafından alınacak mı?
Eklediğim GIF’te gördüğünüz üzere producer tarafı mesajı başarıyla publish ettiğini söylerken broker tarafında bu mesaj yok.
Tabiri caizse publisher ile broker arasında şöyle bir muhabbet var:
Biz producer olarak mesajı publish ettik ama broker mesajımızı alamadı. Dolayısıyla mesaj da kayboldu.
Peki biz burada nasıl bir yapı olmasını isterdik?
Broker keşke mesajı başarıyla alıp almadığının bilgisini bize verseydi, biz de ona göre mesajımızı tekrardan publish eder yada silerdik.
2. Publisher Confirms
Publisher confirms, publisher tarafından broker’a iletilen mesajların broker tarafından başarılı şekilde alınıp alınmadığı bilgisinin producer’a iletilmesini sağlar. Broker, publisher confirm özelinde ACK ve NACK olmak üzere 2 tür bilgilendirmeyi publisher’a iletir.
ACK, mesajın broker tarafından başarıyla kabul edildiğini doğrularken; NACK, broker’ın mesaj için olumlu bir publisher confirm veremediğini ifade eder.
Böylece publisher ilettiği mesajların durumunu takip edebilir ve gerekli aksiyonları alabilir.
Bu yapı, default olarak kapalı geliyor. Dolayısıyla publisher-broker arasındaki etkileşim varsayılan olarak şu şekilde:
Publisher confirm kapalı olduğunda broker tarafından mesajın sağlıklı şekilde alınıp alınmadığı ile ilgili bilgi gönderilmiyor. Dolayısıyla publisher tarafı ilettiği mesajın sağlıklı şekilde alınıp alınmadığı ile ilgili bilgiye sahip değil.
Bu yapıyı açmak için channel oluşturma sırasında ilgili konfigürasyonu channel’a tanımlamak gerekiyor.
Bu tanımlamayı şu şekilde yapabiliriz:
Görseldeki kod ile birlikte gerekli tanımlamayı yaptığımızda ise publisher-broker arasındaki etkileşim şu hale geliyor:
Görüldüğü üzere publisher confirm edilmesiyle birlikte publisher gönderdiği mesajın durumu ile ilgili bilgiye sahip oluyor. Herhangi bir mesaj o anda nack alırsa o mesajı tekrardan publish edebilmesi için ilgili bilgilendirmeyi alabiliyoruz artık.
Dikkat ederseniz ACK durumunda gönderdiğimiz ‘order-created’ mesajını queue içerisine yazmadım. Bu şekilde yapmamın özel bir sebebi var. ACK, ilgili mesaj broker tarafından başarılı şekilde alındığında yayınlanır. Mesajın herhangi bir queue’ya yönlendirilip yönderilmemesiyle ilgili değildir. Dolayısıyla herhangi bir mesaj broker’a sağlıklı şekilde iletilse bile bir queue’ya yönlendirilemeyebilir. Bizler tabii ki burada yönlendirilemeyen mesajların kaybolmasını istemeyiz. Buna yazının devamında değineceğim.
Bu noktaya kadar gönderdiğimiz mesajların durumuyla akalı bilgi alabilir hale geldik. Peki biz bu bilgileri nasıl alıyoruz ve bu bilgiler ışığında nasıl aksiyon alabiliriz?
İlk olarak ACK ve NACK bildirimlerinin publisher tarafında nasıl değerlendirileceğinden bahsedeceğim.
Broker tarafı, kendisine gönderilen mesajı başarılı şekilde işlediğinde ACK gönderdiğinden bahsetmiştik. Bu ACK bildirisi, publisher tarafında BasicAcksAsync() metodunu tetikliyor. Biz de bu sayede publisher tarafında ilgili mesajın onaylanması halinde yapacağımız işlemleri yapabiliyoruz.
Alttaki GIF ile akışı göstermeye çalıştım.
Aynı şey NACK için de geçerli. İlgili mesaj broker tarafından başarıyla işlenemediği durumda NACK bildirisi publisher tarafında BasicNacksAsync() metodunu tetikliyor. Aynı şeyler bu metod için de geçerli.
Bu kısma kadar ACK ve NACK bildirilerinin nasıl ele alınacağından bahsetmiş oldum. Şimdi bahsetmek istediğim konu ise biraz önce değindiğim ‘broker tarafından başarıyla alınan fakat herhangi bir queue’ya route edilemeyen’ mesajlar ile ilgili.
Mesajları publisherdan farklı şekillerde gönderebiliyoruz. Bu şekiller amacımıza göre değişeceğinden mesajlar kimi zaman kesinlikle bir queue’ya route edilmesi gerekirken kimi zamanda route edilmesi şart değildir. Publisher tarafından gönderilen mesajın kesinlikle bir queue’ya route edilmesini istiyorsak mesajı gönderirken “mandatory:true” tanımlamasını yapmamız gerekir.
await channel.BasicPublishAsync(
exchange: string.Empty,
routingKey: "order-created.queue",
mandatory: true,
body: eventBody);
Bu tanımlama ile birlikte ilgili mesaj herhangi bir queue’ya yönlendirilemediğinde publisher tarafında BasicReturnAsync() metodunu tetikleyecektir.
Unroutable mesajları publisher confirm’den ayrı şekilde değerlendirmemiz gerekir. Çünkü unroutable mesajlar broker tarafından başarıyla işlenmiş mesajlardır. Dolayısıyla “mesajım broker tarafından ACK aldı, bu yüzden brokerda illaki bir queue’ya yönlendirilmiştir” gibi bir yanılgıya düşmemeliyiz.
3. Publisher Confirmation Tracking
Buraya kadar publisher confirm mekanizmasının ACK ve NACK bildirimleri üzerinden nasıl çalıştığını gördük. Bu noktada kendimize şu soruyu sorabiliriz:
Broker’dan gelen ACK veya NACK bilgisinin hangi publish işlemimize ait olduğunu nasıl takip edeceğiz?
Bu gayet mantıklı bir soru. İncelediğimiz örneklerde birer mesaj üzerinden gösterim yaptık. Ya binlerce mesaj publish edeceksek bunların takibini nasıl yapacağız?
RabbitMQ .NET Client bu noktada bize publisherConfirmationTrackingEnabled adında bir özellik sunuyor.
Channel oluştururken bu özelliği şu şekilde aktif edebiliriz:
Burada publisherConfirmationsEnabled ile Publisher Confirms mekanizmasını aktif ederken, publisherConfirmationTrackingEnabled ile RabbitMQ Client’ın confirmation işlemlerini bizim yerimize takip etmesini sağlıyoruz.
Peki tracking özelliğini true yaptığımızda tam olarak ne oluyor?
Tracking aktif olduğunda RabbitMQ .NET Client, publish ettiğimiz mesaj ile broker’dan gelen confirmation bilgisini kendi içerisinde eşleştiriyor. Böylece BasicPublishAsync() metodunu await ettiğimizde publish işleminin sonucunu doğrudan ilgili task üzerinden takip edebiliyoruz.
Örneğin:
try
{
await channel.BasicPublishAsync(
exchange: string.Empty,
routingKey: "order-created.queue",
mandatory: true,
body: eventBody);
Console.WriteLine("Message successfully confirmed.");
}
catch (Exception ex)
{
Console.WriteLine($"Publish failed: {ex.Message}");
}
Bu yapıda mesaj ACK aldığında BasicPublishAsync() başarılı şekilde tamamlanır.
Mesaj NACK aldığında veya mandatory : true olduğu halde herhangi bir queue’ya route edilemediğinde ise publish işlemi exception ile sonuçlanır. Dolayısıyla tracking açıkken confirmation sürecinin önemli bir kısmını RabbitMQ Client bizim yerimize yönetiyor.
Client bunu nasıl yönetiyor diye düşünüyor olabiliriz. Tam da şu şekilde yönetiyor :
RabbitMQ.Client gönderdiği her bir mesaj için ilgili channel içerisinde lineer artan bir değişken tutar. Bu değişken değeri her iletilen mesaj ile birlikte birer birer artar. Publisher tarafında ilettiği her bir mesajın sequental number değerini kendi içerisinde saklar ve mesajı publish eder. Brokerdan dönen ACK/NACK bildirisinin argümanlarında ‘DeliveryTag’ değeri bulunur. Bu deliveryTag değeri de aslında mesaja atadığı sequental number’ın kendisidir. Bu değere göre gönderdiği mesajlardan hangisinin başarılı yada başarısız şekilde iletildiğini takip edebilir.
Bu oldukça kullanışlıdır. Çünkü gönderdiğimiz her mesaj için sequence number üretmek, pending mesajları saklamak ve ACK/NACK geldiğinde ilgili mesajı manuel olarak bulmak zorunda kalmayız.
Ancak bu kolaylığın karşılığında bazı kontrolleri de client’a bırakmış oluruz.
Örneğin tracking açıkken confirmation mekanizmasının iç detaylarına uygulama seviyesinde erişimimiz oldukça kısıtlanıyor.
Örneğin:
- Hangi mesajlar ACK aldı?
- Hangi mesajlar hala pending durumda?
- Connection koptuğu anda hangi mesajların durumu artık kesin olarak bilinemeyecek?
- Recovery sonrasında hangi mesajları tekrardan publish etmeliyiz?
İşte bu gibi sorulara cevap bulabilmek için publisher confirm’u kendimiz yönetmeliyiz. Kontrol elimizde olduğunda bu soruların cevaplarını kolaylıkla verebiliyor olacağız.
İsterseniz kontrolü elimize alarak devam edelim.
“publisherConfirmationTrackingEnabled : false” dediğimizde tracking’i rabbitMQ.Client üzerinden alıyor ve kendimizin yöneteceğini dikte ediyoruz.
Biraz önce RabbitMQ.Client’ın sequenceNumber’lar vesilesiyle ACK/NACK yönetimini nasıl yaptığından bahsetmiştim. Aynı mantığı kendimiz uygulayacağız. Gönderdiğimiz mesajların takibini yapabilmek için ConcurrentDictionary oluşturacağız. Mesajlarımızı publish etmeden önce bu dictionary’e atacağız ve brokerdan gelen ACK/NACK bilgisine göre dictionary’i güncelleyeceğiz. Böylece herhangi bir connection failure durumunda mesajların state’leri hakkında bilgiye sahip olacağız.
var pendingMessages = new ConcurrentDictionary<ulong, string>();
Göndereceğimiz mesajların takibini yapabilmek için bir dictionary oluşturduk.
Göndermek istediğimiz mesaj şu şekilde olsun.
var message = new OrderCreatedEvent
{
Id = Guid.NewGuid(),
OrderId = Guid.NewGuid(),
CreatedTime = DateTime.UtcNow
};
Şimdi ise, göndereceğimiz mesajın takibini yapabilmek için ilgili mesaja atanacak olan sequenceNumber bilgisini alalım.
var sequenceNumber = await channel.GetNextPublishSequenceNumberAsync();
SequenceNumber bilgisini de aldık. Artık broker’a göndereceğimiz mesajın takibini yapabilmek için herşeye sahibiz. O zaman göndereceğimiz mesajın kaydını dictionary’e yapalım ve mesajımızı publish edelim.
pendingMessages.TryAdd(
sequenceNumber,
JsonSerializer.Serialize(message));
await channel.BasicPublishAsync(
exchange: "",
routingKey: "order-created",
body: Encoding.UTF8.GetBytes(JsonSerializer.Serialize(message)));
Mesajlarımızı yayınladığımızda pendingMessages şu şekilde olacak:
Bu noktada broker’a göndereceğimiz mesajları bir dictionary aracılığı ile takip ediyoruz. Peki ACK/NACK olduğunda pendingMessages nasıl güncellenmeli?
Sizin de tahmin edeceğiniz üzere daha önce bahsettiğimiz BasicAcksAsync/BasicNacksAsync metodları ile yapacağız:
Bu metodumuz ile birlikte broker mesajı sağlıklı şekilde aldığında ilgili mesajı pendingMessages’tan çıkarıyoruz. Dolayısıyla broker tarafında oluşabilecek herhangi bir fail anında broker’a tekrardan publish edilmesi gereken mesajlar pendingMessages içerisinde bekliyor olacak. Başarıyla iletilen mesajlar pendingMessages içerisinde bulunmayacağı için aynı mesajı tekrar tekrar göndermemiş olacağız. Burada kısmi idempotentlik sağlıyoruz diyebiliriz. Çünkü ilettiğimiz mesajı tekrardan göndermiyoruz. Tabii ki asıl idempotentlik davranışını consumer tarafında kazandıracağız. Buna da bu yazının diğer partlarında değineceğim.
4. Connection Failure ve Unknown Mesajlar
Yazımızın bu noktasına kadar broker’a gönderilen mesajların broker tarafından sağlıklı şekilde alınıp alınmamasıyla ilgili konuştuk.
Peki şunu düşündük mü :
Publish ettiğim mesaj ya broker’a dahi ulaşamadıysa ne olacak? Yada broker’a ulaşan bir mesajdan ACK dönerken connection düşerse ne yapacağız?
Broker ile olan bağlantımız herhangi bir anda koparsa ve biz o anda mesajlarımızı publish ediyorsak, mesajlarımızın broker’a başarılı ulaşıp ulaşmadığı ile ilgili kesin bir bilgi sahibi olamayız. Çünkü mesajımız henüz broker’a ulaşmadan bağlantı kopmuş olabilir. Yada mesajımız broker’a başarılı şekilde ulaşmış ve publisher’a ACK bildirisi yayınlanmışken bağlantı kopmuş olabilir.
Aklımızda canlandırmak amacıyla alttaki görseli inceleyebiliriz:
Üstteki görselde , broker tarafından alınan mesajın connection failure olması sebebiyle ACK bildirisini iletememesini, dolayısıyla ilgili mesajın hala pending durumunda olmasını ifade etmeye çalıştım.
Peki biz ne yapacağız?
Connection failure durumlarında ‘mesajımı gönderdim ve broker’dan confirm bekliyorum’ gibi senaryolarda pendingMessages’ta bulunan mesajları unknown olarak işaretlemeliyiz. Çünkü bu mesajların en az 1 kere iletilip iletilmediği garanti edilmediği için tekrardan publish edilmesi gerekir.
Unknown olarak işaretlediğimiz mesajları, connection recovery olduğunda tekrardan publish ederek mutlaka 1 kez broker’a pushlanmasını garanti etmeye çalışacağız ( at least one delivery ).
İsterseniz bu yapıyı da koda dökelim.
public enum PublishState
{
Pending,
Unknown
}
Artık pendingMessages’ta sadece ‘Pending’ durumundaki mesajları değil, aynı zamanda ‘Unknown’ olan mesajları tutacağız. bu sebeple bir enum ile bunu ifade edelim.
class PendingPublish {
public required string Body { get; init; }
public PublishState State { get; set; }
}
PendingMessages’ta artık sadece mesajın body’sini değil, aynı zamanda state’ini de tutacağız.
var pendingMessages = new ConcurrentDictionary<ulong, PendingPublish>();
Dictionary’in value değerini artık PendingPublish olarak tutacağız. Mesajların republish edilme işlemini de bu State değerine göre yapacağız.
pendingMessages.TryAdd( sequenceNumber, new PendingPublish {
Body = JsonSerializer.Serialize(message),
State = PublishState.Pending });
Mesajları publish etmeden önce dictionaryde tutuyorduk. Artık PendingPublish tutacağımız için pendingMessage’a kayıt ekleme işlemini de bu şekilde güncelleyelim.
Publisher olarak mesajlarımızı artık state’leri ile birlikte tutuyoruz. Bağlantı koptuğunda mesajlarımızın state’ini ‘unknown’ olarak işaretleyeceğimizden bahsetmiştik.
Peki bunu nasıl yapacağız?
connection.ConnectionShutdownAsync() metoduyla.
connection.ConnectionShutdownAsync += async (_, args) =>
{
Console.WriteLine();
Console.WriteLine("========== CONNECTION SHUTDOWN ==========");
Console.WriteLine($"ReplyCode : {args.ReplyCode}");
Console.WriteLine($"ReplyText : {args.ReplyText}");
Console.WriteLine($"Initiator : {args.Initiator}");
Console.WriteLine($"Cause : {args.Cause?.Message}");
var unknownCount = 0;
foreach (var item in pendingMessages)
{
if (item.Value.State != PublishState.Pending)
continue;
item.Value.State = PublishState.Unknown;
unknownCount++;
Console.WriteLine(
$"[UNKNOWN] Seq={item.Key} | MessageId={item.Value.MessageId}");
}
Console.WriteLine(
$"[CONNECTION LOST] {unknownCount} pending message marked as UNKNOWN.");
Console.WriteLine("=========================================");
Console.WriteLine();
await Task.CompletedTask;
};
.ConnectionShutdownAsync() metodu da , çeşitli davranışlar ile tetikleniyor. Tıpkı ACK-NACK bildirimi gibi. Broker ile olan bağlantımız koptuğunda tetiklenen bu metod ile birlikte mesajlarımızın state’ini unknown olarak işaretleyeceğiz. Böylece recovery durumunda mesajlarımızı tekrardan gönderebilelim.
Buraya kadar çoğu adımımızı tanımladık. Mesajlarımızı gönderdik, confirm bekledik, herhangi bir connection fail durumunda da mesajlarımızın durumunu güncelleyerek tekrardan iletebilir hale geldik.
Peki bu mesajları ne zaman tekrardan iletebileceğiz diye sorarsak ta o anda bize ‘connection.RecoverySucceededAsync() yardımcı olacak.
Not: ConnectionFactory oluşturulurken “AutomaticRecoveryEnabled = true” değeri default olarak true’dur. Bu sebeple recovery olduğunda ilk olarak channel’lar tekrardan oluşturulur. İlgili RecoverySucceededAsync() metodu bu aşamadan sonra çalışır durumda olacaktır.
connection.RecoverySucceededAsync += async (sender, args) =>
{
var unknownMessages = pendingMessages
.Where(x => x.Value.State == PublishState.Unknown)
.ToList();
foreach (var item in unknownMessages)
{
var oldSequenceNumber = item.Key;
var message = item.Value;
pendingMessages.TryRemove(
oldSequenceNumber,
out _);
var newSequenceNumber =
await channel.GetNextPublishSequenceNumberAsync();
message.State = PublishState.Pending;
pendingMessages.TryAdd(
newSequenceNumber,
message);
try
{
await channel.BasicPublishAsync(
exchange: string.Empty,
routingKey: "order-created",
mandatory: true,
body: Encoding.UTF8.GetBytes(message.Body));
Console.WriteLine(
$"[REPUBLISH] OldSeq={oldSequenceNumber} | " +
$"NewSeq={newSequenceNumber}");
}
catch (Exception ex)
{
message.State = PublishState.Unknown;
Console.WriteLine(
$"[REPUBLISH FAILED] Seq={newSequenceNumber} | " +
$"Error={ex.Message}");
}
}
};
Metodu bu şekilde tanımlayabiliriz. Dikkat ederseniz unknown durumundaki mesajların state’i pending olarak değiştirilip publish edilmiyor. Bunun sebebi sequence number’ların channel bazlı tanımlanmasıdır. Örneğin bir channel açıldı ve sequence numaralar 1 den başlayarak dağıtılmaya başlandı. Eğer connection koparsa ilgili channel de kapanır. Dolayısıyla ilgili sequence numberlar da eski bir channeldaki tanımlanan referansı gösterir. Buradaki sequence number’ı güncellemek oldukça önemlidir. Buradaki sequence number’ı güncellemezsek 2 farklı mesaj birbirlerini istemeden ACK edilmesine sebebiyet verebilir.
İsterseniz connection fail durumunda mesajlarımızın unkown olarak işaretlenmesini, ardından yaşanacak recovery ile birlikte de unknown mesajların republish edilmesini simüle edelim.
Broker’ı tekrardan ayağa kaldırdığımızda ne olacak? Hadi bakalım.
Broker tekrardan aktif olduğunda mesajlarımız tekrardan publish edilmiş oldu.
Bu yazımda Publisher Confirms, ACK/NACK, mandatory:true, confirmation tracking ve connection failure durumlarında mesajların nasıl takip edilebileceği ve recovery sürecinde neler yapabileceğimiz ile ilgili öğrendiklerimden bahsetmek istedim.
Bu yazının devamı niteliğinde olacak Part 2’de ise, recovery sürecinin beraberinde getirebileceği Duplicate Message problemine odaklanacağız. Aynı mesajın consumer tarafında birden fazla kez işlenmesinin ne gibi sorunlara yol açabileceğini ve bu problemleri nasıl önleyebileceğimizi inceleyeceğiz.
Yazılarımda henüz yeni öğrendiğim şeylerden bahsediyorum. Dolayısıyla yanlış öğrendiğim/bahsettiğim yerler olabilir, mazur görünüz.
İyi okumalar dilerim.
Comments 0
You cannot comment because you are not logged in.
Log in to comment.