Message Queue Showdown: Kafka vs RabbitMQ vs SQS vs Pulsar vs NATS — 2025 Decision Guide

graph LR
    subgraph "Synchronous - Tight Coupling"
        A[Order Service] -->|HTTP call| B[Payment Service]
        B -->|HTTP call| C[Inventory Service]
        C -->|HTTP call| D[Notification Service]
    end

graph LR
    subgraph "Asynchronous - Loose Coupling"
        A2[Order Service] -->|Publish event| MQ[Message Queue]
        MQ -->|Subscribe| B2[Payment Service]
        MQ -->|Subscribe| C2[Inventory Service]
        MQ -->|Subscribe| D2[Notification Service]
    end

graph TB
    subgraph "Kafka Cluster"
        direction TB
        KR[KRaft Controller] --> B1[Broker 1]
        KR --> B2[Broker 2]
        KR --> B3[Broker 3]
    end
    P[Producer] --> B1
    B1 --> T["Topic: orders (3 partitions)"]
    T --> P0[Partition 0 - Leader: B1]
    T --> P1[Partition 1 - Leader: B2]
    T --> P2[Partition 2 - Leader: B3]
    P0 --> CG["Consumer Group"]
    P1 --> CG
    P2 --> CG

graph LR
    P[Producer] --> E["Exchange (type: topic)"]
    E -->|"routing.key.a"| Q1[Queue A]
    E -->|"routing.key.b"| Q2[Queue B]
    E -->|"routing.#"| Q3[Queue C - wildcard]
    Q1 --> C1[Consumer 1]
    Q2 --> C2[Consumer 2]
    Q3 --> C3[Consumer 3]

graph LR
    P[Producer] -->|SendMessage| SQS["SQS Queue"]
    SQS -->|ReceiveMessage| C1[Consumer 1]
    SQS -->|ReceiveMessage| C2[Consumer 2]
    SQS -.->|On failure| DLQ["Dead Letter Queue"]

graph TB
    subgraph "Pulsar Cluster"
        direction TB
        B1[Broker 1 - Serving] --> BK1[BookKeeper 1 - Storage]
        B2[Broker 2 - Serving] --> BK2[BookKeeper 2 - Storage]
        B3[Broker 3 - Serving] --> BK3[BookKeeper 3 - Storage]
    end
    P[Producer] --> B1
    B1 --> T["Topic: events"]
    T --> SUB["Subscription (Shared/Key_Shared/Exclusive)"]
    SUB --> C1[Consumer 1]
    SUB --> C2[Consumer 2]

graph LR
    subgraph "NATS Cluster"
        N1[NATS Server 1] --- N2[NATS Server 2]
        N2 --- N3[NATS Server 3]
        N1 --- N3
    end
    P[Publisher] --> N1
    N1 -->|"subject: sensor.temp"| S1[Subscriber 1]
    N2 -->|"subject: sensor.*"| S2[Subscriber 2 - wildcard]
    N3 -->|JetStream| JS[JetStream Consumer]

graph LR
    P[Producer] -->|XADD| RS["Redis Stream (mystream)"]
    RS -->|XREADGROUP| CG["Consumer Group"]
    CG --> C1["Consumer 1 (pending entries)"]
    CG --> C2["Consumer 2 (pending entries)"]

# Kafka throughput optimization configuration
num.partitions=12
batch.size=65536          # 64KB batch
linger.ms=5               # Wait 5ms before sending batch
compression.type=lz4      # LZ4 compression
buffer.memory=67108864    # 64MB buffer
acks=1                    # Leader-only ACK (max throughput)

# NATS Core (JetStream disabled)
# - No persistence, in-memory only
# - Message size under 10KB
# - Communication within a single datacenter

# Kafka 4.0 KRaft mode configuration
process.roles: controller,broker
node.id: 1
controller.quorum.voters: 1@kafka1:9093,2@kafka2:9093,3@kafka3:9093
listeners: PLAINTEXT://:9092,CONTROLLER://:9093
# All ZooKeeper-related configuration removed

// Kafka 4.0 Share Group example
Properties props = new Properties();
props.put("group.id", "my-share-group");
props.put("group.type", "share");  // New Share type

KafkaShareConsumer<String, String> consumer =
    new KafkaShareConsumer<>(props);
consumer.subscribe(List.of("orders"));

// Distributed per message, not bound to partitions
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        process(record);
        consumer.acknowledge(record);  // Per-message ACK
    }
}

// RabbitMQ 4.2 Stream SQL Filter
// Server-side filtering saves network bandwidth
Consumer consumer = environment.consumerBuilder()
    .stream("events")
    .filter()
        .values("region-us", "priority-high")  // Server-side filter
        .postFilter(msg -> msg.getProperties().getSubject().equals("order"))
    .builder()
    .messageHandler((context, message) -> {
        // Only filtered messages received
        processMessage(message);
        context.storeOffset();
    })
    .build();

# NATS 2.11 JetStream message TTL configuration
nats stream add EVENTS \
  --subjects "events.>" \
  --max-msg-ttl 3600 \
  --storage file \
  --replicas 3

Scenario: 100M messages/day, 1KB average, 3-day retention

Confluent Cloud (Basic):
  - Throughput: ~1,160 msgs/s average
  - Estimated cost: ~800 - 1,200 USD/month
  - Includes: Brokers, storage, network

Amazon MSK Provisioned (kafka.m5.large x 3):
  - Instances: 0.21 USD/h x 3 = ~454 USD/month
  - Storage: 300GB x 0.10 USD/GB = 30 USD
  - Estimated total: ~500 - 700 USD/month

Amazon MSK Serverless:
  - Cluster: 0.75 USD/h = ~540 USD/month
  - Partitions: 0.0015 USD/h/partition
  - Storage: 0.10 USD/GB
  - Estimated total: ~600 - 1,000 USD/month

Amazon SQS Standard:
  - First 1M requests free
  - Then: 0.40 USD / 1M requests
  - ~3B requests/month: ~1,200 USD
  - Data transfer costs extra

CloudAMQP (Dedicated - Tiger):
  - ~500 - 1,000 USD/month
  - Cluster included, monitoring included

Self-Hosted Kafka (c5.2xlarge x 3 + EBS):
  - EC2: 0.34 USD/h x 3 = ~734 USD/month
  - EBS gp3 300GB x 3: ~72 USD/month
  - Operations cost extra (at least 1 DevOps engineer)
  - Estimated total: ~800 USD/month + ops costs

// build.gradle
// implementation 'org.springframework.kafka:spring-kafka:3.3.0'

// application.yml
// spring:
//   kafka:
//     bootstrap-servers: localhost:9092
//     producer:
//       key-serializer: org.apache.kafka.common.serialization.StringSerializer
//       value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
//     consumer:
//       group-id: order-service
//       auto-offset-reset: earliest
//       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
//       value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer

// KafkaProducerService.java
@Service
@RequiredArgsConstructor
public class KafkaProducerService {

    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;

    public CompletableFuture<SendResult<String, OrderEvent>> sendOrderEvent(
            OrderEvent event) {
        return kafkaTemplate.send("orders", event.getOrderId(), event);
    }
}

// KafkaConsumerService.java
@Service
@Slf4j
public class KafkaConsumerService {

    @KafkaListener(
        topics = "orders",
        groupId = "order-service",
        containerFactory = "kafkaListenerContainerFactory"
    )
    public void handleOrderEvent(
            @Payload OrderEvent event,
            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
            @Header(KafkaHeaders.OFFSET) long offset) {

        log.info("Received order: {} from partition: {}, offset: {}",
            event.getOrderId(), partition, offset);
        processOrder(event);
    }

    private void processOrder(OrderEvent event) {
        // Order processing logic
    }
}

// build.gradle
// implementation 'org.springframework.boot:spring-boot-starter-amqp'

// application.yml
// spring:
//   rabbitmq:
//     host: localhost
//     port: 5672
//     username: guest
//     password: guest

// RabbitMQConfig.java
@Configuration
public class RabbitMQConfig {

    public static final String ORDER_EXCHANGE = "order.exchange";
    public static final String ORDER_QUEUE = "order.queue";
    public static final String ORDER_ROUTING_KEY = "order.created";
    public static final String DLQ_QUEUE = "order.dlq";

    @Bean
    public TopicExchange orderExchange() {
        return new TopicExchange(ORDER_EXCHANGE);
    }

    @Bean
    public Queue orderQueue() {
        return QueueBuilder.durable(ORDER_QUEUE)
            .withArgument("x-dead-letter-exchange", "")
            .withArgument("x-dead-letter-routing-key", DLQ_QUEUE)
            .withArgument("x-message-ttl", 60000)
            .build();
    }

    @Bean
    public Queue deadLetterQueue() {
        return QueueBuilder.durable(DLQ_QUEUE).build();
    }

    @Bean
    public Binding orderBinding(Queue orderQueue, TopicExchange orderExchange) {
        return BindingBuilder.bind(orderQueue)
            .to(orderExchange)
            .with(ORDER_ROUTING_KEY);
    }
}

// RabbitMQProducer.java
@Service
@RequiredArgsConstructor
public class RabbitMQProducer {

    private final RabbitTemplate rabbitTemplate;

    public void sendOrderEvent(OrderEvent event) {
        rabbitTemplate.convertAndSend(
            RabbitMQConfig.ORDER_EXCHANGE,
            RabbitMQConfig.ORDER_ROUTING_KEY,
            event,
            message -> {
                message.getMessageProperties().setContentType("application/json");
                message.getMessageProperties().setDeliveryMode(
                    MessageDeliveryMode.PERSISTENT);
                return message;
            }
        );
    }
}

// RabbitMQConsumer.java
@Service
@Slf4j
public class RabbitMQConsumer {

    @RabbitListener(queues = RabbitMQConfig.ORDER_QUEUE)
    public void handleOrderEvent(OrderEvent event, Channel channel,
            @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
        try {
            log.info("Received order: {}", event.getOrderId());
            processOrder(event);
            channel.basicAck(tag, false);
        } catch (Exception e) {
            log.error("Failed to process order: {}", event.getOrderId(), e);
            channel.basicNack(tag, false, false); // Move to DLQ
        }
    }
}

// build.gradle
// implementation 'io.awspring.cloud:spring-cloud-aws-starter-sqs:3.2.0'

// application.yml
// spring:
//   cloud:
//     aws:
//       region:
//         static: ap-northeast-2
//       sqs:
//         endpoint: https://sqs.ap-northeast-2.amazonaws.com

// SQSProducer.java
@Service
@RequiredArgsConstructor
public class SQSProducer {

    private final SqsTemplate sqsTemplate;

    public void sendOrderEvent(OrderEvent event) {
        sqsTemplate.send(to -> to
            .queue("order-queue")
            .payload(event)
            .header("eventType", "ORDER_CREATED")
        );
    }

    // FIFO queue sending
    public void sendOrderEventFifo(OrderEvent event) {
        sqsTemplate.send(to -> to
            .queue("order-queue.fifo")
            .payload(event)
            .header(SqsHeaders.MessageSystemAttributes.SQS_MESSAGE_GROUP_ID_HEADER,
                event.getOrderId())
            .header(SqsHeaders.MessageSystemAttributes
                .SQS_MESSAGE_DEDUPLICATION_ID_HEADER,
                UUID.randomUUID().toString())
        );
    }
}

// SQSConsumer.java
@Service
@Slf4j
public class SQSConsumer {

    @SqsListener(value = "order-queue", maxConcurrentMessages = "10",
        maxMessagesPerPoll = "5")
    public void handleOrderEvent(@Payload OrderEvent event,
            @Header("eventType") String eventType) {
        log.info("Received order: {} with type: {}",
            event.getOrderId(), eventType);
        processOrder(event);
    }
}

// build.gradle
// implementation 'org.springframework.pulsar:spring-pulsar-spring-boot-starter:1.2.0'

// application.yml
// spring:
//   pulsar:
//     client:
//       service-url: pulsar://localhost:6650
//     producer:
//       topic-name: persistent://public/default/orders
//     consumer:
//       subscription-name: order-service-sub
//       subscription-type: shared

// PulsarProducer.java
@Service
@RequiredArgsConstructor
public class PulsarProducer {

    private final PulsarTemplate<OrderEvent> pulsarTemplate;

    public void sendOrderEvent(OrderEvent event) {
        pulsarTemplate.send("persistent://public/default/orders", event);
    }
}

// PulsarConsumer.java
@Service
@Slf4j
public class PulsarConsumer {

    @PulsarListener(
        topics = "persistent://public/default/orders",
        subscriptionName = "order-service-sub",
        subscriptionType = SubscriptionType.Shared,
        schemaType = SchemaType.JSON
    )
    public void handleOrderEvent(OrderEvent event) {
        log.info("Received order: {}", event.getOrderId());
        processOrder(event);
    }
}

// build.gradle
// implementation 'io.nats:jnats:2.20.4'

// NATSConfig.java
@Configuration
public class NATSConfig {

    @Bean
    public Connection natsConnection() throws IOException, InterruptedException {
        Options options = new Options.Builder()
            .server("nats://localhost:4222")
            .reconnectWait(Duration.ofSeconds(2))
            .maxReconnects(-1) // Unlimited reconnects
            .connectionListener((conn, type) -> {
                System.out.println("NATS connection event: " + type);
            })
            .build();
        return Nats.connect(options);
    }

    @Bean
    public JetStream jetStream(Connection connection) throws IOException {
        return connection.jetStream();
    }
}

// NATSProducer.java
@Service
@RequiredArgsConstructor
public class NATSProducer {

    private final JetStream jetStream;
    private final ObjectMapper objectMapper;

    public PublishAck sendOrderEvent(OrderEvent event) throws Exception {
        byte[] data = objectMapper.writeValueAsBytes(event);
        Message msg = NatsMessage.builder()
            .subject("orders.created")
            .data(data)
            .build();
        return jetStream.publish(msg);
    }
}

// NATSConsumer.java
@Service
@Slf4j
public class NATSConsumer {

    @PostConstruct
    public void startConsumer() throws Exception {
        JetStream js = natsConnection.jetStream();

        PushSubscribeOptions options = PushSubscribeOptions.builder()
            .durable("order-consumer")
            .build();

        js.subscribe("orders.>", "order-queue", dispatcher -> {
            dispatcher.onMessage(msg -> {
                try {
                    OrderEvent event = objectMapper.readValue(
                        msg.getData(), OrderEvent.class);
                    log.info("Received order: {}", event.getOrderId());
                    processOrder(event);
                    msg.ack();
                } catch (Exception e) {
                    log.error("Failed to process message", e);
                    msg.nak();
                }
            });
        }, false, options);
    }
}

// build.gradle
// implementation 'org.springframework.boot:spring-boot-starter-data-redis'

// RedisStreamConfig.java
@Configuration
public class RedisStreamConfig {

    @Bean
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>>
            streamListenerContainer(RedisConnectionFactory factory) {

        var options = StreamMessageListenerContainer
            .StreamMessageListenerContainerOptions.builder()
            .pollTimeout(Duration.ofSeconds(1))
            .batchSize(10)
            .targetType(MapRecord.class)
            .build();

        var container = StreamMessageListenerContainer.create(factory, options);

        container.receiveAutoAck(
            Consumer.from("order-group", "consumer-1"),
            StreamOffset.create("orders", ReadOffset.lastConsumed()),
            new OrderStreamListener()
        );

        container.start();
        return container;
    }
}

// RedisStreamProducer.java
@Service
@RequiredArgsConstructor
public class RedisStreamProducer {

    private final StringRedisTemplate redisTemplate;
    private final ObjectMapper objectMapper;

    public RecordId sendOrderEvent(OrderEvent event) throws Exception {
        Map<String, String> fields = Map.of(
            "orderId", event.getOrderId(),
            "type", "ORDER_CREATED",
            "payload", objectMapper.writeValueAsString(event)
        );
        return redisTemplate.opsForStream()
            .add(StreamRecords.newRecord()
                .ofMap(fields)
                .withStreamKey("orders"));
    }
}

// OrderStreamListener.java
@Slf4j
public class OrderStreamListener
        implements StreamListener<String, MapRecord<String, String, String>> {

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        Map<String, String> body = message.getValue();
        log.info("Received order: {} from stream: {}",
            body.get("orderId"), message.getStream());
        processOrder(body);
    }
}

graph TD
    START[Choose a Message Queue] --> Q1["Need event replay?"]

    Q1 -->|Yes| Q1A["Need multi-tenancy or geo-replication?"]
    Q1A -->|Yes| PULSAR["Apache Pulsar"]
    Q1A -->|No| KAFKA["Apache Kafka"]

    Q1 -->|No| Q2["Need complex routing?"]
    Q2 -->|Yes| RABBIT["RabbitMQ"]

    Q2 -->|No| Q3["AWS-native serverless?"]
    Q3 -->|Yes| SQS["Amazon SQS"]

    Q3 -->|No| Q4["Need sub-ms ultra-low latency?"]
    Q4 -->|Yes| Q4A["Need persistence?"]
    Q4A -->|Yes| NATS["NATS + JetStream"]
    Q4A -->|No| NATS_CORE["NATS Core"]

    Q4 -->|No| Q5["Already using Redis?"]
    Q5 -->|Yes| REDIS["Redis Streams"]
    Q5 -->|No| KAFKA2["Kafka (default choice)"]

graph LR
    subgraph "Order Creation"
        API[API Gateway] --> OS[Order Service]
    end

    OS -->|OrderCreated| KAFKA[Kafka - orders topic]

    KAFKA --> PAY[Payment Service]
    KAFKA --> INV[Inventory Service]
    KAFKA --> NOTIF[Notification Service]

    PAY -->|PaymentCompleted| KAFKA2[Kafka - payments topic]
    KAFKA2 --> SHIP[Shipping Service]

    INV -->|StockReserved| KAFKA3[Kafka - inventory topic]
    KAFKA3 --> OS2[Order Service - status update]

    SHIP -->|ShipmentCreated| KAFKA4[Kafka - shipments topic]
    KAFKA4 --> NOTIF2[Notification Service - shipping alerts]

# Order topic configuration
topics:
  orders:
    partitions: 12
    replication-factor: 3
    configs:
      retention.ms: 604800000 # 7 days
      min.insync.replicas: 2
      cleanup.policy: delete

  payments:
    partitions: 6
    replication-factor: 3
    configs:
      retention.ms: 2592000000 # 30 days (audit log)
      min.insync.replicas: 2

graph LR
    subgraph "Data Collection"
        WEB[Web Clickstream] --> KAFKA[Kafka]
        APP[App Events] --> KAFKA
        IOT[IoT Sensors] --> NATS[NATS]
        NATS -->|Bridge| KAFKA
    end

    subgraph "Stream Processing"
        KAFKA --> FLINK[Apache Flink]
        KAFKA --> KSQL[ksqlDB]
    end

    subgraph "Storage and Serving"
        FLINK --> ES[Elasticsearch]
        FLINK --> CH[ClickHouse]
        KSQL --> REDIS[Redis Cache]
    end

    subgraph "Visualization"
        ES --> GRAFANA[Grafana]
        CH --> SUPERSET[Apache Superset]
    end

graph TB
    subgraph "Edge Layer"
        S1[Temperature Sensor] --> LN1[NATS Leaf Node - Factory A]
        S2[Humidity Sensor] --> LN1
        S3[Vibration Sensor] --> LN2[NATS Leaf Node - Factory B]
    end

    subgraph "Core Layer"
        LN1 --> NATS[NATS Cluster - JetStream]
        LN2 --> NATS
        NATS --> PROC[Stream Processor]
        PROC -->|Anomaly detection| ALERT[Alert Service - RabbitMQ]
        PROC -->|Aggregated data| TS[TimescaleDB]
    end

    ALERT -->|Email| EMAIL[Email Service]
    ALERT -->|Slack| SLACK[Slack Notification]

sequenceDiagram
    participant OS as Order Service
    participant K as Kafka
    participant PS as Payment Service
    participant IS as Inventory Service
    participant SS as Shipping Service

    OS->>K: OrderCreated
    K->>PS: Payment request
    PS->>K: PaymentCompleted
    K->>IS: Deduct inventory
    IS->>K: StockReserved
    K->>SS: Create shipment
    SS->>K: ShipmentCreated
    K->>OS: Order complete

    Note over PS,IS: Compensating transactions on failure
    PS-->>K: PaymentFailed (compensate)
    K-->>IS: Restore inventory (compensate)
    K-->>OS: Cancel order (compensate)

1. Broker Configuration
   - min.insync.replicas = 2 (with replication-factor 3)
   - unclean.leader.election.enable = false
   - auto.create.topics.enable = false

2. Producer Configuration
   - acks = all (prevent data loss)
   - retries = Integer.MAX_VALUE
   - enable.idempotence = true
   - max.in.flight.requests.per.connection = 5

3. Consumer Configuration
   - enable.auto.commit = false (manual commits)
   - max.poll.records = 500
   - session.timeout.ms = 30000

4. Monitoring
   - Prometheus + Grafana (JMX Exporter)
   - Consumer Lag alerts
   - Under-Replicated Partitions alerts

1. Cluster Configuration
   - Use Quorum Queues (replaces Classic Mirrored Queues)
   - vm_memory_high_watermark = 0.4
   - disk_free_limit = 2GB

2. Connection Management
   - Use Connection Pools
   - Heartbeat timeout: 60 seconds
   - Reuse channels (channel creation is expensive)

3. Message Configuration
   - Persistent delivery mode
   - Enable Publisher Confirms
   - Configure Dead Letter Exchange

4. Monitoring
   - Management Plugin Dashboard
   - rabbitmq_prometheus plugin
   - Queue depth alerts