Event-Driven Architecture: Kafka vs RabbitMQ vs AWS SNS/SQS

Как построить масштабируемую систему с асинхронной коммуникацией между сервисами? В этом руководстве мы разберем Event-Driven Architecture и сравним три популярных решения: Apache Kafka, RabbitMQ и AWS SNS/SQS. Вы узнаете про throughput, latency, ordering, Event Sourcing, CQRS и примеры реализации.

📡 Что такое Event-Driven Architecture?

Event-Driven Architecture (EDA) — это архитектурный паттерн где компоненты системы взаимодействуют через события (events). Producers публикуют события, Consumers подписываются и реагируют асинхронно.

Producer (Order Service)
    ↓
Publish Event: "OrderCreated"
    ↓
Event Bus (Kafka / RabbitMQ / SNS)
    ↓
    ├─→ Consumer 1 (Payment Service) → Process payment
    ├─→ Consumer 2 (Inventory Service) → Reserve stock
    ├─→ Consumer 3 (Email Service) → Send confirmation
    └─→ Consumer 4 (Analytics Service) → Track metrics

Асинхронная обработка, каждый Consumer независим

Request-Response vs Event-Driven

🌊 Apache Kafka: Distributed Streaming Platform

Apache Kafka — это distributed event streaming platform, изначально созданный LinkedIn. Kafka — это не просто message broker, это distributed commit log.

Topic: "orders" (3 partitions, replication factor 2)

Partition 0:  [msg1] [msg2] [msg3] [msg4] ...
              ↓ replicated to Broker 2
Partition 1:  [msg5] [msg6] [msg7] [msg8] ...
              ↓ replicated to Broker 3
Partition 2:  [msg9] [msg10] [msg11] [msg12] ...
              ↓ replicated to Broker 1

Consumer Group "payment-service" (3 consumers):
  Consumer 1 → reads Partition 0
  Consumer 2 → reads Partition 1
  Consumer 3 → reads Partition 2

Parallel processing, каждый consumer читает свою partition

Kafka: Преимущества

Kafka: Недостатки

Kafka Producer (Node.js)

// kafka-producer.js
const { Kafka } = require('kafkajs');

const kafka = new Kafka({
  clientId: 'order-service',
  brokers: ['localhost:9092', 'localhost:9093', 'localhost:9094']
});

const producer = kafka.producer();

async function publishOrderCreated(order) {
  await producer.connect();

  // Publish event to "orders" topic
  await producer.send({
    topic: 'orders',
    messages: [
      {
        key: order.id.toString(),  // Key для partitioning
        value: JSON.stringify({
          eventType: 'OrderCreated',
          orderId: order.id,
          userId: order.userId,
          items: order.items,
          total: order.total,
          timestamp: new Date().toISOString()
        }),
        headers: {
          'correlation-id': order.correlationId
        }
      }
    ]
  });

  console.log(`✅ Event published: OrderCreated ${order.id}`);
}

// Graceful shutdown
process.on('SIGTERM', async () => {
  await producer.disconnect();
});

module.exports = { publishOrderCreated };

Kafka Consumer (Node.js)

// kafka-consumer.js
const { Kafka } = require('kafkajs');

const kafka = new Kafka({
  clientId: 'payment-service',
  brokers: ['localhost:9092', 'localhost:9093', 'localhost:9094']
});

const consumer = kafka.consumer({
  groupId: 'payment-service-group',
  // Retry configuration
  retry: {
    initialRetryTime: 100,
    retries: 8
  }
});

async function startConsumer() {
  await consumer.connect();

  // Subscribe to "orders" topic
  await consumer.subscribe({
    topic: 'orders',
    fromBeginning: false  // Start from latest offset
  });

  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      const event = JSON.parse(message.value.toString());

      console.log(`📨 Received event: ${event.eventType} from partition ${partition}`);

      try {
        if (event.eventType === 'OrderCreated') {
          await processPayment(event);
        }

        // Commit offset after successful processing
        // (auto-commit is enabled by default)

      } catch (error) {
        console.error(`❌ Error processing event: ${error.message}`);
        // Event will be retried or sent to DLQ
        throw error;
      }
    }
  });
}

async function processPayment(event) {
  console.log(`💳 Processing payment for order ${event.orderId}`);

  // Reserve payment
  const payment = await db.payments.create({
    order_id: event.orderId,
    user_id: event.userId,
    amount: event.total,
    status: 'RESERVED'
  });

  // Publish PaymentReserved event
  await publishPaymentReserved(payment);
}

startConsumer().catch(console.error);

🐰 RabbitMQ: Traditional Message Broker

RabbitMQ — это traditional message broker, реализующий AMQP (Advanced Message Queuing Protocol). Отлично подходит для task queues и routing.

Producer → Exchange (type: topic) → Queues → Consumers

Exchange "orders" (topic):
    ↓ routing key: "order.created"
    ├─→ Queue "payment-service" → Consumer 1
    ├─→ Queue "inventory-service" → Consumer 2
    └─→ Queue "email-service" → Consumer 3

    ↓ routing key: "order.cancelled"
    ├─→ Queue "payment-service" → Consumer 1
    └─→ Queue "email-service" → Consumer 3

Flexible routing, push-based delivery

RabbitMQ: Преимущества

RabbitMQ: Недостатки

RabbitMQ Producer (Node.js)

// rabbitmq-producer.js
const amqp = require('amqplib');

let connection, channel;

async function connect() {
  connection = await amqp.connect('amqp://localhost:5672');
  channel = await connection.createChannel();

  // Declare exchange
  await channel.assertExchange('orders', 'topic', { durable: true });

  console.log('✅ Connected to RabbitMQ');
}

async function publishOrderCreated(order) {
  const event = {
    eventType: 'OrderCreated',
    orderId: order.id,
    userId: order.userId,
    items: order.items,
    total: order.total,
    timestamp: new Date().toISOString()
  };

  // Publish to exchange with routing key
  channel.publish(
    'orders',                    // exchange
    'order.created',             // routing key
    Buffer.from(JSON.stringify(event)),
    {
      persistent: true,          // Save to disk
      contentType: 'application/json',
      headers: {
        'correlation-id': order.correlationId
      }
    }
  );

  console.log(`✅ Event published: OrderCreated ${order.id}`);
}

// Graceful shutdown
process.on('SIGTERM', async () => {
  await channel.close();
  await connection.close();
});

connect().catch(console.error);

module.exports = { publishOrderCreated };

RabbitMQ Consumer (Node.js)

// rabbitmq-consumer.js
const amqp = require('amqplib');

async function startConsumer() {
  const connection = await amqp.connect('amqp://localhost:5672');
  const channel = await connection.createChannel();

  // Declare exchange
  await channel.assertExchange('orders', 'topic', { durable: true });

  // Declare queue
  const queue = 'payment-service-queue';
  await channel.assertQueue(queue, { durable: true });

  // Bind queue to exchange with routing key pattern
  await channel.bindQueue(queue, 'orders', 'order.*');

  // Set prefetch (how many messages to process in parallel)
  channel.prefetch(10);

  console.log('✅ Waiting for messages...');

  // Consume messages
  channel.consume(queue, async (msg) => {
    if (msg !== null) {
      const event = JSON.parse(msg.content.toString());

      console.log(`📨 Received event: ${event.eventType}`);

      try {
        if (event.eventType === 'OrderCreated') {
          await processPayment(event);
        }

        // Acknowledge message (remove from queue)
        channel.ack(msg);

      } catch (error) {
        console.error(`❌ Error processing event: ${error.message}`);

        // Reject and requeue (or send to DLQ)
        channel.nack(msg, false, false);  // false = don't requeue
      }
    }
  });
}

async function processPayment(event) {
  console.log(`💳 Processing payment for order ${event.orderId}`);

  // Reserve payment
  const payment = await db.payments.create({
    order_id: event.orderId,
    user_id: event.userId,
    amount: event.total,
    status: 'RESERVED'
  });
}

startConsumer().catch(console.error);

☁️ AWS SNS/SQS: Managed Services

AWS SNS (Simple Notification Service) — pub/sub для fan-out сообщений.
AWS SQS (Simple Queue Service) — managed message queue.

Producer (Lambda / API)
    ↓
SNS Topic "orders"
    ↓
    ├─→ SQS Queue "payment-service" → Lambda Consumer
    ├─→ SQS Queue "inventory-service" → Lambda Consumer
    └─→ SQS Queue "email-service" → Lambda Consumer

Fan-out pattern: 1 message → N queues

AWS SNS/SQS: Преимущества

AWS SNS/SQS: Недостатки

AWS SNS Producer (Node.js)

// aws-sns-producer.js
const { SNSClient, PublishCommand } = require('@aws-sdk/client-sns');

const snsClient = new SNSClient({ region: 'us-east-1' });

async function publishOrderCreated(order) {
  const event = {
    eventType: 'OrderCreated',
    orderId: order.id,
    userId: order.userId,
    items: order.items,
    total: order.total,
    timestamp: new Date().toISOString()
  };

  const command = new PublishCommand({
    TopicArn: 'arn:aws:sns:us-east-1:123456789012:orders',
    Message: JSON.stringify(event),
    MessageAttributes: {
      eventType: {
        DataType: 'String',
        StringValue: 'OrderCreated'
      }
    }
  });

  const response = await snsClient.send(command);
  console.log(`✅ Event published to SNS: ${response.MessageId}`);
}

module.exports = { publishOrderCreated };

AWS SQS Consumer (Node.js)

// aws-sqs-consumer.js
const { SQSClient, ReceiveMessageCommand, DeleteMessageCommand } = require('@aws-sdk/client-sqs');

const sqsClient = new SQSClient({ region: 'us-east-1' });
const queueUrl = 'https://sqs.us-east-1.amazonaws.com/123456789012/payment-service-queue';

async function pollMessages() {
  while (true) {
    try {
      // Long polling (wait up to 20 seconds)
      const command = new ReceiveMessageCommand({
        QueueUrl: queueUrl,
        MaxNumberOfMessages: 10,
        WaitTimeSeconds: 20,
        VisibilityTimeout: 30
      });

      const response = await sqsClient.send(command);

      if (response.Messages) {
        for (const message of response.Messages) {
          await processMessage(message);
        }
      }

    } catch (error) {
      console.error(`❌ Error polling SQS: ${error.message}`);
      await new Promise(resolve => setTimeout(resolve, 5000));
    }
  }
}

async function processMessage(message) {
  try {
    const snsMessage = JSON.parse(message.Body);
    const event = JSON.parse(snsMessage.Message);

    console.log(`📨 Received event: ${event.eventType}`);

    if (event.eventType === 'OrderCreated') {
      await processPayment(event);
    }

    // Delete message from queue
    await sqsClient.send(new DeleteMessageCommand({
      QueueUrl: queueUrl,
      ReceiptHandle: message.ReceiptHandle
    }));

    console.log(`✅ Message processed and deleted`);

  } catch (error) {
    console.error(`❌ Error processing message: ${error.message}`);
    // Message will become visible again after VisibilityTimeout
  }
}

async function processPayment(event) {
  console.log(`💳 Processing payment for order ${event.orderId}`);
  // ... payment logic
}

pollMessages().catch(console.error);

📊 Kafka vs RabbitMQ vs AWS SNS/SQS: Детальное сравнение

🎯 Когда что использовать?

📜 Event Sourcing Pattern

Event Sourcing — это паттерн где состояние системы хранится как последовательность событий, а не current state. Все изменения записываются как events в append-only log.

Traditional (current state only):
Account { id: 123, balance: 1000 }

Event Sourcing (event log):
1. AccountCreated    { accountId: 123, initialBalance: 0 }
2. MoneyDeposited    { accountId: 123, amount: 500 }
3. MoneyDeposited    { accountId: 123, amount: 1000 }
4. MoneyWithdrawn    { accountId: 123, amount: 500 }

Current Balance = sum of all events = 1000

Можно восстановить баланс на любой момент времени!

Event Sourcing Implementation (Node.js + Kafka)

// event-store.js (Event Sourcing with Kafka)
const { Kafka } = require('kafkajs');

const kafka = new Kafka({ brokers: ['localhost:9092'] });
const producer = kafka.producer();

// Append event to event store (Kafka topic)
async function appendEvent(aggregateId, eventType, eventData) {
  await producer.send({
    topic: 'account-events',  // Event store topic
    messages: [{
      key: aggregateId,
      value: JSON.stringify({
        aggregateId,
        eventType,
        eventData,
        timestamp: new Date().toISOString(),
        version: await getNextVersion(aggregateId)
      })
    }]
  });
}

// Replay events to rebuild state
async function replayEvents(aggregateId) {
  const consumer = kafka.consumer({ groupId: `replay-${Date.now()}` });
  await consumer.connect();
  await consumer.subscribe({ topic: 'account-events', fromBeginning: true });

  let state = { balance: 0 };

  await consumer.run({
    eachMessage: async ({ message }) => {
      const event = JSON.parse(message.value.toString());

      if (event.aggregateId === aggregateId) {
        state = applyEvent(state, event);
      }
    }
  });

  return state;
}

function applyEvent(state, event) {
  switch (event.eventType) {
    case 'AccountCreated':
      return { ...state, balance: event.eventData.initialBalance };
    case 'MoneyDeposited':
      return { ...state, balance: state.balance + event.eventData.amount };
    case 'MoneyWithdrawn':
      return { ...state, balance: state.balance - event.eventData.amount };
    default:
      return state;
  }
}

// Usage
await appendEvent('account-123', 'AccountCreated', { initialBalance: 0 });
await appendEvent('account-123', 'MoneyDeposited', { amount: 500 });
await appendEvent('account-123', 'MoneyWithdrawn', { amount: 200 });

const currentState = await replayEvents('account-123');
console.log(currentState);  // { balance: 300 }

🔀 CQRS (Command Query Responsibility Segregation)

CQRS — это паттерн разделения write model (commands) и read model (queries). Часто используется вместе с Event Sourcing.

Write Side (Commands):
Client → Command → Command Handler → Event Store (Kafka)
                                          ↓
                                    Publish Event
                                          ↓
Read Side (Queries):                      ↓
Event Consumer ← Event ← Event Store (Kafka)
      ↓
Update Read Model (PostgreSQL / Elasticsearch)
      ↓
Client ← Query ← Read Model

Разделение: Commands изменяют state, Queries читают

✅ Best Practices для Event-Driven Architecture

🔍 FAQ: Часто задаваемые вопросы

📝 Выводы

В этой статье мы рассмотрели Event-Driven Architecture и сравнили три решения:

1. Apache Kafka: High-throughput streaming (1M+ msg/sec), event sourcing, replay
2. RabbitMQ: Task queues, low latency, flexible routing
3. AWS SNS/SQS: Managed services, serverless, AWS ecosystem
4. Event Sourcing: Хранение state как последовательность events
5. CQRS: Разделение write и read models
6. Best Practices: Idempotency, schema versioning, DLQ, monitoring

Related Articles

• Saga Pattern: распределенные транзакции
• Service Mesh: Istio vs Linkerd vs Consul
• Circuit Breaker Pattern