RabbitMQ Routings, Message Patterns

rabbitmq

RabbitMQ, dağıtık sistemler için çeşitli asenkron yapılar kullanır. Bu yazımızda hem routing ile ilgili detaylı konulara (Exchange to Exchange Binding, Alternate Exchange) hem de bu message pattern’lere (Work Queues, Round-Robin, Publish-Subscribe, Push-Pull, Request-Reply) göz atıyor olacağız.

Bu yazıda ilerlemeden önce RabbitMQ konusuna giriş yapmak adına RabbitMQ Nedir yazıma göz atabilirsiniz.

Exchange to Exchange Binding

Bir queue’yu nasıl bir exchange’e bind ediyorsak bir exchange’i de başka bir exchange’e bind edebiliriz, yönlendirme kuralları aynıdır. Exchange üzerinden çıkan mesaj, hedef exchange’e binding ayarlarına göre yönlendirilir ve hedef exchange bu mesajı bağlı queue’lere iletir.

RabbitMQ Exchange to Exchange Binding

Yukarıdaki örneği incelediğinizde route.rabbit route keyine sahip exchange’in diğer exchange’e bind edildiğini göreceksiniz. Bu route keye sahip bir mesaj bir exchange’ten diğerine yönlendirilerek eşleşen queue’ye yönlendirilir.

Producer projemizi oluşturup, RabbitMQ.Client isimli Nuget paketini kuralım.

Producer kodumuzu incelediğimizde önceki kodlardan farklı olarak ExchangeBind methoduyla ex.rabbit‘in ex.dog‘a route.rabbit route keyiyle birlikte bind edildiğini göreceksiniz. ex.dog exchange’i üzerinden ilettiğimiz route.rabbit route keyine sahip mesajın ex.rabbit exchange’ine ve oradan queue-rabbit queue’sine aktarıldığını göreceksiniz. Bu mesajı RabbitMQ Management UI üzerinden consume edebilirsiniz.

Alternate Exchange

Dağıtık sistemler üzerinde kurulan asenkron bir yapıda bir exchange’e yönlendirilen bazı mesajlar bağlı queue’lara yönlendirilmek üzere uygun olmayabilir. Bunlar unrouted message olarak adlandırılmaktadır. Bu mesajlar exchange tarafından yok sayılır ve mesaj kaybolur. İşte bu durumlarda bu mesajları toplamak için Alternate Exchange kullanılabilir. Bir alternate exchange tanımlamak için -biraz sonra göreceğimiz gibi- exchange’e alternate-exchange keyi ve hangi exchange’in alternate exchange olarak kullanılacağı değer eklenmelidir, böylece unrouted mesajlar bu exchange’e yönlendirilecektir. Fanout Exchange bir filtreleme işlemi yapmadığından dolayı bu yapıya gayet uygundur.

RabbitMQ Alternate Exchange
Yukarıdaki örnekte de göreceğiniz üzere Exchange1’e yönlendirilen mesaj bağlı queue’lar ile eşleşmeyeceğinden dolayı fanout tipindeki alternate exchange olarak tanımlanmış Exchange2’ye yönlendirilecektir.

Producer kodumuzu yazıp gerekli paketi kuralım.

Exchange ex.fanout‘un ex.direct‘e alternate-exchange olarak atandığına dikkat edelim, böylece publish edilen text mesajı fanout exchange yönlendirilerek bağlı tüm queue’lara aktarılacaktır.

Push & Pull

Bir queue üzerinden mesajları consume etmek için iki yol mevcuttur, bunlar push ve pull’dur.

Şimdiye kadar yaptığımız örnekler tavsiye edilen yol olmakla birlikte push modelini kullanıyordu. Consumer uygulaması bir queue’ya subscribe olur ve mesajları dinler. Kuyrukta bekleyen mesajlarla birlikte kuyruğa yeni eklenen mesajlar doğrudan consumer uygulamasına push edilerek gönderilir.

Pull modelindeyse, consumer uygulaması queue’ya subscribe olmaz. Bunun yerine kuyruğu periyodik olarak kontrol ederek yeni mesajlara bakar. Kuyrukta yeni bir mesaj olması durumunda consumer uygulaması manuel fetch işlemi gerçekleştirir. Bu model her ne kadar tavsiye edilmese de Message Broker ile consumer uygulaması arasında doğrudan bir bağlantı olmadığı durumlarda izlenecek tek yoldur.

Örnek consumer projemizi oluşturup RabbitMQ.Client isimli Nuget paketini kuralım.

Bağlantımızı yaparak, bir channel oluşturuyoruz. Push modelini kullanan consumer kodlarına bakalım.

Daha önceden aşina olduğumuz yapı, farklı olarak BasicCancel methodu yardımıyla subscribe olduğumuz queue’dan unsubscribe oluyoruz.

Pull modelini kullanan consumer kodlarına göz atalım.

5 saniye aralıklarla BasicGet methoduyla queue üzerinden tekil bir mesaj alınıyor, bir sonuç olmaması durumunda null dönüyor. Sonuç null değilse mesaj konsola basılıyor.

Work Queues

Work Queues, yoğun iş gücü gerektiren durumlarda görevlerin birden fazla worker arasında dağıtmasında kullanılır. Producer kuyruğa yeni bir görev ekler ve bu görevler worker uygulamaları arasında dağıtılır. Görevlerin dağıtılmasında pull veya push modeli kullanılabilir. Push modelinde Message Broker kuyruktaki mesajları çalışan worker’lara otomatik olarak dağıtır. Pull modelindeyse worker, yeni bir görev için hazır hale geldiğinde kuyruktan yeni bir mesaj alır. Work Queues aynı zamanda Competing Consumer Pattern olarak da anılır.

Round-robin

Task Queue kullanmanın avantajlarından bir tanesi görevlerin paralel şekilde worker’lara dağıtılmasıdır. Aynı zamanda sistemin yetersiz kaldığı durumlarda daha fazla worker dahil edilmesiyle ölçeklendirmenin de yapılabilmesidir. Message Broker default olarak her mesajı bir sonraki consumer’a gönderir, böylece her consumer aynı sayıda mesaj consume etmiş olur. Bu çoğu zaman güzel olsa da bazı senaryolarda sorunlara yol açacaktır. Bu soruna değinmeden önce bir kavrama göz atmamız gerekiyor.

Message Acknowledgment

Bir mesaj yönlendirildiğinde, yönlendirilen mesaj işlenmeden consumer ölürse bu mesajın kaybolmasına yol açacaktır. Çünkü gönderilen mesaj iletildiğinde silinmek üzere ayarlanmıştır. Bu senaryoda yalnızca mevcut mesaj değil, işlenmek üzere gönderilmiş diğer mesajların da kaybolmasına yol açar.

Bu durumun önüne geçmek için (bir mesajın asla kaybolmaması) mesaj bildirimlerini manuel olarak yönetmemiz gerekir. Bu işlem, autoAck property’si ile yapılır ve alınan mesajın başarıyla işlenmesi durumunda consumer’ın Message Broker’a bunu bildirmesiyle yapılır, böylece mesaj silinebilir. Bu senaryoda bir görev tamamlanamadan consumer’in ölmesi durumunda Message Broker bu durumu anlayarak mesajı yeniden kuyruğa atıp başka consumer’a -varsa- iletecektir. Bir mesaj alındıktan sonra default timeout 30 dakikadır, bu zaman değiştirilebilir. (Delivery Acknowledgement Timeout)

Round-robin‘e geri dönecek olursak, kuyrukta iki çeşit görevimiz olduğunu varsayalım, bunlardan bir tanesi uzun zaman alıyor, bir diğeri de çok kısa sürüyor. Message Broker ilgili mesajları sırayla dağıttığı için bir consumer üzerinde daha fazla iş yükü oluşacak ve bu kuyruktaki mesajların tüketilmesi bir diğer consumer’a göre daha uzun sürecek. Bu davranışı BasicQos methoduyla değiştirebiliriz. prefetchCount = 1 olarak ayarlanarak worker’a aynı anda birden fazla görev vermemesi istenir. İşte bu durumda da Message Ack‘leri devreye giriyor, bir işin tamamlandığını BasicAck methoduyla bildiriyoruz. Aynı zamanda reddedilmek üzere BasicNack ve BasicReject methodları da mevcut.

Consumer uygulaması kodlarına yakından bakalım, senaryoya göre çalışan worker’a bir isim veriyor ve gelen mesaj -saniye cinsinden- kadar uygulamayı Thread.Sleep ile uyutuyoruz.

Bağlantı oluşturulup bir kanal açıldıktan sonra BasicQos ile aynı anda bir görev alacağımızı bildiriyoruz. BasicConsume methodunda autoAck‘i false olarak işaretleyip, acknowledgment’i manuel yöneteceğimizi bildiriyoruz. Aynı şekilde mesaj Received olup, işlendikten sonra BasicAck ile başarılı sonuç dönüyoruz.

Publish & Subscribe

Bu pattern tüm abonelere aynı mesajı iletmek için kullanılır. Her subscriber için ayrı bir queue vardır, mesajlar bu queue’lardan her birine ulaştırılır, böylece her subscriber aynı mesajın kopyasını almış olur.

Publish – Subscribe pattern’i genellikle olay bildirimlerinde kullanılır. Örneğin; bir sipariş servisinde, sipariş oluşturuldukça oluşturulan mesaj dahilinde hem loglama hem de e-posta servisi kendi kuyruğundan bu sipariş ile ilgili işlemleri yapacaktır.

Publisher ve Subscriber projelerini oluşturarak, RabbitMQ.Client paketini kuralım.

Publisher uygulamasında iki adet queue oluşturarak bir fanout exchange’ine bind edildi ve sürekli mesaj gönderimi için döngü içerisine alındı.

Consumer uygulamasını iki terminal üzerinden çalıştıracağız, sırasıyla her biri için queue isimlerini (queue.one, queue.two) girerek subscribe olacağız. Publisher uygulamasından mesaj ürettikçe subscriber uygulamalarının mesaj kopyalarını aldığını göreceğiz.

Request & Reply

Bir messaging altyapısı kullanıldığında iletişim producer’dan consumer’a doğru tek yönlüdür, uygulama çift yönlü iletişim ile gönderdiği mesaja bir sonuç bekleyebilir. Request-Reply pattern bu durumda devreye girmektedir, sistemde biri request diğeri response olmak üzere en az iki adet queue tanımlanmalıdır. Bunun için RabbitMQ ile biri client diğeri RPC sunucu olan bir RPC sistemi oluşturacağız.

Replier projemizi oluşturup RabbitMQ.Client paketini kuralım.

Requester projesini oluşturup, gerekli paketi kuralım.

Requester projesi ayağa kalktığında anonim bir callback queue oluşturuyor. Bir mesaj rpc-queue‘ye gönderilirken mesaj yanında iki özellik daha ekleniyor. Bunlar ReplyTo (callback queue için) ve CorelationId (her request için bir guid). Replier projesi bu queue üzerinden bir istek geldiğinde gerekli işlemleri yaparak ReplyTo property’sindeki queue’ya göre sonucu Requester’a döner. Requester callback queue üzerinden gelen mesajları CorelationId‘ye bakarak karşılaştırmaktadır.

Channel & Queue & Message Priorities

Kuyruğa gönderilen her mesaj aynı aciliyet veya önem durumuna sahip olmayabilir. Bazı mesajların daha önce işlenmesi gerekirken bazılarının ise sonra işlenmesinde bir sorun olmayabilir. Örneğin; bir e-ticaret platformunda siparişlerin işlenmesiyle, kampanya bildirimleri aynı öneme sahip değildir.

Bir channel ve queue için öncelik x-max-priority keyi ile belirlenir. 0-255 arasında değer alabilir. 0, bir öncelik olmayacağını belirtirken 1 ila 10 arasındaki değerler önerilir.

Bir mesajın önceliği IBasicProperties.Priority üzerinden belirtilir, mesajların düzgünce sıralanabilmesi için kuyrukta bir süre kalması gerekmektedir.

Producer ve Consumer projelerini oluşturup gerekli paketleri kurup kodumuzu yazalım.

Consumer projesini yazalım.

Kaynak dosyasına buradan ulaşabilirsiniz.

You may also like...

Bir cevap yazın

E-posta hesabınız yayımlanmayacak.