C# ile RabbitMQ Client’ı kullanarak Publisher ve Consumer Yapısı

Merhaba arkadaşlar.

Bir önceki makalem olan “RabbitMQ Nedir ve Windows’a Kurulumu” isimli makale ile, RabbitMQ konusuna bir giriş yapmış idik. Bu makale kapsamında ise RabbitMQ’nun C# provider’ını kullanarak en sık kullanılan Messaging Pattern‘lerinden birisi olan “Publish / Subscribe” pattern’i mantığında, basit bir şekilde nasıl mesaj gönderilir ve alınıra bakıyor olacağız.

RabbitMQ’nun Procuder, Queue ve Consumer yapısını basit olarak hatırlamak gerekirse;

• Producer: Queue’ya mesaj gönderen uygulamadır. Yani Publisher’ımız.
• Consumer: Queue’daki mesajları dinleyecek olan uygulamamızdır.

Genel bilgileri tekrardan hatırladığımıza göre şimdi kodlamaya başlayabiliriz. Örnek olarak bir Console Application oluşturacağız ve Queue’ya mesaj gönderecek olan Publisher’ı ve bu mesajları dinleyecek olan Consumer yapısını oluşturacağız.

Öncelikle Nuget Package Manager üzerinden aşağıda görmüş olduğumuz “RabbitMQ.Client” paketini kuralım.

Kurmuş olduğumuz bu client paketi sayesinde RabbitMQ ile haberleşebileceğiz.

RabbitMQ’nun C# Client provider’ı connection işlemlerini “ConnectionFactory” üzerinden yaratarak gerçekleştirmektedir. Bu connection’a hem “Publisher” hemde “Consumer” içerisinde aynı şekilde ihtiyacımız olacağı için, “RabbitMQService” isminde bir class ekleyelim ve burada “ConnectionFactory” class’ını kullanarak ortak işlemleri kodlamaya başlayalım.

using RabbitMQ.Client;

namespace RabbitMQSample
{
    public class RabbitMQService
    {
        // localhost üzerinde kurulu olduğu için host adresi olarak bunu kullanıyorum.
        private readonly string _hostName = "localhost";

        public IConnection GetRabbitMQConnection()
        {
            ConnectionFactory connectionFactory = new ConnectionFactory()
            {
                // RabbitMQ'nun bağlantı kuracağı host'u tanımlıyoruz. Herhangi bir güvenlik önlemi koymak istersek, Management ekranından password adımlarını tanımlayıp factory içerisindeki "UserName" ve "Password" property'lerini set etmemiz yeterlidir.
                HostName = _hostName
            };

            return connectionFactory.CreateConnection();
        }
    }
}

RabbitMQService class’ımız hazır durumda. Connection işlemleri “IConnection” interface’inden türemektedir ve factory class’ı üzerinden “HostName” property’sini set ederek, “CreateConnection()” method’u aracılığı ile yeni bir connection oluşturulabilmektedir.

Şimdi “Publisher” görevini görecek olan Publisher class’ını kodlamaya geçelim.

using System;
using System.Text;

namespace RabbitMQSample
{
    public class Publisher
    {
        private readonly RabbitMQService _rabbitMQService;

        public Publisher(string queueName, string message)
        {
            _rabbitMQService = new RabbitMQService();

            using (var connection = _rabbitMQService.GetRabbitMQConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    channel.QueueDeclare(queueName, false, false, false, null);

                    channel.BasicPublish("", queueName, null, Encoding.UTF8.GetBytes(message));

                    Console.WriteLine("{0} queue'su üzerine, \"{1}\" mesajı yazıldı.", queueName, message);
                }
            }
        }
    }
}

Şimdi neler yaptığımıza adım adım bir bakalım.

var channel = connection.CreateModel()

Yukarıdaki satır ile açmış olduğumuz connection üzerinden “CreateModel” method’unu çağırarak, RabbitMQ üzerinde yeni bir channel/session yaratmaktayız. Bu Channel sayesinde bir Queue oluşturabilirken, mesaj gönderme işlemlerini de gerçekleştirebilmekteyiz.

channel.QueueDeclare(queueName, false, false, false, null);

Bu satırda ise method isminden de anlaşılabileceği üzere, yeni bir queue tanımlıyoruz. Burada önemli olan ilk üç parametresine bakmak gerekirse;

• queue: Oluşturulacak olan Queue’nun ismi.
• durable: Bu parametre ile in-memory olarak çalışan Queue disk üzerinden çalışmaya başlayacaktır. Bu sayede RabbitMQ servisi dursa bile Queue kaybolmayacaktır. Her güzelliğin getirdiği bir kötü tarafın olduğu gibi bununda beraberinde getireceği latency problemi bulunmaktadır haliyle.

channel.BasicPublish("", queueName, null, Encoding.UTF8.GetBytes(message));

Artık Channel ve Queue hazır bir durumda olduğu için, “BasicPublish” method’u ile kolay bir şekilde oluşturmuş olduğumuz ilgili Queue’ya mesaj gönderiyoruz.

“BasicPublish” method’un kullandığımız parametrelerine bakarsak:

• exchange: Bu parametreyi es geçiyoruz. Exchange genel olarak mesajı ilgili Routing Key’e göre ilgili Queue’ya yönlendiren bölümdür. Direct Exchange, Fanout Exchange ve Topic Exchange gibi tipleri bulunmaktadır. Bunları bir sonraki makalemde detaylı olarak ele alacağım.
• routingKey: Burada girmiş olduğumuz key’e göre ilgili Queue’ya yönlendirilecektir mesaj.
• body: Queue’ya göndermek istediğimiz mesajı byte[] tipinde gönderiyoruz.

Publisher’ı test edebilmek için “Program.cs” class’ını açalım ve aşağıdaki gibi kodlayalım.

namespace RabbitMQSample
{
    class Program
    {
        private static readonly string _queueName = "GOKHANGOKALP";
        private static Publisher _publisher;

        static void Main(string[] args)
        {
            _publisher = new Publisher(_queueName, "Hello RabbitMQ World!");

             Console.ReadKey();
        }
    }
}

Queue ismini “GOKHANGOKALP” olarak belirleyelim ve constructor üzerinden Queue ismini ve göndermek istediğimiz mesajı set edelim. Uygulamayı çalıştırmadan önce RabbitMQ Management ekranı üzerinden Queues tab’ına bir bakalım. Management ekranına “localhost:15672” adresinden erişebilirsiniz.

Yukarıda gördüğümüz gibi şuan herhangi bir Queue bulunmamaktadır. Şimdi oluşturmuş olduğumuz Console uygulamasını çalıştıralım.

Publisher Console uygulaması üzerinden başarılı bir şekilde çalıştı ve “GOKHANGOKALP” isminde bir Queue oluşturarak “Hello RabbitMQ World!” mesajını Queue’ya ekledi. Management ekranını bir de şimdi kontrol edelim.

Tada… “GOKHANGOKALP” isminde bir Queue oluşmuş ve “Read” state’inde bir mesajın bulunduğunu söylüyor. Şimdi sıra geldi bu mesajları alacak yani Subscribe edecek olan Consumer’ı kodlamaya. “Consumer” isminde yeni bir class oluşturarak, aşağıdaki gibi kodlamaya başlayalım.

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

namespace RabbitMQSample
{
    public class Consumer
    {
        private readonly RabbitMQService _rabbitMQService;

        public Consumer(string queueName)
        {
            _rabbitMQService = new RabbitMQService();

            using (var connection = _rabbitMQService.GetRabbitMQConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    var consumer = new EventingBasicConsumer(channel);
                    // Received event'i sürekli listen modunda olacaktır.
                    consumer.Received += (model, ea) =>
                    {
                        var body = ea.Body;
                        var message = Encoding.UTF8.GetString(body);

                        Console.WriteLine("{0} isimli queue üzerinden gelen mesaj: \"{1}\"", queueName, message);
                    };

                    channel.BasicConsume(queueName, true, consumer);
                    Console.ReadLine();
                }
            }
        }
    }
}

Consumer kısmında adım adım ne yaptığımızı inceleyelim şimdi.

var consumer = new EventingBasicConsumer(channel);

Asıl işlemlerimizi gerçekleştirecek oldıuğumuz “consumer” nesnemizdir. “EventingBasicConsumer” class’ı sayesinde constructor üzerinden channel verildiğinde, “Recevied” event’i sayesinde sürekli “listening” modunda olacaktır. Queue’daki ilgili mesajları sırasıyla almaktadır ve “Body” property’sinde barındırmaktadır. Hatırlarsak mesajları “byte[]” tipinde göndermiştik. Göndermiş olduğumuz mesajın “string” tipinde olduğunu bildiğimiz için “Encoding” altında bulunan “GetString” method’u ile “Decode” işlemini gerçekleştiriyoruz.

NOT: RabbitMQ Queue’su FIFO(First in Frist out) mantığında çalışmaktadır.

channel.BasicConsume(queueName, true, consumer);

Burada ise method isminden yine anlaşılabileceği üzere (isimlendirmenin ne kadar önemli olduğu buradan da anlaşılabilir) basic bir şekilde verilmiş olan Queue ismine göre mesajları alma işlemini başlatıyoruz. Burada dikkat çekmek istediğim bir kaç parametre bulunmaktadır:

• queue: Hangi Queue’nun mesajları alınacak ise.
• noAck: True olarak set edildiği taktirde, consumer mesajı aldığı zaman otomatik olarak mesaj Queue’dan silinecektir. Eğer Queue üzerinden silinmesini istemiyor iseniz, False olarak set etmeniz gerekmektedir.

Console uygulamasının “Program.cs” class’ını şimdi aşağıdaki gibi güncelleyelim ve tekrardan çalıştıralm.

namespace RabbitMQSample
{
    class Program
    {
        private static readonly string _queueName = "GOKHANGOKALP";
        private static Publisher _publisher;
        private static Consumer _consumer;

        static void Main(string[] args)
        {
            _publisher = new Publisher(_queueName, "Hello RabbitMQ World!");

            _consumer = new Consumer(_queueName);
        }
    }
}

Publisher’ı aynı şekilde başlatarak tek fark olarak Consumer’ı tanımlıyor ve aynı “_queueName” parametresini geçerek, aynı Queue üzerindeki mesajları dinlemesini söylüyoruz.

Bunun sonucunda Console uygulamasının çıktısı aşağıdaki gibi olacaktır.

Gördüğümüz gibi sonuç olarak Consumer başarılı bir şekilde “GOKHANGOKALP” Queue’sunu dinleyerek, içerisinde bulunan mesajı çekip ekrana basmıştır. “noAck” parametresini True olarak set etmiş idik. Şimdi Management ekranını bir kontrol edelim bakalım Queue üzerinde herhangi bir mesaj var mı?

Gördüğümüz gibi “GOKHANGOKALP” Queue’su olduğu gibi durmaktadır fakat herhangi bir mesaj içerisinde bulunmamaktadır. Şimdi “noAck” parametresini False olarak set edip, tekrardan Console uygulamasını çalıştıralım ve sonuca tekrardan Management ekranı üzerinden bir bakalım.

channel.BasicConsume(queueName, false, consumer);

“noAck” parametre değeri False olduğundan dolayı Consumer mesajı çekmesine rağmen, mesaj Queue üzerinde “Ready” state’inde durmaktadır. Queue üzerindeki mesajları silme gibi kararları, business rule’larımıza bağlı olarak bizlerin ele alması daha doğru bir karar olacaktır.

Bu makalemde basic bir şekilde RabbitMQ’nun C# Client’ını kullanarak, mesaj işlemlerini nasıl handle edebileceğimizi hep beraber inceledik. Bir sonraki RabbitMQ serimde görüşmek dileğiyle.

Sağlıcakla kalın.

RabbitMQSample