RabbitMQ – Topic Exchange

Merhaba,

Bu içeriğimizde RabbitMQ’da çalışma mantığı belirli bir formata dayalı sistem üzerine oturtulmuş Topic Exchange’i inceleyeceğiz.

Topic Exchange İşleyişi Nasıldır?

Topic Exchange’de atılan mesajların routing key değeri .(nokta) operatörü kullanılarak formatlandırılmakta ve bu formattaki isimlerde yapılan filtrelemelere göre uygun düşen kuyruklara mesajlar gönderilmektedir. Yukarıdaki şemaya göz atarsak eğer;

Message Routing Key
‘usa.news’ ‘usa.weather’ ‘europe.news’ ‘europe.weather’	Yandaki routing key değerlerine sahip olan mesajları hedef kuyruğa gönderebilmek için oluşturduğumuz kuyruklarda aşağıdakine benzer bir filtreleme yapılması yeterlidir.
Topic Filtrelemesi ‘usa.news’ Routing key değeri ‘usa.news’ olan mesajlar bu kuyruğa girecektir. ‘*.weather’ ‘*<önemli değil>.weather’ olan mesajlar bu kuyruğa girecektir. ‘#.news’ ‘#< başı önemli değil>.news’ olan mesajlar bu kuyruğa girecektir. ‘*.weather’ ‘*<önemli değil>.weather’ olan mesajlar bu kuyruğa girecektir. ‘usa.#’ ‘usa.#< sonu önemli değil>’ olan mesajlar bu kuyruğa girecektir. ‘europe.*’ ‘europe.*<önemli değil>’ olan mesajlar bu kuyruğa girecektir. ‘#’ ‘#< hiçbiri önemli değil>’ olan mesajlar bu kuyruğa girecektir.

Kullanımı

Bizler örnek olarak silahlı kuvvetler hiyerarşisine uygun bir şekilde görevlendirme senaryosu benimseyip mesajlarımızı ilgili rütbelere özel görev barındıran kuyruklara ileteceğiz.

Publisher

        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.ExchangeDeclare("topicexchange", type: ExchangeType.Topic);
                for (int i = 1; i <= 100; i++)
                {
                    byte[] bytemessage = Encoding.UTF8.GetBytes($"{i}. görev verildi.");

                    IBasicProperties properties = channel.CreateBasicProperties();
                    properties.Persistent = true;
                    channel.BasicPublish(exchange: "topicexchange", routingKey: $"Asker.Subay.{(i % 2 == 0 ? "Yuzbasi" : (i % 11 == 0 ? "Binbasi" : "Tegmen"))}", basicProperties: properties, body: bytemessage);
                }
            }
        }

Yukarıdaki örnek publisher kod bloğunu incelerseniz eğer 16. satırda Topic exchange’e uygun formatta “Asker.Subay.[Tegmen – Yuzbasi – Binbasi]” routing keyi oluşturulmaktadır. Burada senaryo gereği oluşturulan 100 adet görevin her birini ‘Teğmen’, yarısını ‘Yüzbaşı’ ve hatrı sayılır azınlıktaki kadarını ‘Binbaşı’ rütbeli askerlerin üstleneceği şekilde bir algoritma oluşturulmuş bulunmaktadır ve her bir görev mesaj olarak ilgili rütbeyi temsil eden formattaki routing key değerine sahip kuyruğa gönderilmektedir. Şimdi sıra kendi rütbesine uygun görevleri barındıran kuyruğa subscribe olan consumerları tanımlayama gelmiş bulunmaktadır.

Consumer

        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.ExchangeDeclare("topicexchange", type: ExchangeType.Topic);

                string queueName = channel.QueueDeclare().QueueName;
                string routingKey = "";

                routingKey = args[0] switch
                {
                    "1" => $"*.*.Tegmen",
                    "2" => $"*.#.Yuzbasi",
                    "3" => $"#.Binbasi.#",
                    "4" => $"Asker.Subay.Tegmen",
                };

                channel.QueueBind(queue: queueName, exchange: "topicexchange", routingKey: routingKey);

                channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);

                EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
                channel.BasicConsume(queueName, false, consumer);
                consumer.Received += (sender, e) =>
                {
                    Console.WriteLine($"{routingKey} {Encoding.UTF8.GetString(e.Body)} görevi aldı.");
                    channel.BasicAck(e.DeliveryTag, false);
                };
                Console.Read();
            }
        }

Yukarıdaki consumer örnek kodlarını incelerseniz eğer 14. satırda console uygulamasına verilen argüman değerine göre Topic formatına uygun bir nitelendirme yapılmakta ve buradan elde edilen routing key 22. satırda oluşturulacak kuyruğa verilmektedir. Böylece belirtilen formatta uyan her bir kuyruğa uygun mesajları consumer elde ederek tüketmektedir.

Test Edelim

Uygulamayı derleyip çalıştırdığımızda hornet.rmq.cloudamqp.com cloud adresine gider ve göz atarsak eğer Topic exchange’in oluşturulduğunu görmüş olacağız.

Görüldüğü üzere Topic exchange ile belli bir formatta girdiğimiz routing key değerlerine göre mesajlarımız kuyruklara yerleştiriliyor ve ardından ‘#’ yahut ‘*’ operatörüleri ile yapılan filtreleme neticesinde uygun kuyruğa bağlanan consumerlar tarafından tüketiliyor. Gayet güzel 🙂

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

Not : Örnek uygulamayı indirmek için buraya tıklayınız.