Bölüm 2: RabbitMQ Temelleri

İçindekiler

• RabbitMQ Nedir?
• AMQP Protokolü
• Temel Bileşenler
• Exchange Türleri
• Mesaj Akışı
• Acknowledgment ve Durability
• Virtual Host Kavramı

RabbitMQ-Mesaj Akışı

RabbitMQ Nedir?

RabbitMQ, Erlang dilinde yazılmış, açık kaynaklı bir message broker’dır. AMQP (Advanced Message Queuing Protocol) protokolünü implement eder ve “smart broker / dumb consumer” felsefesini benimser.

RabbitMQ, sistemler arasındaki iletişimi asenkron hale getiren, yüksek performanslı ve güvenilir bir Message Broker çözümüdür. Erlang dilinin doğasında bulunan yüksek eşzamanlılık (concurrency) ve hata toleransı yeteneklerini arkasına alır.

Temel İşleyiş Mekanizması

RabbitMQ’da mesajlar doğrudan bir kuyruğa bırakılmaz. Süreç, mesajın akıllı bir yönlendirme katmanından geçmesiyle başlar:

• Producer (Yayıncı): Mesajı oluşturan ve gönderen uygulama.
• Exchange (Santral): Mesajı karşılayan ilk duraktır. Gelen mesajın hangi kuyruğa (veya kuyruklara) gideceğine dair “trafik polisliği” yapar.
• Binding (Bağlantı): Exchange ile Queue (Kuyruk) arasındaki kural setidir.
• Queue (Kuyruk): Mesajların tüketilene kadar güvenle saklandığı, FIFO (First-In-First-Out) prensibiyle çalışan tampon bellek alanıdır.
• Consumer (Tüketici): Kuyruğu dinleyen ve mesaj geldiğinde onu işleyen uygulama.

Temel Özellikler

Ne Zaman RabbitMQ?

1. Kompleks Routing (Yönlendirme) Senaryoları Gerektiğinde

RabbitMQ’yu rakiplerinden ayıran en güçlü yanı, mesajın izleyeceği yolu çok hassas bir şekilde belirleyebilmesidir. Topic Exchange yapısı sayesinde, mesajları siparis.turkiye.istanbul veya odeme.kredikarti.basarili gibi hiyerarşik anahtarlarla etiketleyebilir, tüketicilerin sadece kendi uzmanlık alanlarına giren (örneğin sadece İstanbul siparişleri veya tüm başarılı ödemeler gibi) alt kümeleri dinlemesini sağlayabilirsiniz. Bu esneklik, monolitik yapılardan mikroservislere geçişte trafiği yönetmek için muazzam bir konfor sunar.

2. Farklı Protokol Desteği Gerektiğinde

Modern sistemlerde her zaman sadece HTTP veya gRPC ile haberleşemeyebilirsiniz. RabbitMQ, doğuştan gelen AMQP 0–9–1 desteğinin yanı sıra; IoT cihazları için MQTT, web tabanlı gerçek zamanlı iletişim için STOMP ve hatta HTTP üzerinden mesaj gönderimi gibi geniş bir protokol yelpazesine sahiptir. Bu “çok dilli” yapısı, farklı teknoloji yığınlarına (Legacy sistemler, mobil cihazlar, gömülü sistemler) sahip platformları tek bir mesaj merkezinde birleştirmeyi mümkün kılar.

3. Mesaj Önceliklendirme (Priority Queues) Gerektiğinde

Tüm mesajlar eşit yaratılmamıştır. Örneğin, binlerce standart e-posta gönderilirken araya giren bir “Şifre Sıfırlama” e-postasının en öne geçmesi kritik olabilir. RabbitMQ, kuyruk bazlı Priority desteği ile mesajlara 0–255 arası önem derecesi atamanıza izin verir. Eğer sisteminizde bazı işlemlerin (VIP müşteri siparişleri veya sistem alarm sinyalleri gibi) yoğun trafik altında bile “sırayı beklemeden” işlenmesi gerekiyorsa, RabbitMQ burada devreye girer.

4. Request-Reply Pattern İçin

Mesajlaşma sistemleri genellikle “at ve unut” (fire-and-forget) prensibiyle çalışsa da, bazen gönderdiğiniz mesajın sonucunu senkron bir şekilde beklemeniz gerekebilir. RabbitMQ, Reply-To ve Correlation ID özellikleriyle bu “istek-cevap” mekanizmasını yerleşik olarak destekler. Producer bir mesaj gönderirken bir “temp” (geçici) kuyruk adresi tanımlar; Consumer işi bitince sonucu bu adrese döner. Bu sayede asenkron bir altyapı üzerinde, iki servis arasında güvenli bir RPC (Remote Procedure Call) köprüsü kurulmuş olur.

5. Düşük Latency (Gecikme) Gerektiğinde

RabbitMQ, mesajları RAM üzerinde işlemeye odaklanan bir mimariye sahiptir. Mesajın bir producer’dan çıkıp bir consumer’a ulaşması milisaniyeler, hatta bazen mikrosaniyeler mertebesinde gerçekleşir. Özellikle Kafka gibi disk tabanlı (log-structured) sistemlerin aksine, RabbitMQ mesajı “tüketildiği anda yok edilecek bir canlı” gibi gördüğü için kuyruklar boş olduğunda inanılmaz bir hız performansı sergiler. Bu, anlık tepki süresinin (real-time responsiveness) kritik olduğu finansal işlemler veya oyun sunucuları için büyük bir avantajdır.

AMQP Protokolü

AMQP (Advanced Message Queuing Protocol), mesajlaşma sistemleri için tasarlanmış açık standart bir uygulama katmanı protokolüdür.

AMQP Model

AMQP Frame Yapısı

RabbitMQ’da iletişim, “Frame” adı verilen bu küçük veri paketleri üzerinden yürütülür. Her bir frame, TCP bağlantısı üzerinden akan disiplinli birer atomik birimdir.

1. Frame Header (7 Byte)

İletişimin “pasaportu” burasıdır. Alıcıya (Broker veya Client) gelen verinin ne olduğunu söyler:

• Type (1 byte): Frame’in türünü belirtir. Örneğin; bir metod mu (Method Frame), mesaj içeriği mi (Content Header) yoksa ham veri mi (Content Body) olduğunu burası belirler.
• Channel (2 byte): RabbitMQ’nun en güçlü yanlarından biri olan Multiplexing burada hayat bulur. Tek bir TCP bağlantısı içinde binlerce bağımsız kanal açılabilmesini sağlar. Her frame, hangi kanala ait olduğunu bu ID ile bilir.
• Size (4 byte): Payload kısmının tam olarak kaç byte olduğunu belirtir. Bu, alıcının bellekte ne kadar yer ayıracağını bilmesi için kritiktir.

2. Frame Payload (Değişken)

Asıl “yükün” taşındığı kısımdır. Eğer bu bir Method Frame ise hangi AMQP komutunun (örneğin Basic.Publish veya Queue.Declare) çalıştırılacağı bilgisini taşır. Eğer bir Content Body ise gönderdiğin asıl mesaj (JSON, XML veya Binary veri) burada yer alır.

3. Frame End (1 Byte: 0xCE)

Bu, frame’in güvenli bir şekilde sona erdiğini belirten bir “stop” işaretidir. AMQP protokolünde bu değer her zaman onaltılık tabanda 0xCE (decimal: 206) olarak belirlenmiştir. Eğer alıcı bu byte'ı beklediği yerde bulamazsa, veri iletiminde bir hata (framing error) olduğunu anlar ve bağlantıyı keser.

Neden Bu Yapı Önemli?

Bu katmanlı yapı sayesinde RabbitMQ, Binary Protocol kullanmanın avantajlarını sonuna kadar kullanır:

• Hız: Metin tabanlı (HTTP/JSON gibi) protokollerin aksine, binary formatta ayrıştırma (parsing) çok daha hızlıdır.
• Güvenlik: Paketlerin sonundaki 0xCE kontrolü, veri bütünlüğünü donanım seviyesine yakın bir hızda doğrular.
• Verimlilik: Tek bir TCP bağlantısı üzerinden binlerce kanalın (Channel) akabilmesi, sistem kaynaklarını (CPU/RAM) minimize eder.

AMQP Frame Tipleri

Temel Bileşenler

Mimari Genel Bakış

1. TCP Connection ve Channel (Bağlantı ve Kanallar)

Görselde görüldüğü gibi, fiziksel bir TCP Connection içerisinde birden fazla Channel (Kanal) yer almaktadır. RabbitMQ mimarisinde TCP bağlantısı kurmak maliyetli bir iştir; bu yüzden “Multiplexing” tekniği kullanılır. Tek bir bağlantı üzerinden açılan bu sanal kanallar (Channel 1, 2, 3), uygulamanın kaynak tüketimini minimize ederken aynı anda birden fazla mesajlaşma işlemini (yayınlama veya tüketme) birbirinden bağımsız şekilde yönetmemize olanak tanır.

2. Virtual Host (vhost)

Broker içerisindeki Virtual Host (/production), RabbitMQ’nun çok kiracılı (multi-tenancy) yapısını temsil eder. Bir fiziksel sunucuyu mantıksal olarak bölümlere ayırmamızı sağlar. Görseldeki /production alanı; kendi Exchange'lerine, Kuyruklarına ve yetkilendirme kurallarına sahip izole bir evrendir. Bu yapı sayesinde test, geliştirme ve üretim ortamlarını aynı broker üzerinde birbirine karışmadan yönetebiliriz.

3. Exchange ve Binding (Yönlendirme Mantığı)

Mesajın kalbi burasıdır. Exchange (orders ve notifications), Publisher’dan gelen mesajı karşılayan ve kurallara göre dağıtan mekanizmadır. Aradaki Binding okları ise “yönlendirme kurallarını” temsil eder. Örneğin, bir sipariş oluştuğunda bu mesaj orders exchange'ine gelir ve tanımlı binding kuralları sayesinde doğruca order-processing kuyruğuna akar. Bu sayede mesajın nereye gideceğine gönderici değil, broker karar verir.

4. Queues (Kuyruklar)

Görseldeki order-processing, email-queue ve sms-queue yapıları, mesajların güvenle bekletildiği tampon alanlardır. RabbitMQ burada bir "buffer" görevi görerek, bir sistem çok yoğun olsa bile mesajların kaybolmasını engeller. Her bir kuyruk, kendi iş yüküne göre bağımsız olarak izlenebilir ve ölçeklendirilebilir; örneğin SMS gönderimi yavaşlarsa sadece sms-queue birikir, diğer süreçler bundan etkilenmez.

5. Publisher ve Consumer (Yayıncı ve Tüketici)

Sistemin dış dünyayla bağlantısını sağlayan uç noktalardır. Publisher 1, işi başlatan ve mesajı atan birimdir. Consumer 1 ve 2 ise farklı kanallar üzerinden ilgili kuyrukları dinleyerek gelen veriyi işleyen servislerdir. Görselde dikkat çeken en önemli detay, bir tüketicinin (Consumer 1) order-processing ve email-queue gibi birden fazla kuyruğu aynı anda yönetebilmesidir.

1. Connection (Bağlantı)

TCP bağlantısı üzerinden kurulan AMQP bağlantısı.

// ===== pom.xml =====
/*
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
*/

// ===== application.yml =====
/*
spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    connection-timeout: 5000

    # Connection recovery
    template:
      retry:
        enabled: true
        initial-interval: 5000
        max-attempts: 3
*/

// ===== RabbitMQConnectionConfig.java =====
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConnectionConfig {

    @Bean
    public CachingConnectionFactory connectionFactory() {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");

        // Heartbeat (saniye cinsinden)
        factory.getRabbitConnectionFactory().setRequestedHeartbeat(600);

        // Connection recovery
        factory.getRabbitConnectionFactory().setAutomaticRecoveryEnabled(true);
        factory.getRabbitConnectionFactory().setNetworkRecoveryInterval(5000);

        return factory;
    }
}

**EK-C RabbitMQ Consumer Çalışma Prensibi: Push Modeli & QoS**

2. Channel (Kanal)

Connection içinde oluşturulan hafif, sanal bağlantı. Çoğu işlem channel üzerinden yapılır.

// Spring AMQP'de channel otomatik yönetilir
// RabbitTemplate veya @RabbitListener kullanıldığında channel açılır/kapanır

@Autowired
private RabbitTemplate rabbitTemplate;  // Channel dahili olarak yönetilir

// Manuel channel erişimi gerekirse:
rabbitTemplate.execute(channel -> {
    // Channel üzerinde işlemler
    channel.queueDeclare("my-queue", true, false, false, null);
    return null;
});

Neden Channel?

• TCP bağlantısı maliyetlidir
• Bir connection içinde birden fazla channel açılabilir
• Her thread için ayrı channel önerilir
• İzolasyon sağlar

3. Exchange

Exchange, RabbitMQ’da mesajların “yönlendirme merkezidir”. Producer mesajı doğrudan queue’ya yazmaz; mesajı exchange’e publish eder. Exchange de mesajı, tanımlı kurallara (routing key / binding) göre doğru queue(lar)a dağıtır. Bunun faydası, producer ile consumer’ı birbirinden ayırmasıdır: Producer hangi queue’ların var olduğunu bilmeden mesaj gönderebilir; siz de sonradan yeni queue/consumer ekleyebilir, mesajı birden fazla kuyruğa yönlendirebilir ve routing kurallarını değiştirerek producer koduna dokunmadan sistemi esnekçe büyütebilirsiniz.

Bunu bir postane gibi düşünebilirsiniz: Gönderen (producer) mektubu doğrudan alıcının posta kutusuna bırakmaz; postaneye (exchange) teslim eder. Postane, zarfın üzerindeki adrese (routing key) ve dağıtım kurallarına (binding) bakarak mektubu doğru posta kutusuna (queue) ulaştırır; alıcı (consumer) da mektubu posta kutusundan alıp işler. (*EK-E RabbitMQ Exchange: Kapsamlı Rehber ve Temel Kavramlar)*

// Exchange tanımlama - Configuration sınıfında @Bean olarak
@Configuration
public class ExchangeConfig {

    @Bean
    public DirectExchange ordersExchange() {
        return ExchangeBuilder
            .directExchange("orders")
            .durable(true)         // Broker restart'ta hayatta kalır
            .build();

        // Alternatif kısa syntax:
        // return new DirectExchange("orders", true, false);
    }
}

4. Queue (Kuyruk)

Mesajların saklandığı buffer. FIFO (First-In-First-Out) mantığıyla çalışır. Queue, RabbitMQ’da mesajların güvenli şekilde biriktiği ve tüketilmeyi beklediği “bekleme kuyruğudur”. Exchange tarafından yönlendirilen mesajlar queue’ya yazılır; consumer’lar da mesajları buradan alıp işler. Faydası, producer ile consumer hızlarını birbirinden ayırmasıdır: Consumer yavaşsa mesajlar queue’da sıraya girer, consumer ölçeklenirse (birden fazla consumer) kuyruktaki mesajlar daha hızlı tüketilir; böylece sistem yük altında daha dayanıklı ve dengeli çalışır.

Bunu bir posta kutusu gibi düşünebilirsiniz: Postane (exchange) mektupları ilgili posta kutusuna (queue) bırakır. Alıcı (consumer) müsait olduğunda posta kutusunu açar, mektubu alır ve okur/işler. Alıcı gecikse bile mektup kaybolmaz; posta kutusunda bekler.

// Queue tanımlama - Configuration sınıfında @Bean olarak
@Configuration
public class QueueConfig {

    @Bean
    public Queue orderProcessingQueue() {
        return QueueBuilder
            .durable("order-processing")    // Broker restart'ta hayatta kalır
            // .exclusive()                 // Tek connection erişebilir (false)
            // .autoDelete()                // Son consumer ayrılınca sil (false)
            .ttl(86400000)                  // 24 saat TTL (milisaniye)
            .maxLength(10000)               // Max 10000 mesaj
            .deadLetterExchange("dlx")      // Dead letter exchange
            .build();
    }

    // Alternatif: Quorum Queue (production için önerilen)
    @Bean
    public Queue orderProcessingQuorumQueue() {
        return QueueBuilder
            .durable("order-processing")
            .quorum()                       // Quorum queue (HA)
            .ttl(86400000)
            .deadLetterExchange("dlx")
            .build();
    }
}

5. Binding

Exchange ile Queue arasındaki bağlantı kuralı.

Binding, RabbitMQ’da exchange ile queue arasındaki “bağlantı ve yönlendirme kuralıdır”. Yani “Bu exchange’e gelen mesajlar hangi koşulda bu queue’ya düşsün?” sorusunun cevabıdır. Exchange tek başına sadece mesajı alır; binding olmadan mesajın hangi queue’ya gideceği netleşmez. Faydası, routing kurallarını kod değiştirmeden yönetebilmenizdir: Yeni bir queue ekleyip sadece binding tanımlayarak aynı mesajları yeni consumer’lara da ulaştırabilirsiniz; ya da kuralı değiştirip mesaj akışını farklı queue’lara yönlendirebilirsiniz.

Bunu postane dağıtım kuralı gibi düşünebilirsiniz: Postanedeki “Şu mahalleye giden mektuplar şu posta kutusuna bırakılır” talimatı binding’dir. Zarfın üzerindeki adres (routing key) bu kurala uyarsa mektup ilgili posta kutusuna (queue) bırakılır.

Detay ve Stratejiler için (EK-G RabbitMQ’da Consumer, Exchange ve Binding İlişkisi ile Stratejileri)

// Binding tanımlama - Configuration sınıfında @Bean olarak
@Configuration
public class BindingConfig {

    @Bean
    public Binding orderCreatedBinding(Queue orderProcessingQueue, DirectExchange ordersExchange) {
        return BindingBuilder
            .bind(orderProcessingQueue)
            .to(ordersExchange)
            .with("order.created");  // Routing key
    }

    // Birden fazla routing key için birden fazla binding
    @Bean
    public Binding orderUpdatedBinding(Queue orderProcessingQueue, DirectExchange ordersExchange) {
        return BindingBuilder
            .bind(orderProcessingQueue)
            .to(ordersExchange)
            .with("order.updated");
    }
}

Exchange Türleri

Exchange türleri, RabbitMQ’da bir mesajın hangi kurala göre hangi queue(lar)a gideceğini belirler. Producer mesajı exchange’e gönderir; exchange’in tipi, mesajın dağıtım mantığını belirler: bazen tam eşleşme ile tek kuyruğa, bazen pattern ile seçili kuyruğa, bazen de herkese kopyalayarak tüm kuyruklara gönderir. Bu sayede aynı mesajlaşma altyapısında, ihtiyaca göre farklı dağıtım stratejileri uygulanır.

Kısaca türler:

• Direct Exchange: Routing key tam eşleşirse ilgili queue’ya gider.
• Topic Exchange: Routing key pattern ile eşleşirse gider (*, #).
• Fanout Exchange: Anahtar bakmadan tüm bağlı queue’lara kopyalar (broadcast).
• Headers Exchange: Routing key yerine header kurallarına göre yönlendirir.

Postane analojisiyle: Exchange türü, postanenin “dağıtım politikasıdır”:

• Direct = adres birebir tutarsa tek kutuya bırak,
• Topic = adreste şablon/pattern tutarsa bırak,
• Fanout = tüm kutulara fotokopi dağıt,
• Headers = zarf üzerindeki ek etiketlere göre ayır.

**Kapsamlı rehbere** EK-E RabbitMQ Exchange: Kapsamlı Rehber ve Temel Kavramlar EK-F RabbitMQ Exchange, Routing Key ve Binding Key Yapısı

1. Direct Exchange

Routing key tam eşleşmesine göre mesajları yönlendirir.

Direct Exchange, mesajları routing key’in binding key ile tam (birebir) eşleşmesine göre ilgili queue’ya yönlendirir. Yani “adres etiketi neyse, aynısı hangi kuyruğa bağlanmışsa oraya gitsin” mantığı vardır. Faydası, net ve kontrollü bir dağıtım sağlamasıdır: Mesajlar karışmaz, her iş türü kendi kuyruğuna gider; ayrıca yeni bir consumer eklemek için producer’ı değiştirmeden sadece yeni bir queue ve binding tanımlayabilirsiniz.

Postane analojisiyle: Zarfın üzerinde “Kadıköy” yazıyorsa, postane (exchange) sadece “Kadıköy” etiketine birebir bağlı posta kutusuna (queue) bırakır; “Kadikoy” ya da “Kadıköy/Moda” gibi farklı yazıldıysa o kutuya bırakmaz.

// ===== Direct Exchange Konfigürasyonu =====
@Configuration
public class DirectExchangeConfig {

    public static final String LOGS_EXCHANGE = "logs";
    public static final String ERROR_QUEUE = "error-queue";
    public static final String INFO_QUEUE = "info-queue";

    @Bean
    public DirectExchange logsExchange() {
        return new DirectExchange(LOGS_EXCHANGE);
    }

    @Bean
    public Queue errorQueue() {
        return new Queue(ERROR_QUEUE);
    }

    @Bean
    public Queue infoQueue() {
        return new Queue(INFO_QUEUE);
    }

    @Bean
    public Binding errorBinding(Queue errorQueue, DirectExchange logsExchange) {
        return BindingBuilder.bind(errorQueue).to(logsExchange).with("error");
    }

    @Bean
    public Binding infoBinding(Queue infoQueue, DirectExchange logsExchange) {
        return BindingBuilder.bind(infoQueue).to(logsExchange).with("info");
    }
}

// ===== Producer Service =====
@Service
@RequiredArgsConstructor
public class LogPublisher {

    private final RabbitTemplate rabbitTemplate;

    public void publishError(String message) {
        rabbitTemplate.convertAndSend("logs", "error", "Kritik hata oluştu!");
    }

    public void publishInfo(String message) {
        rabbitTemplate.convertAndSend("logs", "info", "İşlem başarılı");
    }
}

2. Fanout Exchange

Tüm bağlı queue’lara mesajı kopyalar. Routing key’i yok sayar. Fanout Exchange, gelen mesajı routing key’e bakmadan kendisine bağlı tüm queue’lara kopyalayarak dağıtır. Yani tek bir publish ile aynı mesaj, birden fazla kuyruğa “broadcast” edilir. Faydası, aynı olayı birden çok farklı amaçla tüketmek istediğiniz senaryolarda (loglama, audit, bildirim, cache invalidation, analytics vb.) producer’ı değiştirmeden mesajı herkese ulaştırmasıdır.

Postane analojisiyle: Postane (exchange) gelen duyuruyu alır ve adres kontrolü yapmadan her mahalledeki tüm posta kutularına aynı duyurunun fotokopisini bırakır.

// ===== Fanout Exchange Konfigürasyonu =====
@Configuration
public class FanoutExchangeConfig {

    @Bean
    public FanoutExchange notificationsExchange() {
        return new FanoutExchange("notifications");
    }

    @Bean
    public Queue emailQueue() {
        return new Queue("email-notifications");
    }

    @Bean
    public Queue smsQueue() {
        return new Queue("sms-notifications");
    }

    @Bean
    public Queue pushQueue() {
        return new Queue("push-notifications");
    }

    // Fanout'ta routing key yok sayılır
    @Bean
    public Binding emailBinding(Queue emailQueue, FanoutExchange notificationsExchange) {
        return BindingBuilder.bind(emailQueue).to(notificationsExchange);
    }

    @Bean
    public Binding smsBinding(Queue smsQueue, FanoutExchange notificationsExchange) {
        return BindingBuilder.bind(smsQueue).to(notificationsExchange);
    }

    @Bean
    public Binding pushBinding(Queue pushQueue, FanoutExchange notificationsExchange) {
        return BindingBuilder.bind(pushQueue).to(notificationsExchange);
    }
}

// ===== Producer Service =====
@Service
@RequiredArgsConstructor
public class NotificationPublisher {

    private final RabbitTemplate rabbitTemplate;

    public void broadcast(String message) {
        // Routing key önemli değil (boş veya herhangi değer)
        rabbitTemplate.convertAndSend(
            "notifications", 
            "",  // Fanout routing key'i yok sayar
            "Sistem bakımı yarın saat 03:00'te yapılacak"
        );
    }
}

**Kullanım Alanları:

• **Broadcast mesajlar
• Cache invalidation
• Real-time updates
• Sistem duyuruları

3. Topic Exchange

Routing key pattern eşleşmesine göre yönlendirme yapar.

Wildcard Karakterleri: * (yıldız): Tam olarak bir kelime # (hash): Sıfır veya daha fazla kelime

Topic Exchange, mesajları routing key’in bir pattern (şablon) ile eşleşmesine göre ilgili queue(lar)a yönlendirir. Direct’te “tam eşleşme” varken burada “kuralı tutan herkes alsın” mantığı vardır. Pattern tarafında genelde * (tek kelime) ve # (0 veya daha fazla kelime) jokerleri kullanılır. Faydası, routing’i esnek hale getirmesidir: Aynı producer mesaj formatını bozmadan farklı consumer’lar kendi ilgi alanlarına göre pattern ile abone olabilir; yeni tüketiciler eklemek için producer koduna dokunmanız gerekmez.

Postane analojisiyle: Zarfın üzerinde “İstanbul / Sarıyer / Kargo” gibi bir adres hiyerarşisi varmış gibi düşünün. Postane, “İstanbul’daki tüm kargolar” veya “Sarıyer’deki her şey” gibi şablon kuralları olan kutulara mektubu bırakır; kurala uyan tüm posta kutuları aynı mektubu alır.

// ===== Topic Exchange Konfigürasyonu =====
@Configuration
public class TopicExchangeConfig {

    @Bean
    public TopicExchange eventsExchange() {
        return new TopicExchange("events");
    }

    @Bean
    public Queue euOrdersQueue() {
        return new Queue("eu-orders");
    }

    @Bean
    public Queue allCreatedQueue() {
        return new Queue("all-created");
    }

    @Bean
    public Queue everythingQueue() {
        return new Queue("everything");
    }

    // Binding örnekleri - Wildcard kullanımı
    @Bean
    public Binding euOrdersBinding(Queue euOrdersQueue, TopicExchange eventsExchange) {
        return BindingBuilder
            .bind(euOrdersQueue)
            .to(eventsExchange)
            .with("order.eu.*");    // order.eu.created, order.eu.shipped vb.
    }

    @Bean
    public Binding allCreatedBinding(Queue allCreatedQueue, TopicExchange eventsExchange) {
        return BindingBuilder
            .bind(allCreatedQueue)
            .to(eventsExchange)
            .with("*.*.created");   // order.eu.created, payment.tr.created vb.
    }

    @Bean
    public Binding everythingBinding(Queue everythingQueue, TopicExchange eventsExchange) {
        return BindingBuilder
            .bind(everythingQueue)
            .to(eventsExchange)
            .with("#");             // Tüm mesajlar
    }
}

// ===== Producer Service =====
@Service
@RequiredArgsConstructor
public class EventPublisher {

    private final RabbitTemplate rabbitTemplate;

    public void publishEuOrder() {
        // eu-orders ve all-created queue'larına gider
        rabbitTemplate.convertAndSend(
            "events",
            "order.eu.created",
            "Yeni EU siparişi"
        );
    }
}

4. Headers Exchange

Routing key yerine mesaj header’larına göre yönlendirme yapar. Headers Exchange, mesajları routing key yerine mesajın header alanlarına bakarak yönlendirir. Yani “anahtar şu olsun” yerine “header’da type=vip ve region=tr varsa şu queue’ya gitsin” gibi özellik bazlı kurallar tanımlanır. Faydası, routing key yapısını büyütmeden veya karmaşıklaştırmadan, mesajı etiketler/metadata üzerinden esnek şekilde sınıflandırabilmenizdir; özellikle çok boyutlu filtreleme gereken durumlarda işe yarar.

Postane analojisiyle: Zarfın adresinden çok, zarfın üzerindeki etiketlere bakıldığını düşünün: “Acele”, “Kırılacak”, “Yurtiçi”, “VIP” gibi. Postane (exchange) bu etiket kombinasyonlarına göre mektubu uygun posta kutusuna (queue) yönlendirir.

// ===== Headers Exchange Konfigürasyonu =====
@Configuration
public class HeadersExchangeConfig {

    @Bean
    public HeadersExchange documentsExchange() {
        return new HeadersExchange("documents");
    }

    @Bean
    public Queue pdfProcessorQueue() {
        return new Queue("pdf-processor");
    }

    // Binding - tüm header'lar eşleşmeli (x-match: all)
    @Bean
    public Binding pdfFinanceBinding(Queue pdfProcessorQueue, HeadersExchange documentsExchange) {
        return BindingBuilder
            .bind(pdfProcessorQueue)
            .to(documentsExchange)
            .whereAll(Map.of(
                "format", "pdf",
                "department", "finance"
            ))
            .match();
    }

    // Binding - herhangi bir header eşleşirse (x-match: any)
    @Bean
    public Binding anyReportBinding(Queue reportsQueue, HeadersExchange documentsExchange) {
        return BindingBuilder
            .bind(reportsQueue)
            .to(documentsExchange)
            .whereAny(Map.of(
                "type", "report",
                "priority", "high"
            ))
            .match();
    }
}

// ===== Producer Service =====
@Service
@RequiredArgsConstructor
public class DocumentPublisher {

    private final RabbitTemplate rabbitTemplate;

    public void publishDocument(byte[] content) {
        rabbitTemplate.convertAndSend(
            "documents",
            "",  // Headers exchange'de routing key kullanılmaz
            content,
            message -> {
                MessageProperties props = message.getMessageProperties();
                props.setHeader("format", "pdf");
                props.setHeader("department", "finance");
                props.setHeader("priority", "high");
                return message;
            }
        );
    }
}

Exchange Türleri Karşılaştırması

Mesaj Akışı

Detaylı Mesaj Yaşam Döngüsü

Mesaj Properties

// ===== MessageProperties ile Mesaj Gönderme =====
@Service
@RequiredArgsConstructor
public class MessagePublisher {

    private final RabbitTemplate rabbitTemplate;

    public void publishWithProperties(String content) {
        rabbitTemplate.convertAndSend(
            "orders",
            "order.created",
            content,
            message -> {
                MessageProperties props = message.getMessageProperties();

                // Content
                props.setContentType("application/json");
                props.setContentEncoding("utf-8");

                // Headers
                props.setHeader("custom-header", "value");

                // Delivery
                props.setDeliveryMode(MessageDeliveryMode.PERSISTENT); // 2: persistent
                props.setPriority(5);  // 0-9 arası

                // Correlation
                props.setCorrelationId("abc-123");
                props.setReplyTo("response-queue");

                // TTL
                props.setExpiration("60000");  // 60 saniye

                // Identification
                props.setMessageId("msg-001");
                props.setTimestamp(new Date());
                props.setType("order.created");
                props.setUserId("guest");
                props.setAppId("order-service");

                return message;
            }
        );
    }
}

// ===== DTO ile Otomatik Serialize (Önerilen) =====
public record OrderCreatedEvent(
    String orderId,
    String customerId,
    BigDecimal totalAmount,
    Instant timestamp
) {}

@Service
public class OrderEventPublisher {

    private final RabbitTemplate rabbitTemplate;

    public void publish(OrderCreatedEvent event) {
        // Jackson2JsonMessageConverter otomatik serialize eder
        rabbitTemplate.convertAndSend("orders", "order.created", event);
    }
}

Acknowledgment ve Durability

Acknowledgment (Ack) nedir?

Ack, consumer’ın “Bu mesajı aldım ve başarıyla işledim” diye broker’a (RabbitMQ) verdiği onaydır. Faydası: Consumer çökerse / timeout olursa, ack gelmeyen mesaj kayıp olmaz, başka bir consumer’a yeniden teslim edilebilir (redelivery).

Acknowledgment modları (Consumer tarafı)

1) Auto Ack (No-Ack)

• Consumer mesajı alır almaz RabbitMQ mesajı “işlendi” sayar.
• Avantaj: En düşük gecikme, en yüksek throughput.
• Dezavantaj: Consumer mesajı aldıktan sonra çökerse mesaj kaybolabilir (çünkü broker zaten sildi).

“Kaybolması sorun değil / telemetri gibi” senaryolarda.

2) Manual Ack (Explicit Ack)

• Consumer işi gerçekten bitirince basic.ack gönderir.
• Avantaj: En güvenli teslim (en yaygın kullanılan).
• Dezavantaj: Doğru yönetilmezse kuyruk şişebilir (prefetch/backpressure önemli).

Bu modun içinde pratikte 3 davranış var:

a) basic.ack

• “Başarılı işlendi, kuyruktan sil.”

b) basic.nack / basic.reject

• “İşleyemedim.”
• requeue=true ise mesaj kuyruğa geri döner (yeniden denenecek).
• requeue=false ise mesaj düşer → genelde DLQ (dead-letter queue)’ya gider (varsa).

***reject tek mesaj içindir; `nack`* toplu/çoklu kullanım ve bazı client’larda daha esnek kontrol sağlar.

c) “Multiple” (toplu ack)

• Consumer birden fazla mesajı işliyorsa, “şu deliveryTag’e kadar olanların hepsi ok” diye topluca ack atabilir.
• Avantaj: Network overhead azalır.
• Risk: Arada bir mesaj fail olursa tasarımınız karışabilir.

Durability (Kalıcılık) Seviyeleri

Durability, broker restart/crash gibi durumlarda mesaj ve yapıların diskte kalması ile ilgilidir.

Durable olsun diye genelde 3 şey birlikte gerekir:

1. Durable Exchange
2. Durable Queue
3. Persistent Message (mesaj “delivery mode = persistent”)

Sadece queue durable yapmak yetmez; mesaj persistent değilse broker restart’ta uçabilir.

Ek dayanıklılık için (özellikle node kaybı senaryosunda):

• Quorum Queue (replicated, daha güvenli) gibi replikasyon seçenekleri kullanılır.

Çok kritik ayrım

• Ack → “Consumer işledi mi?” (işleme garantisi)
• Durability → “Broker düşerse veri kalır mı?” (saklama garantisi)

Seviye 1 — Geçici (Transient)

• Non-durable queue/exchange + transient mesaj
• Broker restart olursa: Queue/exchange de gider, mesajlar da gider.
• Kullanım: “Kaybolsa da olur” telemetry, cache benzeri akışlar.

Seviye 2 — Yapı Kalıcı, Mesaj Geçici

• Durable queue/exchange var, ama mesajlar persistent değil
• Broker restart olursa: Queue/exchange durur, ama mesajlar uçabilir.
• Kullanım: Yapı sabit kalsın ama mesaj kaybı kritik değilse.

Seviye 3 — Tam Kalıcılık (Disk Üzerinde)

• Durable queue + durable exchange + persistent mesaj
• Broker restart olursa: Hem yapı hem mesajlar büyük ölçüde korunur.
• Not: Bu seviyede bile en sağlam garanti için genelde Publisher Confirms (producer tarafı onay) kullanılır; yoksa “mesaj gerçekten diske yazıldı mı?” kısmı zayıf kalabilir.

Seviye 4— Replikasyonlu Kalıcılık (Node Kaybına Dayanıklı)

• Tek node değil, cluster içinde kopyalı saklama (pratikte en yaygın: Quorum Queue)
• Bir node düşse bile: mesajlar diğer node’lardan yaşamaya devam eder.
• Kullanım: Finansal/ödeme/cüzdan gibi “kayıp kabul etmem” sistemler.

Tam Kalıcılık için:

// ===== Tam Kalıcılık Konfigürasyonu =====
@Configuration
public class DurableConfig {

    // 1. Durable Exchange
    @Bean
    public DirectExchange durableOrdersExchange() {
        return ExchangeBuilder
            .directExchange("orders")
            .durable(true)  // Broker restart'ta korunur
            .build();
    }

    // 2. Durable Queue
    @Bean
    public Queue durableOrderQueue() {
        return QueueBuilder
            .durable("order-queue")  // Broker restart'ta korunur
            .build();
    }

    @Bean
    public Binding orderBinding(Queue durableOrderQueue, DirectExchange durableOrdersExchange) {
        return BindingBuilder
            .bind(durableOrderQueue)
            .to(durableOrdersExchange)
            .with("new-order");
    }
}

// ===== Persistent Message Gönderme =====
@Service
@RequiredArgsConstructor
public class DurableMessagePublisher {

    private final RabbitTemplate rabbitTemplate;

    public void publishPersistentMessage(Object orderData) {
        // 3. Persistent Message
        rabbitTemplate.convertAndSend(
            "orders",
            "new-order",
            orderData,
            message -> {
                // delivery_mode = 2 (PERSISTENT)
                message.getMessageProperties()
                    .setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                return message;
            }
        );
    }
}

// ===== application.yml'de varsayılan persistent yapma =====
/*
spring:
  rabbitmq:
    template:
      default-receive-queue: order-queue
      # Varsayılan olarak tüm mesajlar persistent
      delivery-mode: PERSISTENT
*/

Publisher Confirms

Producer’ın mesajın broker’a ulaştığından emin olması için:

// ===== Publisher Confirms Konfigürasyonu =====

// application.yml
/*
spring:
  rabbitmq:
    publisher-confirm-type: correlated   # Async confirms
    publisher-returns: true               # Return callback etkinleştir
*/

@Configuration
public class PublisherConfirmsConfig {

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        template.setMandatory(true);  // Route edilemezse callback tetikle

        // Return callback - mesaj route edilemezse
        template.setReturnsCallback(returned -> {
            log.error("Mesaj route edilemedi! Exchange: {}, RoutingKey: {}, Reason: {}",
                returned.getExchange(),
                returned.getRoutingKey(),
                returned.getReplyText()
            );
        });

        return template;
    }
}

// ===== Publisher Confirms ile Mesaj Gönderme =====
@Service
@RequiredArgsConstructor
@Slf4j
public class ConfirmedPublisher {

    private final RabbitTemplate rabbitTemplate;

    public void publishWithConfirm(String message) {
        // CorrelationData ile confirm takibi
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());

        rabbitTemplate.convertAndSend(
            "orders",
            "new-order",
            message,
            correlationData
        );

        // Async callback
        correlationData.getFuture().whenComplete((confirm, ex) -> {
            if (ex != null) {
                log.error("Publish hatası: {}", ex.getMessage());
            } else if (confirm.isAck()) {
                log.info("Mesaj onaylandı: {}", correlationData.getId());
            } else {
                log.error("Mesaj reddedildi: {}", confirm.getReason());
            }
        });
    }

    // Senkron confirm (timeout ile)
    public boolean publishWithSyncConfirm(String message) {
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());

        rabbitTemplate.convertAndSend("orders", "new-order", message, correlationData);

        try {
            CorrelationData.Confirm confirm = correlationData.getFuture().get(5, TimeUnit.SECONDS);
            return confirm != null && confirm.isAck();
        } catch (Exception e) {
            log.error("Confirm timeout veya hata: {}", e.getMessage());
            return false;
        }
    }
}

Virtual Host Kavramı

Virtual Host (vhost), RabbitMQ içinde izole (ayrı) bir çalışma alanı sağlar. Her vhost, kendi içinde ayrı exchange, queue ve binding setine sahiptir; yani aynı isimli bir queue veya exchange, farklı vhost’larda birbirinden bağımsız olarak var olabilir. Faydası, tek bir RabbitMQ cluster’ı üzerinde farklı uygulama/ortamları (ör. dev–test–prod, ya da ekip/tenant bazlı) birbirine karıştırmadan yönetebilmenizdir; ayrıca yetkilendirmeyi (hangi kullanıcı hangi queue/exchange’e erişebilir) vhost üzerinden net biçimde ayırabilirsiniz.

Postane analojisiyle: Vhost, aynı şehirdeki farklı postane şubeleri gibidir. Her şubenin kendi posta kutuları (queue), dağıtım masaları (exchange) ve dağıtım kuralları (binding) vardır; bir şubedeki işler diğer şubeyi etkilemez.

VHost Kullanım Alanları

# VHost oluşturma
rabbitmqctl add_vhost /production
rabbitmqctl add_vhost /staging

# Kullanıcı izinleri
rabbitmqctl set_permissions -p /production myuser ".*" ".*" ".*"

Özet

Bu bölümde RabbitMQ’nun temel kavramlarını öğrendik:

• ✅ AMQP protokolü ve frame yapısı
• ✅ Connection, Channel, Exchange, Queue, Binding kavramları
• ✅ 4 farklı Exchange türü ve kullanım senaryoları
• ✅ Mesaj akışı ve yaşam döngüsü
• ✅ Acknowledgment modları ve durability
• ✅ Virtual Host ile izolasyon

Sonraki Bölüm

Bir sonraki bölümde Apache Kafka’nın temel kavramlarını, log-based mimarisini ve RabbitMQ’dan farklı yaklaşımını inceleyeceğiz.

***İçindekiler… « Önceki [Bölüm 1: Mesaj Sistemlerine Giriş] » Sonraki *[Bölüm 3: Apache Kafka Temelleri]