Anypoint MQ ile Yoğun Trafik Nasıl Yönetilir?

Yüksek trafikli sistemlerde senkron HTTP çağrıları bir noktada yetersiz kalır — timeout'lar artar, consumer'lar çöker ve veri kaybolur. Anypoint MQ, MuleSoft'un yönetilen mesajlaşma servisi olarak bu sorunu asenkron mimariye geçerek çözer. Bu yazıda gerçek kod örnekleri ve mimari diyagramlarla Anypoint MQ'nun yoğun trafik altında nasıl çalıştığını inceliyoruz.

Neden Asenkron Mesajlaşma? Senkron vs MQ Karşılaştırması

Senkron HTTP entegrasyonunda her istek, consumer'ın yanıt vermesini bekler. 1000 eş zamanlı istek geldiğinde thread pool tükenir, timeout'lar zincir halinde yayılır. Anypoint MQ bu problemi producer-consumer ayrışmasıyla çözer.

Kriter	Senkron HTTP	Anypoint MQ (Asenkron)
Yük altında davranış	Timeout & hata	Mesajlar queue'da birikir
Producer-Consumer bağımlılığı	Sıkı bağlı (tight coupling)	Gevşek bağlı (loose coupling)
Hata durumunda veri kaybı	Yüksek risk	DLQ ile korunur
Ölçeklendirme	Dikey ölçek gerekir	Consumer sayısı artırılır
Gecikme (latency)	Düşük (ms)	Orta (saniye mertebesi)

Ne zaman MQ kullanmalısınız? Yanıtın anında gelmesi kritik değilse, consumer yavaşsa veya yük dengesizse — MQ doğru seçimdir. Anlık kullanıcı yanıtı gereken senaryolarda senkron tercih edilebilir.

Producer Flow: Mesaj Yayınlama

Producer tarafında Anypoint MQ Publish bileşeni kullanılır. Mesaj gövdesi, correlation ID ve öncelik gibi özellikler header'larla iletilir.

Mule 4 — XML Flowproducer-flow.xml

<!-- Producer Flow: HTTP → Anypoint MQ -->
<flow name="order-producer-flow">

  <http:listener
    path="/orders"
    allowedMethods="POST"
    config-ref="HTTP_Listener_config"/>

  <!-- Mesajı doğrula -->
  <validation:is-not-null
    value="#[payload.orderId]"
    message="orderId zorunludur"/>

  <!-- MQ'ya yayınla -->
  <anypoint-mq:publish
    config-ref="Anypoint_MQ_Config"
    destination="orders-exchange"
    messageId=#[payload.orderId]>
    <anypoint-mq:properties>
      <anypoint-mq:property
        key="priority"
        value=#[payload.priority default 'NORMAL']/>
      <anypoint-mq:property
        key="correlationId"
        value=#[correlationId]/>
    </anypoint-mq:properties>
  </anypoint-mq:publish>

  <!-- Hemen 202 Accepted dön -->
  <set-payload
    value=#[output application/json --- {status: "queued", messageId: payload.orderId}]/>
  <http:response statusCode="202"/>

</flow>

İpucu: Producer'ın hemen 202 Accepted dönmesi kritiktir. Consumer ne kadar yavaş işlerse işlesin, client beklemez ve timeout yaşamaz.

Consumer Flow: Mesaj Tüketme ve İdempotency

Consumer tarafında anypoint-mq:subscriber bileşeni kullanılır. Yüksek trafik senaryolarında aynı mesajın iki kez işlenmemesi için Object Store ile idempotency kontrolü şarttır.

Mule 4 — XML Flowconsumer-flow.xml

<flow name="order-consumer-flow">

  <anypoint-mq:subscriber
    config-ref="Anypoint_MQ_Config"
    destination="orders-queue"
    maxConcurrency="10"
    ackMode="MANUAL"/>

  <!-- İdempotency: aynı mesajı iki kez işleme -->
  <os:retrieve
    config-ref="ObjectStore_Config"
    key=#[attributes.messageId]
    target="alreadyProcessed"
    defaultValue="false"/>

  <choice>
    <when expression=#[vars.alreadyProcessed == false]>
      <flow-ref name="process-order-subflow"/>
      <os:store
        config-ref="ObjectStore_Config"
        key=#[attributes.messageId]
        value="true"
        ttl="86400"
        ttlUnit="SECONDS"/>
      <anypoint-mq:ack ackToken=#[attributes.ackToken]/>
    </when>
    <otherwise>
      <!-- Duplicate: sadece ACK, işleme -->
      <logger message="Duplicate mesaj atlandı: #[attributes.messageId]"/>
      <anypoint-mq:ack ackToken=#[attributes.ackToken]/>
    </otherwise>
  </choice>

</flow>

✓maxConcurrency="10" — aynı anda 10 mesaj paralel işlenir, yük dengelenir
✓ackMode="MANUAL" — işlem başarılı olunca ACK gönderilir, hata olursa NACK ile queue'ya döner
✓Object Store TTL — 24 saat sonra idempotency kaydı silinir, hafıza şişmez

Hata Yönetimi: Dead Letter Queue (DLQ)

Yoğun trafikte bazı mesajlar işlenemez — bağlantı kopması, veri hatası veya downstream sistem arızası. DLQ bu mesajları kaybetmeden yakalar ve retry mekanizmasıyla yeniden işleme alır.

Mule 4 — Hata Yönetimierror-handler.xml

<flow name="order-consumer-flow">

  <anypoint-mq:subscriber
    destination="orders-queue"
    ackMode="MANUAL"/>

  <try>
    <flow-ref name="process-order-subflow"/>
    <anypoint-mq:ack ackToken=#[attributes.ackToken]/>

    <error-handler>
      <!-- Geçici hata: NACK, queue'ya döner -->
      <on-error-continue type="CONNECTIVITY, TIMEOUT">
        <logger message="Geçici hata, NACK: #[error.description]" level="WARN"/>
        <anypoint-mq:nack ackToken=#[attributes.ackToken]/>
      </on-error-continue>

      <!-- Kalıcı hata: direkt DLQ'ya gönder -->
      <on-error-continue type="VALIDATION, TRANSFORMATION">
        <anypoint-mq:publish destination="orders-dlq">
          <anypoint-mq:properties>
            <anypoint-mq:property key="errorReason" value=#[error.description]/>
            <anypoint-mq:property key="originalMessageId" value=#[attributes.messageId]/>
          </anypoint-mq:properties>
        </anypoint-mq:publish>
        <anypoint-mq:ack ackToken=#[attributes.ackToken]/>
      </on-error-continue>
    </error-handler>
  </try>

</flow>

Dikkat: DLQ'daki mesajları düzenli izleyin. Anypoint Monitoring'de DLQ derinliği için alert kurulması, operasyonel olgunluğun temel göstergesidir.

DataWeave ile Mesaj Dönüşümü

MQ üzerinden gelen mesajlar çoğunlukla farklı sistemler arasında format dönüşümü gerektirir. DataWeave bu dönüşümü consumer flow içinde yapmanın en temiz yoludur.

DataWeave 2.0order-transform.dwl

// MQ'dan gelen sipariş → SAP formatına dönüşüm
%dw 2.0
output application/xml

var priorityMap = {
  "HIGH":   "01",
  "NORMAL": "02",
  "LOW":    "03"
}
---
SALESORDER: {
  HEADER: {
    ORDER_ID:    payload.orderId,
    CUSTOMER:   payload.customerId,
    PRIORITY:   priorityMap[attributes.properties."priority" default "NORMAL"],
    CREATED_AT: now() as String { format: "yyyyMMddHHmmss" },
    EXT_REF:    attributes.properties."correlationId" default ""
  },
  ITEMS: {
    (payload.items map (item, idx) -> {
      ITEM: {
        LINE_NO:  (idx + 1) * 10,
        MATERIAL: item.sku,
        QTY:      item.quantity,
        UNIT:     item.unit default "EA"
      }
    })
  }
}

✓MQ attributes.properties üzerinden header değerlerine erişilir — payload'ı kirletmez
✓default operatörü null-safe mapping sağlar — hatalı mesajlarda NullPointerException olmaz
✓SAP tarih formatı yyyyMMddHHmmss — DataWeave'de as String {format: ...} ile dönüştürülür

Performans Optimizasyonu: Hangi Ayarlar Fark Yaratır?

Anypoint MQ subscriber'ında doğru konfigürasyon, throughput'u dramatik biçimde etkiler. Aşağıdaki parametreler üretim ortamı için kritiktir.

Mule 4 — Subscriber Konfigürasyonuoptimized-subscriber.xml

<anypoint-mq:subscriber
  config-ref="Anypoint_MQ_Config"
  destination="orders-queue"
  <!-- Kaç mesaj paralel işlensin -->
  maxConcurrency="20"
  <!-- Tek polling'de kaç mesaj çekilsin (1-10) -->
  fetchSize="10"
  <!-- Queue boşsa ne kadar beklensin (ms) -->
  pollingTime="1000"
  <!-- Mesaj işleme timeout'u -->
  acknowledgementTimeout="60000"
  <!-- Manuel ACK modu -->
  ackMode="MANUAL"/>

Parametre	Düşük Trafik	Yüksek Trafik	Açıklama
maxConcurrency	5	20–50	CloudHub worker vCPU'suna göre ayarlanır
fetchSize	1–3	10	Tek seferde çekilen mesaj sayısı
pollingTime	5000 ms	500–1000 ms	Düşük tutunca latency azalır, maliyet artar
ackTimeout	30 sn	60–120 sn	İşlem süresi + buffer ekleyin

Eltay deneyimi: Üretim sektöründeki bir müşterimizde maxConcurrency'yi 5'ten 20'ye çıkararak ve fetchSize'ı 10 yaparak aynı CloudHub worker'da throughput'u 4x artırdık. Worker upgrade gerekmedi.

Paylaş

Anypoint MQ ile Yoğun Trafik Nasıl Yönetilir?

Neden Asenkron Mesajlaşma? Senkron vs MQ Karşılaştırması

Producer Flow: Mesaj Yayınlama

Consumer Flow: Mesaj Tüketme ve İdempotency

Hata Yönetimi: Dead Letter Queue (DLQ)

DataWeave ile Mesaj Dönüşümü

Performans Optimizasyonu: Hangi Ayarlar Fark Yaratır?

İlgili yazılar

Salesforce ile Gerçek Zamanlı Entegrasyon: Salesforce CDC ve Anypoint MQ ile Event-Driven Pratikleri

MuleSoft ve MCP: AI Asistanlarına Kurumsal API'lerin Kapısını Açmak

MuleSoft Database Connector'da Connection Pool: Neden Bu Kadar Önemli?

MuleSoft Yolculuğunuza Doğru İş Ortağıyla Başlayın