Налаштування черг повідомлень: Apache Kafka для мікросервісів

Черги повідомлень — критичний компонент у мікросервісній архітектурі. Apache Kafka (Wikipedia) справляється з навантаженнями від 100 тисяч повідомлень на секунду, але неправильне налаштування призводить до втрати даних або недетермінованого порядку. Наприклад, в одному проєкті клієнт втратив замовлення через установку acks=0: продюсер не чекав підтвердження, і при збої брокера повідомлення зникали безслідно. Після міграції на acks=all з enable.idempotence=true проблема зникла — доставка стала гарантованою, а порядок подій за одним ключем зберігся. Крім того, ми впровадили Schema Registry для контролю версій повідомлень: тепер різні сервіси можуть безпечно еволюціонувати свої дані. Налаштуємо Kafka під ваш проєкт за 3–5 днів з повною документацією.

Чому Kafka, а не RabbitMQ для потокової обробки?

Kafka зберігає повідомлення за retention-політикою (наприклад, 7 днів або 100 ГБ), дозволяючи множині consumer'ів читати один топік з різними зміщеннями. RabbitMQ видаляє повідомлення після підтвердження — це добре для task-черг, але погано для аудиту та replay. У потокових сценаріях Kafka приблизно в 10 разів продуктивніша за RabbitMQ за пропускною здатністю. При цьому Kafka в 3 рази економніша на зберіганні великого обсягу даних за рахунок послідовного запису та стиснення.

Пропускна здатність Kafka досягає 2–3 мільйонів повідомлень на секунду на кластері з трьох нод, тоді як RabbitMQ — не більше 300–500 тисяч. Retention за замовчуванням — 7 днів, але ми налаштовуємо під бізнес-логіку, наприклад, для аудиту — 30 днів, для логів — 100 ГБ.

Як ми налаштовуємо Kafka: стек, конфіги, процес

Використовуємо Confluent-образи Kafka 7.6+ з обов'язковим Schema Registry для контракту даних. Для PHP застосовуємо librdkafka, для Node.js — kafkajs. Процес:

1. Аналіз навантаження та проєктування топіків (кількість партицій, фактор реплікації).
2. Розгортання через Docker Compose або Kubernetes (Strimzi).
3. Налаштування producer з idempotent та acks=all.
4. Реалізація consumer'ів з ручним commit after processing.
5. Моніторинг через JMX + Grafana з алертами при lag > 10 000.

Установка через Docker

---
# docker-compose.yml services:
zookeeper:
  image: confluentinc/cp-zookeeper:7.6.0
  environment:
    ZOOKEEPER_CLIENT_PORT: 2181
    ZOOKEEPER_TICK_TIME: 2000
  volumes:
    - zookeeper_data:/var/lib/zookeeper/data
    - zookeeper_log:/var/lib/zookeeper/log
kafka:
  image: confluentinc/cp-kafka:7.6.0
  depends_on: [zookeeper]
  environment:
    KAFKA_BROKER_ID: 1
    KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
    KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
    KAFKA_LOG_RETENTION_HOURS: 168 # 7 днів
    KAFKA_LOG_RETENTION_BYTES: 107374182400 # 100 GB
    KAFKA_NUM_PARTITIONS: 6
    KAFKA_DEFAULT_REPLICATION_FACTOR: 1
  volumes:
    - kafka_data:/var/lib/kafka/data
  ports:
    - "9092:9092"
kafka-ui:
  image: provectuslabs/kafka-ui:latest
  depends_on: [kafka]
  environment:
    KAFKA_CLUSTERS_0_NAME: local
    KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
  ports:
    - "8080:8080"
volumes:
  zookeeper_data:
  zookeeper_log:
  kafka_data:

Створення топіків

# Створити топік з 6 партиціями та реплікою 1 (для одиночного брокера)
kafka-topics.sh --bootstrap-server kafka:9092 \
  --create \
  --topic user-events \
  --partitions 6 \
  --replication-factor 1 \
  --config retention.ms=604800000 \
  --config cleanup.policy=delete

# Перегляд
kafka-topics.sh --bootstrap-server kafka:9092 --describe --topic user-events

PHP: Producer (librdkafka)

use RdKafka\Producer;
use RdKafka\Conf;

class KafkaProducer
{
    private Producer $producer;

    public function __construct()
    {
        $conf = new Conf();
        $conf->set('bootstrap.servers', config('kafka.brokers'));
        $conf->set('security.protocol', 'PLAINTEXT');
        $conf->set('acks', 'all'); // підтвердження від усіх реплік
        $conf->set('retries', '3');
        $conf->set('enable.idempotence', 'true'); // рівно один запис
        $conf->set('compression.type', 'snappy');
        $conf->setDrMsgCb(function ($kafka, $message) {
            if ($message->err !== RD_KAFKA_RESP_ERR_NO_ERROR) {
                Log::error('Kafka delivery failed', [
                    'error' => $message->errstr(),
                    'topic' => $message->topic_name,
                ]);
            }
        });
        $this->producer = new Producer($conf);
    }

    public function publish(string $topic, string $key, array $payload): void
    {
        $rdTopic = $this->producer->newTopic($topic);
        $rdTopic->produce(
            partition: RD_KAFKA_PARTITION_UA, // автовибір партиції по key
            msgflags: 0,
            payload: json_encode($payload),
            key: $key, // один ключ → одна партиція → порядок подій
        );
        $this->producer->poll(0);
    }

    public function flush(): void
    {
        $result = $this->producer->flush(10000); // 10 секунд таймаут
        if (RD_KAFKA_RESP_ERR_NO_ERROR !== $result) {
            throw new \RuntimeException('Kafka flush failed: ' . rd_kafka_err2str($result));
        }
    }
}

// Використання
$producer->publish('user-events', (string) $user->id, [
    'event' => 'user.registered',
    'user_id' => $user->id,
    'email' => $user->email,
    'timestamp' => now()->toIso8601String(),
]);
$producer->flush();

Node.js: Producer та Consumer на kafkajs

import { Kafka, CompressionTypes } from 'kafkajs';

const kafka = new Kafka({
  clientId: 'myapp-api',
  brokers: [process.env.KAFKA_BROKERS!],
  retry: {
    retries: 5,
    initialRetryTime: 300,
    factor: 0.2,
  },
});

// Producer
const producer = kafka.producer({
  allowAutoTopicCreation: false,
  idempotent: true,
  maxInFlightRequests: 5,
});

await producer.connect();
await producer.send({
  topic: 'user-events',
  compression: CompressionTypes.Snappy,
  messages: [
    {
      key: String(userId),
      value: JSON.stringify({
        event: 'user.login',
        userId,
        ip,
        timestamp: Date.now(),
      }),
      headers: {
        'content-type': 'application/json',
      },
    },
  ],
});

// Consumer
const consumer = kafka.consumer({
  groupId: 'audit-service',
});

await consumer.connect();
await consumer.subscribe({
  topic: 'user-events',
  fromBeginning: false,
});

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

Як досягти ідемпотентності producer?

Ідемпотентність означає, що запис повідомлення в топік відбувається рівно один раз, навіть при повторному відправленні. Для цього в Kafka потрібно включити enable.idempotence=true та acks=all. При цьому продюсер отримує унікальний ідентифікатор (producer id та sequence number), а брокер відкидає дублікати. Це обов'язкове налаштування для фінансових операцій та аудиту.

Як налаштувати моніторинг лагу consumer?

Для моніторингу використовуйте команду kafka-consumer-groups.sh --describe --bootstrap-server localhost:9092 --group <group_name>. Вона показує зміщення consumer'а (current-offset) та останнє повідомлення в топіку (log-end-offset). Різниця — це lag. Для автоматизації ми встановлюємо JMX Exporter, який публікує метрики в Prometheus, а потім будуємо дашборди в Grafana з порогом алерту при lag > 10 000. Це дозволяє вчасно помітити відставання та перерозподілити партиції.

Що входить у налаштування під ключ

• Аудит поточної архітектури — аналіз навантаження, вибір кількості топіків та партицій.
• Розгортання кластера — Docker Compose або Kubernetes (Strimzi) з урахуванням відмовостійкості.
• Інтеграція з додатком — написання producer/consumer на PHP або Node.js з обробкою помилок.
• Schema Registry — впровадження Avro-схем для контролю версій повідомлень.
• Моніторинг — дашборд Grafana з алертами по лагу та завантаженню брокерів.
• Навчання команди — документація з експлуатації та аварійного відновлення.

Строки реалізації

Вартість розраховується індивідуально — залежить від складності інтеграції та необхідності Streams/KSQL. Ми працюємо з Kafka більше 5 років і реалізували 20+ проєктів з налаштування черг. Отримайте консультацію з налаштування Kafka — оцінимо завдання за 1 день.