Github Proje: AyazDuru.Samples.RabbitMQ

RabbitMQ Nedir?

RabbitMQ, farklı uygulama ve servislerin birbirine mesaj göndermesini sağlayan açık kaynaklı bir mesaj kuyruklama (message broker) sistemidir.
Temel amacı, bir uygulamanın ürettiği veriyi, başka bir uygulamanın güvenli ve sıra ile almasını sağlamaktır. Özellikle mikroservis mimarisinde servislerin birbirinden bağımsız ve esnek şekilde haberleşmesine olanak tanır.

Başlıca Özellikleri:

• Mesajları sıraya alır ve iletir.
• Servisler arasında gevşek bağlılık (decoupling) sağlar.
• Yük paylaşımı ve asenkron işleme imkânı sunar.
• Farklı programlama dilleriyle kolayca entegre edilebilir.
• Güvenilir, ölçeklenebilir ve esnek bir altyapı sunar.

Kısaca:
RabbitMQ, uygulamalar arasında mesaj akışını yöneterek, sistemlerin daha güvenli, düzenli ve ölçeklenebilir çalışmasını sağlar.

RabbitMQ Temel Kavramları – Kargo Şirketi Örneğiyle

1. Exchange (Dağıtım Merkezi)

Kargo şirketinde dağıtım merkezi, gelen kargoları hangi şubeye gönderileceğine karar verir.
RabbitMQ’daki Exchange, gelen mesajların (kargoların) hangi kuyruğa (şubeye) yönlendirileceğini belirler.

• Direct Exchange: Kargonun üzerindeki adres (routing key) ile doğrudan belirli bir şubeye gönderilir.
  Örnek: "Ankara" adresli kargo doğrudan Ankara şubesine gider.
• Fanout Exchange: Kargo, adresine bakılmaksızın tüm şubelere dağıtılır.
  Örnek: Kampanya broşürü tüm şubelere gönderilir.
• Topic Exchange: Adresteki desenlere göre yönlendirme yapılır.
  Örnek: "İstanbul.Üsküdar" adresli kargo Üsküdar şubesine gider; "İstanbul." desenli şube ise tüm İstanbul kargolarını alır.*

2. Queue (Şube/Kargo Kuyruğu)

Her şube, kendisine gelen kargoları sıraya alır ve teslim eder.
RabbitMQ’da Queue, mesajların (kargoların) beklediği yerdir.

3. Routing Key (Adres)

Kargonun üzerinde yazan adres gibi düşünülebilir.
RabbitMQ’da Routing Key, mesajın hangi kuyruğa (şubeye) gideceğini belirler.

4. Virtual Host (Şirket Bölümü)

Kargo şirketinin farklı bölümleri olabilir (Express, Ekonomik, vs.).
Her bölümün kendi dağıtım merkezi ve şubeleri vardır.
RabbitMQ’da Virtual Host, izole çalışma alanları sağlar; Exchange, Queue ve diğer nesneler vhost içinde yönetilir.

Örnek Akılda Kalıcı Senaryo:

Bir müşteri, bir kargo gönderiyor. Kargo dağıtım merkezine (exchange) ulaşıyor. Adresi (routing key) okunuyor ve ilgili şubeye (queue) yönlendiriliyor. Şube, kargoyu teslim alıyor (consume ediyor).

RabbitMQ Nerelerde Kullanılır?

1. Mikroservis Mimarisi
Farklı mikroservislerin birbirinden bağımsız ve güvenli bir şekilde iletişim kurması için kullanılır. Örneğin; sipariş, fatura ve stok servisleri birbirine mesaj gönderebilir.

2. Asenkron İşlemler
Uzun süren işlemlerin (e-posta gönderimi, rapor üretimi, dosya işleme vb.) arka planda, ana iş akışını yavaşlatmadan yapılabilmesi için tercih edilir.

3. Yük Dengeleme (Load Balancing)
Çok sayıda mesajın birden fazla tüketici (worker) arasında adil şekilde dağıtılması, sistemin ölçeklendirilmesi için kullanılır.

4. Olay Tabanlı Sistemler (Event Driven Architecture)
Bir uygulamada gerçekleşen bir olayın diğer uygulama veya servislere bildirilmesi (örneğin; bir ürün stoğa girince diğer servislerin haberdar edilmesi).

5. Entegrasyonlar (Integration)
Farklı yazılım sistemleri arasında veri alışverişinin güvenli ve düzenli yapılabilmesi için kullanılır.

RabbitMQ Neden Kullanılır?

• Gevşek Bağlantı (Decoupling): Servislerin birbirine doğrudan bağlı olmasını önler; sistemlerin bağımsız ve esnek çalışmasını sağlar.
• Güvenilir Mesajlaşma: Mesajların kaybolmamasını, sıralı ve güvenli şekilde iletilmesini sağlar.
• Ölçeklenebilirlik: İş yükünün birden fazla tüketiciye dağıtılması ve sistemin kolayca büyütülmesi mümkündür.
• Hata Toleransı: Bir servis veya uygulama kısa süreliğine çalışmazsa bile mesajlar kaybolmaz; tekrar işlenebilir.
• Performans Artışı: Uzun süren işlemlerin ana iş akışını etkilemeden yapılmasını sağlar.

Özet:
RabbitMQ, servisler arası iletişim, asenkron işleme, olay tabanlı sistemler ve entegrasyon gibi birçok alanda; güvenilir, ölçeklenebilir ve esnek bir mesajlaşma altyapısı sağladığı için tercih edilir.

RabbitMQ Kurulumu

RabbitMQ’yu işletim sisteminize ve mevcut yazılım altyapınıza göre farklı yöntemlerle kurabilirsiniz.

Docker ile Hızlı Kurulum

Aşağıdaki komut ile RabbitMQ Docker imajını indirip çalıştırabilirsiniz:

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 -v rabbitmq_data:/var/lib/rabbitmq -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin rabbitmq:3-management

Bu komutu çalıştırdığınızda RabbitMQ sunucusu kısa sürede ayağa kalkar ve tarayıcı üzerinden Yönetim paneline (http://localhost:15672) erişebilirsiniz.
Varsayılan kullanıcı adı ve şifre: admin / admin

Docker ile kurulum hem hızlı hem de ek yapılandırma gerektirmeden çalışır.

AyazDuru.Samples.RabbitMQ.Consumer Projesi Örnek Kodları:

Program.cs

static void Main(string[] args)
{
    // Uygulama için varsayılan host yapılandırmasını oluşturur
    var builder = Host.CreateDefaultBuilder(args);

    builder.ConfigureServices(services =>
    {
        // RabbitMQ Client dinleyici arkaplan servisini ekler
        services.AddHostedService<RabbitMQClientConsumerService>();
        // MassTransit yapılandırması başlatılır
        services.AddMassTransit(x =>
        {
            // Mesaj tüketici sınıfını ekler
            x.AddConsumer<MassTransitMessageConsumer>();
            // RabbitMQ bağlantı ve endpoint ayarları yapılır
            x.UsingRabbitMq((context, cfg) =>
            {
                // RabbitMQ sunucu bağlantı bilgileri
                cfg.Host("localhost", "/", h => {
                    h.Username("admin"); // Kullanıcı adı
                    h.Password("admin"); // Şifre
                });

                // Mesajların alınacağı endpoint tanımlanır
                cfg.ReceiveEndpoint("masstransit-messages", e =>
                {
                    // Tüketici sınıfı endpoint'e bağlanır
                    e.Consumer<MassTransitMessageConsumer>(context);
                });
            });
        });

    });

    // Host nesnesi oluşturulur ve uygulama başlatılır
    var host = builder.Build();
    host.Run();
}

MassTransitMessageConsumer.cs

 public class MassTransitMessageConsumer : IConsumer<Message>
 {       
     public Task Consume(ConsumeContext<Message> context)
     {
         Console.WriteLine($" [MassTransit] {context.Message.Text}");
         return Task.CompletedTask;
     }
 }

RabbitMQClientConsumerService.cs

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        // RabbitMQ bağlantı ayarları oluşturuluyor.
        var factory = new ConnectionFactory()
        {
            HostName = "localhost",
            Port = 5672,
            UserName = "admin",
            Password = "admin",
            VirtualHost = "/"
        };
        // RabbitMQ sunucusuna bağlantı kuruluyor.
        using var connection = await factory.CreateConnectionAsync();
        // Bağlantı üzerinden bir kanal oluşturuluyor.
        using var channel = await connection.CreateChannelAsync();

        // "messages" isminde bir exchange (değişim noktası) oluşturuluyor. Tipi Fanout.
        await channel.ExchangeDeclareAsync(exchange: "messages",
            type: ExchangeType.Fanout);

        // Geçici bir kuyruk oluşturuluyor.
        QueueDeclareOk queueDeclareResult = await channel.QueueDeclareAsync();
        string queueName = queueDeclareResult.QueueName;
        // Kuyruk, "messages" exchange'ine bağlanıyor.
        await channel.QueueBindAsync(queue: queueName, exchange: "messages", routingKey: string.Empty);


        // Tüketici (consumer) nesnesi oluşturuluyor.
        var consumer = new AsyncEventingBasicConsumer(channel);
        // Mesaj alındığında çalışacak olay tanımlanıyor.
        consumer.ReceivedAsync += (model, ea) =>
        {
            // Mesajın içeriği byte dizisinden okunuyor.
            byte[] body = ea.Body.ToArray();
            var message = Encoding.UTF8.GetString(body);
            // Mesaj ekrana yazdırılıyor.
            Console.WriteLine($" [RabbitMQ.Client] {message}");
            return Task.CompletedTask;
        };

        // Tüketici, kuyruğa bağlanıyor ve mesajları otomatik olarak onaylıyor.
        await channel.BasicConsumeAsync(queueName, autoAck: true, consumer: consumer);

        // Servis durdurulana kadar bekleniyor.
        await Task.Delay(Timeout.Infinite, stoppingToken);
    }
}

AyazDuru.Samples.RabbitMQ.Publisher Projesi Örnek Kodları:

Program.cs

public static void Main(string[] args)
{
    var builder = WebApplication.CreateBuilder(args);            
    builder.Services.AddRazorPages();

    // Servis ekleniyor (MassTransit)
    builder.Services.AddMassTransit(x =>
    {
        // RabbitMQ konfigürasyonu yapılıyor
        x.UsingRabbitMq((context, cfg) =>
        {
            // RabbitMQ sunucusuna bağlantı ayarları
            cfg.Host("localhost", "/", h =>
            {
                h.Username("admin"); // Kullanıcı adı
                h.Password("admin"); // Şifre
            });
            // Mesajların alınacağı endpoint tanımlanıyor
            cfg.ReceiveEndpoint("masstransit-messages", e =>
            {
                // Burada mesaj işleyicileri eklenebilir
            });
        });
    });
    var app = builder.Build();

    if (!app.Environment.IsDevelopment())
    {
        app.UseExceptionHandler("/Error"); 
        app.UseHsts();
    }

    app.UseHttpsRedirection();

    app.UseRouting();

    app.UseAuthorization();

    app.MapStaticAssets(); 
    app.MapRazorPages()
       .WithStaticAssets(); 

    app.Run();
}

RabbitMQClientExample.cshtml.cs

public async Task<IActionResult> OnPostAsync()
{
    if (!ModelState.IsValid)
    {
        // Model doğrulaması başarısız ise sayfayı tekrar göster
        return Page();
    }
    // RabbitMQ bağlantı ayarlarını yapılandır
    var factory = new ConnectionFactory()
    {
        HostName = "localhost",
        Port = 5672,
        UserName = "admin",
        Password = "admin",
        VirtualHost = "/"
    };

    // RabbitMQ sunucusuna asenkron bağlantı oluştur
    using var connection = await factory.CreateConnectionAsync();
    // Kanal oluştur
    using var channel = await connection.CreateChannelAsync();

    // 'messages' isminde bir exchange tanımla (Fanout tipi)
    await channel.ExchangeDeclareAsync(exchange: "messages", type: ExchangeType.Fanout);

    // Mesajı UTF8 formatında byte dizisine dönüştür
    var body = Encoding.UTF8.GetBytes(Message);
    // Mesajı exchange'e yayınla
    await channel.BasicPublishAsync(exchange: "messages", routingKey: string.Empty, body: body);

    // Kanalı asenkron olarak kapat
    await channel.DisposeAsync();
    // Başarılı gönderim bilgisini TempData ile ilet
    TempData["Success"] = "Mesaj başarıyla gönderildi!";
    // Sayfayı yeniden yönlendir
    return RedirectToPage();
}

MassTransitExample.cshtml.cs

public class MassTransitExampleModel : PageModel
{
    private readonly ILogger<MassTransitExampleModel> _logger; // Logger servisi, loglama işleri için kullanılır
    private readonly IBus _bus; // Mesaj göndermek için kullanılan bus nesnesi

    public MassTransitExampleModel(ILogger<MassTransitExampleModel> logger, IBus bus)
    {
        _logger = logger;
        _bus = bus;
    }

    [BindProperty]
    public string Message { get; set; } // Gönderilecek mesajın içeriği

    public void OnGet()
    {

    }

    public async Task<IActionResult> OnPostAsync()
    {
        if (!ModelState.IsValid)
        {
            return Page();
        }

        await _bus.Publish(new Message { Text = Message }); // Mesajı yayınla
        TempData["Success"] = "Mesaj başarıyla gönderildi!"; // Başarı mesajını ayarla
        return RedirectToPage(); // Sayfayı yeniden yönlendir
    }
}

Son olarak iki ayrı kütüphaneyi kullanarak ekran çıktısı aşağıda göründüğü gibi test edildi.

Sağlıcakla kalın. Başka yazılarda görüşmek üzere..