メッセージキュー完全比較：Kafka vs RabbitMQ vs SQS vs Pulsar vs NATS — 2025年実践選択ガイド

graph LR
    subgraph "同期方式 - 強い結合"
        A[注文サービス] -->|HTTP呼出| B[決済サービス]
        B -->|HTTP呼出| C[在庫サービス]
        C -->|HTTP呼出| D[通知サービス]
    end

graph LR
    subgraph "非同期方式 - 緩い結合"
        A2[注文サービス] -->|イベント発行| MQ[メッセージキュー]
        MQ -->|購読| B2[決済サービス]
        MQ -->|購読| C2[在庫サービス]
        MQ -->|購読| D2[通知サービス]
    end

graph TB
    subgraph "Kafkaクラスタ"
        direction TB
        KR[KRaftコントローラー] --> B1[Broker 1]
        KR --> B2[Broker 2]
        KR --> B3[Broker 3]
    end
    P[Producer] --> B1
    B1 --> T["Topic: orders（3パーティション）"]
    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 - ワイルドカード]
    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 -.->|失敗時| DLQ["Dead Letter Queue"]

graph TB
    subgraph "Pulsarクラスタ"
        direction TB
        B1[Broker 1 - サービング] --> BK1[BookKeeper 1 - ストレージ]
        B2[Broker 2 - サービング] --> BK2[BookKeeper 2 - ストレージ]
        B3[Broker 3 - サービング] --> BK3[BookKeeper 3 - ストレージ]
    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クラスタ"
        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 - ワイルドカード]
    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スループット最適化設定
num.partitions=12
batch.size=65536          # 64KBバッチ
linger.ms=5               # 5ms待機後バッチ送信
compression.type=lz4      # LZ4圧縮
buffer.memory=67108864    # 64MBバッファ
acks=1                    # Leaderのみ確認（最大スループット）

# NATS Core（JetStream無効化）
# - 永続性なし、インメモリのみ
# - 10KB以下のメッセージサイズ
# - 単一データセンター内の通信

# Kafka 4.0 KRaftモード設定
process.roles: controller,broker
node.id: 1
controller.quorum.voters: 1@kafka1:9093,2@kafka2:9093,3@kafka3:9093
listeners: PLAINTEXT://:9092,CONTROLLER://:9093
# ZooKeeper関連設定は完全除去

// Kafka 4.0 Share Group使用例
Properties props = new Properties();
props.put("group.id", "my-share-group");
props.put("group.type", "share");  // 新しいShareタイプ

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

// メッセージ単位で分配 - パーティションに依存しない
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        process(record);
        consumer.acknowledge(record);  // 個別メッセージACK
    }
}

// RabbitMQ 4.2 Stream SQLフィルター
// サーバーサイドでフィルタリング - ネットワーク帯域幅を節約
Consumer consumer = environment.consumerBuilder()
    .stream("events")
    .filter()
        .values("region-us", "priority-high")  // サーバーサイドフィルター
        .postFilter(msg -> msg.getProperties().getSubject().equals("order"))
    .builder()
    .messageHandler((context, message) -> {
        // フィルタリングされたメッセージのみ受信
        processMessage(message);
        context.storeOffset();
    })
    .build();

# NATS 2.11 JetStreamメッセージTTL設定
nats stream add EVENTS \
  --subjects "events.>" \
  --max-msg-ttl 3600 \
  --storage file \
  --replicas 3

シナリオ：1日1億メッセージ、平均1KB、3日間保持

Confluent Cloud（Basic）：
  - スループット：平均約1,160 msgs/s
  - 予想コスト：月額約800～1,200 USD
  - 含む：ブローカー、ストレージ、ネットワーク

Amazon MSK Provisioned（kafka.m5.large x 3）：
  - インスタンス：0.21 USD/h x 3 = 月額約454 USD
  - ストレージ：300GB x 0.10 USD/GB = 30 USD
  - 予想総コスト：月額約500～700 USD

Amazon MSK Serverless：
  - クラスタ：0.75 USD/h = 月額約540 USD
  - パーティション：0.0015 USD/h/partition
  - ストレージ：0.10 USD/GB
  - 予想総コスト：月額約600～1,000 USD

Amazon SQS Standard：
  - 最初の100万件は無料
  - 以降：0.40 USD / 100万件
  - 月約30億件：約1,200 USD
  - データ転送コスト別途

CloudAMQP（Dedicated - Tiger）：
  - 月額約500～1,000 USD
  - クラスタ含む、モニタリング含む

自己ホスティングKafka（c5.2xlarge x 3 + EBS）：
  - EC2：0.34 USD/h x 3 = 月額約734 USD
  - EBS gp3 300GB x 3：月額約72 USD
  - 運用人件費別途（DevOpsエンジニア1名）
  - 予想総コスト：月額約800 USD + 運用費

// 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) {
        // 注文処理ロジック
    }
}

// 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); // DLQに移動
        }
    }
}

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

// application.yml
// spring:
//   cloud:
//     aws:
//       region:
//         static: ap-northeast-1
//       sqs:
//         endpoint: https://sqs.ap-northeast-1.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キュー送信
    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) // 無制限再接続
            .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[メッセージキューの選択] --> Q1["イベントリプレイが必要か？"]

    Q1 -->|はい| Q1A["マルチテナンシーまたはGeo複製が必要か？"]
    Q1A -->|はい| PULSAR["Apache Pulsar"]
    Q1A -->|いいえ| KAFKA["Apache Kafka"]

    Q1 -->|いいえ| Q2["複雑なルーティングが必要か？"]
    Q2 -->|はい| RABBIT["RabbitMQ"]

    Q2 -->|いいえ| Q3["AWSネイティブサーバーレスか？"]
    Q3 -->|はい| SQS["Amazon SQS"]

    Q3 -->|いいえ| Q4["サブミリ秒の超低レイテンシーが必須か？"]
    Q4 -->|はい| Q4A["永続性が必要か？"]
    Q4A -->|はい| NATS["NATS + JetStream"]
    Q4A -->|いいえ| NATS_CORE["NATS Core"]

    Q4 -->|いいえ| Q5["すでにRedisを使用中か？"]
    Q5 -->|はい| REDIS["Redis Streams"]
    Q5 -->|いいえ| KAFKA2["Kafka（デフォルト選択）"]

graph LR
    subgraph "注文作成"
        API[API Gateway] --> OS[Order Service]
    end

    OS -->|OrderCreated| KAFKA[Kafka - ordersトピック]

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

    PAY -->|PaymentCompleted| KAFKA2[Kafka - paymentsトピック]
    KAFKA2 --> SHIP[Shipping Service]

    INV -->|StockReserved| KAFKA3[Kafka - inventoryトピック]
    KAFKA3 --> OS2[Order Service - ステータス更新]

    SHIP -->|ShipmentCreated| KAFKA4[Kafka - shipmentsトピック]
    KAFKA4 --> NOTIF2[Notification Service - 配送通知]

# 注文トピック設定
topics:
  orders:
    partitions: 12
    replication-factor: 3
    configs:
      retention.ms: 604800000 # 7日
      min.insync.replicas: 2
      cleanup.policy: delete

  payments:
    partitions: 6
    replication-factor: 3
    configs:
      retention.ms: 2592000000 # 30日（監査ログ）
      min.insync.replicas: 2

graph LR
    subgraph "データ収集"
        WEB[Webクリックストリーム] --> KAFKA[Kafka]
        APP[アプリイベント] --> KAFKA
        IOT[IoTセンサー] --> NATS[NATS]
        NATS -->|ブリッジ| KAFKA
    end

    subgraph "ストリーム処理"
        KAFKA --> FLINK[Apache Flink]
        KAFKA --> KSQL[ksqlDB]
    end

    subgraph "保存とサービング"
        FLINK --> ES[Elasticsearch]
        FLINK --> CH[ClickHouse]
        KSQL --> REDIS[Redis Cache]
    end

    subgraph "可視化"
        ES --> GRAFANA[Grafana]
        CH --> SUPERSET[Apache Superset]
    end

graph TB
    subgraph "エッジレイヤー"
        S1[温度センサー] --> LN1[NATS Leaf Node - 工場A]
        S2[湿度センサー] --> LN1
        S3[振動センサー] --> LN2[NATS Leaf Node - 工場B]
    end

    subgraph "コアレイヤー"
        LN1 --> NATS[NATSクラスタ - JetStream]
        LN2 --> NATS
        NATS --> PROC[Stream Processor]
        PROC -->|異常検知| ALERT[Alert Service - RabbitMQ]
        PROC -->|集約データ| TS[TimescaleDB]
    end

    ALERT -->|メール| EMAIL[Email Service]
    ALERT -->|Slack| SLACK[Slack通知]

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: 決済リクエスト
    PS->>K: PaymentCompleted
    K->>IS: 在庫差引
    IS->>K: StockReserved
    K->>SS: 配送作成
    SS->>K: ShipmentCreated
    K->>OS: 注文完了

    Note over PS,IS: 失敗時の補償トランザクション
    PS-->>K: PaymentFailed（補償）
    K-->>IS: 在庫復旧（補償）
    K-->>OS: 注文キャンセル（補償）

1. Broker設定
   - min.insync.replicas = 2（replication-factor 3基準）
   - unclean.leader.election.enable = false
   - auto.create.topics.enable = false

2. Producer設定
   - acks = all（データ損失防止）
   - retries = Integer.MAX_VALUE
   - enable.idempotence = true
   - max.in.flight.requests.per.connection = 5

3. Consumer設定
   - enable.auto.commit = false（手動コミット）
   - max.poll.records = 500
   - session.timeout.ms = 30000

4. モニタリング
   - Prometheus + Grafana（JMX Exporter）
   - Consumer Lagアラート設定
   - Under-Replicated Partitionsアラート

1. クラスタ設定
   - Quorum Queueを使用（Classic Mirrored Queueの代替）
   - vm_memory_high_watermark = 0.4
   - disk_free_limit = 2GB

2. 接続管理
   - Connection Poolを使用
   - Heartbeatタイムアウト：60秒
   - チャネル再利用（チャネル作成コストが高い）

3. メッセージ設定
   - Persistent delivery mode
   - Publisher Confirmsを有効化
   - Dead Letter Exchangeを設定

4. モニタリング
   - Management Pluginダッシュボード
   - rabbitmq_prometheusプラグイン
   - Queue深度アラート設定