Node.js İle Rabbitmq Kullanarak Asenkron Mesaj Kuyruğu Nasıl Yapılır?

Gereksinimler ve Ön Hazırlık

Çalışmaya başlamadan önce sisteminizde Node.js (LTS sürümü önerilir) ve RabbitMQ sunucusunun kurulu olması gerekir. RabbitMQ, mesajların geçici olarak tutulduğu bir broker görevi görür.

• Node.js (v20 veya üzeri önerilir).
• RabbitMQ Sunucusu (Docker üzerinde rabbitmq:3-management imajı ile kolayca ayağa kaldırılabilir).
• amqplib kütüphanesi: Node.js ile RabbitMQ arasındaki iletişimi sağlayan standart kütüphane.

Projenizi başlatmak için terminalinizde şu komutları çalıştırın:

mkdir rabbitmq-node-app
cd rabbitmq-node-app
npm init -y
npm install amqplib

Bu komutlar, projenizin temel yapılandırmasını oluşturacak ve RabbitMQ ile iletişim kurmamızı sağlayacak amqplib paketini projenize dahil edecektir.

RabbitMQ Temel Kavramları ve Çalışma Mantığı

RabbitMQ'yu verimli kullanmak için temel terimlere hakim olmalısınız. Mesaj kuyruğu yapısı, mesajı gönderen (Producer) ve mesajı işleyen (Consumer) birimlerden oluşur.

Adım Adım Mesaj Gönderici (Producer) Oluşturma

Producer, veriyi RabbitMQ'ya ileten kısımdır. İlk olarak bir bağlantı kurmalı, bir kanal (channel) açmalı ve mesajı ilgili kuyruğa göndermeliyiz.

const amqp = require('amqplib');

async function sendMessage() {
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();
    const queue = 'task_queue';
    const msg = 'Merhaba, bu bir asenkron mesajdır!';

    await channel.assertQueue(queue, { durable: true });
    channel.sendToQueue(queue, Buffer.from(msg), { persistent: true });

    console.log("Mesaj gönderildi:", msg);
    setTimeout(() => { connection.close(); }, 500);
}

sendMessage();

Burada durable: true parametresi, RabbitMQ sunucusu yeniden başlatılsa bile kuyruğun silinmemesini sağlar. persistent: true ise mesajın diskte saklanmasını garanti altına alır.

Adım Adım Mesaj İşleyici (Consumer) Oluşturma

Consumer, kuyruğu sürekli dinleyerek yeni bir mesaj geldiğinde onu işler. Mesaj başarıyla işlendikten sonra RabbitMQ'ya "mesajı aldım ve işledim" bilgisini (ack) göndermeliyiz.

const amqp = require('amqplib');

async function consumeMessages() {
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();
    const queue = 'task_queue';

    await channel.assertQueue(queue, { durable: true });
    channel.prefetch(1); // Aynı anda sadece 1 mesaj işle

    console.log("Mesajlar bekleniyor...");

    channel.consume(queue, (msg) => {
        console.log("Alınan mesaj:", msg.content.toString());
        setTimeout(() => {
            channel.ack(msg); // İşlemin bittiğini onayla
        }, 1000);
    }, { noAck: false });
}

consumeMessages();

channel.prefetch(1) kullanımı, bir consumer mesajı tam olarak işleyip onay (ack) gönderene kadar ona yeni mesaj verilmemesini sağlar. Bu, yük dengelemesi için hayati önem taşır.

Hata Yönetimi ve Güvenlik

Mesaj kuyruklarında en büyük risk, işlenemeyen mesajların kaybolması veya sistemin kilitlenmesidir. "Dead Letter Exchange" (DLX) yapısı kullanarak hatalı mesajları ayrı bir kuyruğa yönlendirebilirsiniz.

Kritik Uyarı: Üretim ortamında (production) RabbitMQ bağlantı bilgilerinizi (kullanıcı adı ve şifre) asla kod içerisinde açık metin olarak bulundurmayın. Ortam değişkenlerini (dotenv paketi ile) kullanın ve bağlantı hatalarını try-catch blokları ile yöneterek uygulamanın çökmesini engelleyin.

Aşağıdaki kod örneğinde bağlantı kopmaları için temel bir hata yönetimi gösterilmiştir:

try {
    const connection = await amqp.connect('amqp://user:password@localhost');
} catch (error) {
    console.error("RabbitMQ bağlantı hatası:", error.message);
    process.exit(1);
}

Sıkça Sorulan Sorular

RabbitMQ ile Redis Pub/Sub arasındaki fark nedir?

Redis Pub/Sub mesajları geçici olarak iletir ve alıcı çevrimdışıysa mesaj kaybolur. RabbitMQ ise mesajları kuyrukta tutar, alıcı müsait olduğunda iletir; bu da onu görev kuyrukları için daha güvenilir kılar.

Mesajlar neden işlenmiyor?

Genellikle channel.ack(msg) unutulmuştur. Eğer onay göndermezseniz, RabbitMQ mesajın işlenmediğini düşünür ve mesaj kuyrukta "unacked" durumunda takılı kalır.

Birden fazla Consumer çalıştırabilir miyim?

Evet, aynı kuyruğu dinleyen birden fazla Node.js süreci çalıştırarak iş yükünü dağıtabilirsiniz. RabbitMQ mesajları Round-Robin yöntemiyle consumer'lar arasında paylaştırır.

Mesajlar diskte mi tutulur?

Kuyruğu durable olarak tanımlarsanız ve mesajı persistent olarak gönderirseniz, RabbitMQ mesajları diske yazar.

Node.js uygulamasında performans kaybı yaşanır mı?

Asenkron yapısı sayesinde RabbitMQ, Node.js'in event-loop mekanizmasıyla mükemmel uyum sağlar. Ancak çok yoğun mesaj trafiğinde bağlantı havuzlarını (connection pooling) yönetmeniz gerekebilir.

RabbitMQ ile İleri Seviye Mesaj Yönlendirme: Exchange Tipleri

Temel kuyruk yapısının ötesine geçmek istediğinizde, RabbitMQ'nun sunduğu Exchange (Değiştirici) mekanizmalarını anlamak kritiktir. Mesajların doğrudan kuyruğa gitmesi yerine, bir Exchange üzerinden belirli kurallara göre dağıtılması, mikroservis mimarilerinde esneklik sağlar.

Direct, Fanout ve Topic Exchange Kullanımı

Exchange tipleri, mesajın hangi kuyruğa iletileceğini belirleyen "routing key" mantığını yönetir:

• Direct: Mesaj, routing key ile kuyruk ismi tam eşleşirse iletilir.
• Fanout: Mesaj, kendisine bağlı tüm kuyruklara kopyalanır. Yayın/Abone (Pub/Sub) modeli için idealdir.
• Topic: Routing key üzerinde joker karakterler (wildcard) kullanarak mesajları filtrelemenize olanak tanır.

Aşağıdaki örnekte, bir topic exchange kullanarak mesajları kategorize etmeyi görebilirsiniz:

// Mesaj gönderici (Producer) - Topic Exchange
const exchange = 'logs_topic';
const args = process.argv.slice(2);
const key = (args.length > 0) ? args[0] : 'anonymous.info';
const msg = args.slice(1).join(' ') || 'Merhaba Dünya!';

await channel.assertExchange(exchange, 'topic', { durable: false });
channel.publish(exchange, key, Buffer.from(msg));
console.log(" [x] Gönderildi: '%s':'%s'", key, msg);

RabbitMQ Mesaj İşleme Performansını Optimize Etme

Node.js uygulamalarında yüksek trafik altında RabbitMQ kullanırken, darboğazları önlemek için bazı optimizasyon tekniklerini uygulamanız gerekir. Özellikle mesaj işleme hızını artırmak için Prefetch Count ayarı hayati önem taşır.

Prefetch Count (QoS) Ayarının Önemi

Varsayılan olarak RabbitMQ, tüm mesajları aynı anda tüm consumer'lara dağıtmaya çalışır. Ancak, bir consumer'ın aynı anda kaç mesajı işleyebileceğini sınırlamak, sistemin çökmesini engeller. prefetch(1) ayarı, bir mesaj işlenip onaylanmadan (ack) consumer'a yeni mesaj gönderilmemesini sağlar.

// Consumer tarafında performans optimizasyonu
channel.prefetch(1); // Aynı anda sadece 1 mesajı işle

channel.consume(queue, (msg) => {
  const secs = msg.content.toString().split('.').length - 1;
  
  setTimeout(() => {
    console.log(" [x] İşlem tamamlandı");
    channel.ack(msg); // İşlem bittiğinde onay gönder
  }, secs * 1000);
}, { noAck: false });

Performans İçin İpuçları

Unutmayın, RabbitMQ üzerinde çok fazla kuyruk oluşturmak bellek kullanımını artırabilir. İhtiyaç duymadığınız kuyrukları auto-delete özelliği ile otomatik olarak temizlenmesini sağlayarak sistem kaynaklarını verimli kullanabilirsiniz.

Sonuç

Node.js ile RabbitMQ kullanarak asenkron mesaj kuyruğu mimarisini kurmak, uygulamanızın ölçeklenebilirliğini doğrudan artırır. Bu rehberde, temel bağlantı kurma, mesaj gönderme ve işleme süreçlerini öğrendiniz. Bir sonraki adım olarak, "Exchange" tiplerini (Fanout, Direct, Topic) araştırarak mesajlarınızın daha karmaşık kurallarla yönlendirilmesini sağlayabilirsiniz. Yazılım geliştirme süreçlerinizde her zaman hata toleransını ön planda tutmayı unutmayın.

Yasal Sorumluluk Reddi: Bu makalede paylaşılan kod örnekleri eğitim amaçlıdır. Ticari projelerinizde kullanmadan önce güvenlik denetimlerini yapmalı, bağlantı şifrelerini güvenli bir şekilde yönetmeli ve sisteminizin ihtiyaçlarına göre yapılandırmalısınız.