Настройка очередей сообщений: 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 день.