RabbitMQ – Fanout Exchange

Yazar evrenaktas

Selamlar,bu makalede Fanout Exchange yapısı üzerine konuşuyor olacağız.Temel bir girişin ardından örnek bir uygualama ile konuyu kavramaya çalışacağız.

Exchange yapısı gönderilen mesajların karşılanıp ilgili kuyruklara, belirlenen kurallar çervesinde dağıtımını gerçekleştiren yapıdır. RabbitMQ Nedir? makalemizde gerçekleştirdiğimiz örnekte herhangi bir exchange yapısı kullanmadan direk olarak Queue bağlanıp mesajımızı göndermiş ve okumuştuk. Şimdi ise gönderilen mesajların ilgili kuyruklara dağıtımı için araya bir exchange yapısı ekliyor olacağız.

Fanout Exchange yapısı gelen mesajların alınıp ayırım yapmadan tüm kuyruklara eşit bir şekilde iletilmesini sağlayan yapıdır. Producer mesajı gönderir Fanout Exchange mesajı karşılar ve tüm kuyruklara dağıtımını gerçekleştirir. Geriye kalan ise Consumer‘ın mesajları okuyup işlemesidir.

Şimdi ise örnek uygulamamız üzerinden Fanout Exchange yapısını daha iyi anlamaya çalışalım.

Producer

public class ProducerService
    {
        public void Producer()
        {
            try
            {
                var message = "Bu bir fanout exchange kullanım örneğidir";
                var factory = new ConnectionFactory();
                factory.HostName = "localhost";
                using (var connection = factory.CreateConnection())
                {
                    using (var channel = connection.CreateModel())
                    {
                        channel.ExchangeDeclare(exchange: "logs", durable: true, type: ExchangeType.Fanout);
                        var propertis = channel.CreateBasicProperties();
                        propertis.Persistent = true;
                        var body = Encoding.UTF8.GetBytes(message);
                        channel.BasicPublish("logs", routingKey: "", propertis, body: body);
                        Console.WriteLine("Mesaj Fanout Exchange'e gönderilmiştir");
                        Console.ReadLine();
                    }
                }
            }
            catch (Exception exception)
            {

                throw new Exception(exception.Message);
            }
        }
    }

public class ProducerService

{

public void Producer()

{

try

{

var message = "Bu bir fanout exchange kullanım örneğidir";

var factory = new ConnectionFactory();

factory.HostName = "localhost";

using (var connection = factory.CreateConnection())

{

using (var channel = connection.CreateModel())

{

channel.ExchangeDeclare(exchange: "logs", durable: true, type: ExchangeType.Fanout);

var propertis = channel.CreateBasicProperties();

propertis.Persistent = true;

var body = Encoding.UTF8.GetBytes(message);

channel.BasicPublish("logs", routingKey: "", propertis, body: body);

Console.WriteLine("Mesaj Fanout Exchange'e gönderilmiştir");

Console.ReadLine();

}

catch (Exception exception)

{

throw new Exception(exception.Message);

}

Kodlarımızı inceleyecek olursak ConnectionFactory sınıfından bir nesne yaratılır. Yaratılan bu sınıf üzerinden HostName set edilir. ConnectionFactory sınıfı bizim RabbitMQ hostuna bağlanmamıza sağlayacak olan sınıftır. Daha sonra CreateConnection()methodu ile bağlantı gerçekleştirilir.

CreateModel() methodu ile yeni bir channel oluşturulur.Channel üzerinden ExchangeDeclare() methodu ile yeni bir exchange oluşturulur. Son olarak ilgili channel kullanılarak BasicPublish() methodu ile mesaj exchang’e gönderilir.

Consumer

public class ConsumerService
    {
        public void Consumer()
        {
			try
			{
                var factory = new ConnectionFactory();
                factory.HostName = "localhost";
                using (var connection = factory.CreateConnection())
                {
                    using (var channel = connection.CreateModel())
                    {
                        channel.ExchangeDeclare(exchange: "logs", durable: true, type: ExchangeType.Fanout);

                        var queueName = channel.QueueDeclare().QueueName;
                        channel.QueueBind(queue: queueName, exchange: "logs", routingKey: "");
                        channel.BasicQos(0, 1, false);
                        var consumer = new EventingBasicConsumer(channel);
                        channel.BasicConsume(queueName, false, consumer: consumer);
                        consumer.Received += (render, argument) =>
                        {
                            string message = Encoding.UTF8.GetString(argument.Body.ToArray());
                            channel.BasicAck(deliveryTag: argument.DeliveryTag, false);
                            Console.WriteLine(message);
                        };
                        Console.ReadLine();
                    }
                }
            }
			catch (Exception exception)
			{

				throw new Exception(exception.Message);
			}
        }
    }

public class ConsumerService

{

public void Consumer()

{

try

{

var factory = new ConnectionFactory();

factory.HostName = "localhost";

using (var connection = factory.CreateConnection())

{

using (var channel = connection.CreateModel())

{

channel.ExchangeDeclare(exchange: "logs", durable: true, type: ExchangeType.Fanout);

var queueName = channel.QueueDeclare().QueueName;

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

channel.BasicQos(0, 1, false);

var consumer = new EventingBasicConsumer(channel);

channel.BasicConsume(queueName, false, consumer: consumer);

consumer.Received += (render, argument) =>

{

string message = Encoding.UTF8.GetString(argument.Body.ToArray());

channel.BasicAck(deliveryTag: argument.DeliveryTag, false);

Console.WriteLine(message);

};

Console.ReadLine();

}

catch (Exception exception)

{

throw new Exception(exception.Message);

}

QueueDeclare() methodu ile yeni bir Queue oluşturulur.Oluşturulan Queue QueueBind()methodu kullanılarak ilgili exchange bind edilir. EventingBasicConsumer sınıfından bir nesne yaratılır.Oluşturulan channel parametre olarak geçilir.Üretilen nesne üzerinden Received event’i tetiklenir ve kuyruk dinlemeye alınır.Son olarak BasicConsume methodu ile ilgili kuyruktan mesaj alınır ve işlenir.

Fanout Exchange‘in çalışma yapısına baktığınızda gönderilen mesajın kuyruğu dinleyen tüm consumer’lara ayrım yapılmadan eşit bir şekilde dağıtıldığını göreceğiz.Örneğin döviz kuru,hava durumu vb bilgilerinin alınıp ilgili kuyruğu dinleyen tüm consumerlara gönderilmesi. Bu tür global bilgilendirme ve dağıtım senaryoları için oldukça uygun bir yapıdır diyebiliriz.Senaryolar değişkenlik gösterebilir. Hangi exchange yapısını kuracağınızı senaryonuz belirleyecektir.

Benim bu makalede aktaracaklarım bu kadar. Umarım faydalı ve anlaşılır bir yazı olmuştur. Bir sonraki makalede Direct Exchange konusu ile devam ediyor olacağız. Hoşçakalın 🙂

Source

https://www.rabbitmq.com/documentation.html
https://github.com/eaktassssss/RabbitMQExchange

P	S	Ç	P	C	C	P
				1	2	3
4	5	6	7	8	9	10
11	12	13	14	15	16	17
18	19	20	21	22	23	24
25	26	27	28	29	30	31