RabbitMQ – Basitçe Kuyruğa Mesaj Gönderme ve Okuma

Merhaba,

Bu içeriğimizde bir uygulama üzerinden kuyruğa mesaj göndermeyi ve başka bir uygulama tarafından bu mesajı okumayı ele alan basit bir RabbitMQ uygulaması üzerine çalışıyor olacak ve böylece RabbitMQ’nun yazılımsal temelleriyle birlikte yapısal prensiplerini de ortaya koyuyor olacağız.

İlk olarak en sade mesaj gönderme senaryosunda kullanacağımız kuyruk yapılanmasını ve temel aktörleri şematik bir şekilde ele alarak başlayalım.

Yukarıdaki şemaya göz atarsanız eğer RabbitMQ’ya dair ilk olarak ele aldığımız makalemizdeki(bknz: RabbitMQ Nedir? Ne Amaca Hizmet Etmektedir?) görselde Exchange diye bir yapılanmanın olduğunu lakin bu yukarıdaki şemada ise olmadığını görmekteyiz. İlerideki makalelerimizde bol bol kendisinden bahsedeceğimiz exchange mekanizması basit bir mesaj gönderme işlemi gibi en temel yapılanmalarda kullanılmasına gerek olmayan bir yapılanmadır. Her ne kadar tarafımızca kullanılmasada sistem tarafından olması gereken bu yapılanmanın bu tarz durumlarda varsayılan olarak tanımlanmış olan default exchange’i devreye girmektedir.

Tekrardan şemaya dönersek eğer en temel işlem olan kuyruğa mesaj gönderme işini Publisher, kuyruktan okuma işini ise Consumer üstlenmektedir. Kuyruğa gönderilen mesajlar First In First Out(FIFO) mantığına göre sıralanmakta ve ilk gönderilen mesaj ilk olarak tüketilmektedir.

Şimdi bu teorik bilgilendirmeden sonra pratik uygulama üzerinden örneklendirmeye geçebiliriz.

Publisher ve Consumer Tasarlama

Yukarıda bahsedildiği gibi en temel senaryoda bile sadece Publisher ve Consumer olmak üzere iki aktörümüz mevcuttur. Bizler bu aktörleri birer Console Application olarak oluşturacak ve tasarlayacağız. Tabi ki de sizler gerçek ihtiyaçlarınıza dönük uygulamalar seçebilir ve içerikte ele alacağımız tüm RabbitMQ nimetlerinden faydalanabilirsiniz.

• Publisher Tasarımı – Kuyruğa Mesaj Gönderme

  

  Publisher, RabbitMQ servisindeki kuyruğa mesaj gönderen uygulamadır.

  .NET Core platformunda RabbitMQ kullanabilmek için öncelikle ilgili projeye RabbitMQ.Client Nuget paketinin yüklenmesi gerekmektedir.
  

  • Bağlantı Oluşturma
    Publisher’ın mesaj gönderebilmesi için herşeyden önce RabbitMQ sunucularına bağlantı oluşturmamız gerekmektedir. Bunun için;

            static void Main(string[] args)
            {
                ConnectionFactory factory = new ConnectionFactory();
                factory.Uri = new Uri("amqp://hkhjerrt:hcqiavAqll6-co4abXnSqUBh_hHifz-Z@hornet.rmq.cloudamqp.com/hkhjerrt");
                //factory.HostName = "localhost";
            }
    

    şeklinde çalışabiliriz. Koda göz atarsanız eğer “ConnectionFactory” sınıfı üzerinden bu bağlantının tanımlandığını görebilirsiniz. Burada 4. satır ile 5. satır arasındaki farka değinmem gerekirse eğer; 4. satır CloudAMQP sunucularında barındırılan RabbitMQ instance’ına bağlantı kurabilmek için tarafımıza verilen AMQP URL değeri iken, 5. satırda ise localdeki RabbitMQ sunucuna bağlanmak içindir.

  • Bağlantı Sağlama ve Kanal Açma
    Oluşturulan bağlantı hattı üzerinden bağlantıyı aktifleştirelim ve ardından bu bağlantı üzerinden kullanacağımız kuyruğu oluşturabilmek için bir kanal açalım.

            static void Main(string[] args)
            {
                ConnectionFactory factory = new ConnectionFactory();
                factory.Uri = new Uri("amqp://hkhjerrt:hcqiavAqll6-co4abXnSqUBh_hHifz-Z@hornet.rmq.cloudamqp.com/hkhjerrt");
                //factory.HostName = "localhost";
    
                using (IConnection connection = factory.CreateConnection())
                using (IModel channel = connection.CreateModel())
                {
    
                }
            }
    

    7. satırda bağlantı sağlanmakta, 8. satırda bir kanal oluşturulmaktadır. Bu işlemden sonra kuyruk oluşturabiliriz.

  • Kuyruk Oluşturma

            static void Main(string[] args)
            {
                ConnectionFactory factory = new ConnectionFactory();
                factory.Uri = new Uri("amqp://hkhjerrt:hcqiavAqll6-co4abXnSqUBh_hHifz-Z@hornet.rmq.cloudamqp.com/hkhjerrt");
                //factory.HostName = "localhost";
    
                using (IConnection connection = factory.CreateConnection())
                using (IModel channel = connection.CreateModel())
                {
                    channel.QueueDeclare("mesajkuyrugu", false, false, true);
                }
            }
    

    Burada 10. satıra göz atarsanız eğer kanal üzerinden QueueDeclare metodu aracılığıyla bir kuyruk oluşturulmuştur. İlgili metodun parametrelerini izah edebilmek için aşağıdaki görseli ele alalım;
    

    • queue : Oluşturulacak kuyruğun adını belirliyoruz.
    • durable : Normal şartlarda kuyruktaki mesajların hepsi bellek üzerinde dizilirler. Hal böyleyken RabbitMQ sunucuları bir sebepten dolayı restart atarlarsa tüm veriler kaybolabilir. durable parametresine true değerini verirsek eğer verilerimiz güvenli bir şekilde sağlamlaştırılacak yani fiziksel hale getirilecektir.
    • exclusive : Oluşturulacak bu kuyruğa birden fazla kanalın bağlanıp, bağlanmayacağını belirtir.
    • autoDelete : True değerine karşılık tüm mesajlar bitince kuyruğu otomatik imha eder.
  • Kuyruğa Mesaj Gönderme

            static void Main(string[] args)
            {
                ConnectionFactory factory = new ConnectionFactory();
                factory.Uri = new Uri("amqp://hkhjerrt:hcqiavAqll6-co4abXnSqUBh_hHifz-Z@hornet.rmq.cloudamqp.com/hkhjerrt");
                //factory.HostName = "localhost";
    
                using (IConnection connection = factory.CreateConnection())
                using (IModel channel = connection.CreateModel())
                {
                    channel.QueueDeclare("mesajkuyrugu", false, false, true);
                    byte[] bytemessage = Encoding.UTF8.GetBytes("sebepsiz boş yere ayrılacaksan");
                    channel.BasicPublish(exchange: "", routingKey: "mesajkuyrugu", body: bytemessage);
                }
            }
    

    11. satırda kuyruğa göndereceğimiz mesajımızı oluşturmuş bulunuyoruz. Dikkat ederseniz mesajımızı byte dizisine çeviriyoruz. Bunun nedeni RabbitMQ’nun byte türünde veriyi kabul etmesidir. Basit bir string ifadeden tutun, kendinizi bile kuyruğa gönderecekseniz byte dizisine çevirmelisiniz. 12. satırda ise “BasicPublish” metodu aracılığıyla açtığımız kanal üzerinden mesajımızı kuyruğa gönderiyoruz. İlgili metodu daha da detaylandırırsak eğer;

    • exchange : Eğer exchange kullanmıyorsanız boş bırakınız. Böylece default exchange devreye girecek ve kullanılmış olacaktır.
    • routingKey : Eğer ki default exchange kullanıyorsanız routingKey olarak oluşturduğunuz kuyruğa verdiğiniz ismin birebir aynısını veriniz.
    • body : Gönderilecek mesajın ta kendisidir.
  • Publisher’ı Ayağa Kaldırma ve Kuyruğa Mesaj Gönderme
    Oluşturduğumuz uygulamayı derleyip çalıştırdığımızda tanımladığımız mesaj kuyruğu oluşturulacak ve içerisine gönderdiğimiz mesaj eklenmiş olacaktır. Bunu görebilmek için tekrar CloudAMQP adresi üzerinden instancelarımızı listeleyelim ve ilgili instance üzerinden aşağıdaki gibi RabbitMQ Manager sekmesini seçelim.
    
    Açılan pencere üzerinden “Queues” sekmesini seçtikten sonra ilgili kuyruğun oluşturulduğunu ve içerisine de bir adet mesaj atıldığını göreceksiniz.
    

  Evet… Mesaj üreticimizi bitirmiş bulunmaktayız. Şimdi sıra tüketiciyi oluşturmaya geldi.

• Consumer(Receiving) Tasarımı – Kuyruktan Mesaj Alma/Okuma

  

  Consumer, RabbitMQ servisinde kuyruktaki mesajları tüketen uygulamadır.

  Consumer uygulamasında da RabbitMQ.Client Nuget paketini kurmalı ve ardından Publisher’da incelediğimiz bağlantı, kanal vs. oluşturma işlemlerini adım adım gerçekleştirmeliyiz.

          static void Main(string[] args)
          {
              ConnectionFactory factory = new ConnectionFactory();
              factory.Uri = new Uri("amqp://hkhjerrt:hcqiavAqll6-co4abXnSqUBh_hHifz-Z@hornet.rmq.cloudamqp.com/hkhjerrt");
  
              using (IConnection connection = factory.CreateConnection())
              using (IModel channel = connection.CreateModel())
              {
                  channel.QueueDeclare("mesajkuyrugu", false, false, false);
                  EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
                  channel.BasicConsume("mesajkuyrugu", false, consumer);
                  consumer.Received += (sender, e) =>
                  {
                      //e.Body : Kuyruktaki mesajı verir.
                      Console.WriteLine(Encoding.UTF8.GetString(e.Body));
                  };
              }
              Console.Read();
          }
  

  Her ne kadar cunsomer kuyruktaki mesajları tüketecek olan bir uygulama olsa dahi 9. satırdaki “QueueDeclare” metodu ile publisher’da ki değerlerle birebir aynı olacak şekilde kuyruk tanımlamalıyız. 10. satırda ise tanımladığımız kuyruktaki mesajları yakalayacak bir event oluşturuyor ve ardından 11. satırda “BasicConsume” metodu ile ilgili mesajları tüketiyoruz. “BasicConsume” metodunun parametrelerini incelersek eğer;

  • queue : Mesajların alınacağı kuyruk adı.
  • autoAck : Kuruktan alınan mesajın silinip silinmemesini sağlıyor. Bazen kuyruktan alınan mesaj işlenirken beklenmeyen hatalarla karşılaşılabiliyor. O yüzden mesajı başarılı bir şekilde işlemeksizin kuyruktan silinmesini pek önermeyiz.
  • consumer : Tüketici.

  12. satıra göz atarsanız eğer “EventingBasicConsumer” tipinden tanımladığımız consumer nesnesinin Received olayı bizlere kuyruktaki mesajları getirecektir. Dikkat ederseniz mesaj byte dizisi olarak gönderildiğinden dolayı haliyle byte dizisi olarak elde edilecek ve biz ilgili dönüşümü yapmak suretiyle mesajı işlemekteyiz.

Derleyip, Test Edelim

Her iki projeyide derleyip sırasıyla Publisher ve Consumer şeklinde çalıştırdığımızda cloud’da ki bağlantı kurulan sunucuda kuyruğa atılan mesajların consumer tarafından tüketildiğini ve tek tek ekrana yazıldığını göreceksiniz.

Evet… Böylece basit bir mesaj kuyruk uygulaması yapmış bulunmaktayız. Tabi ki de sonraki içeriklerimizde olayı daha da detaylandıracak ve derinlerde anlamlı sistemler oluşturacağız.

İlgilenenlerin faydalanması dileğiyle…
Sonraki yazılarımda görüşmek üzere…
İyi çalışmalar…