Налаштування черг повідомлень: 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 день.

Послуги бекенд-розробки: production-grade надійність

На production-сервері о 3:14 ночі черга Laravel Jobs перестала оброблятися — 40 000 необроблених завдань у Redis. Причина: worker упав через memory leak у статичній змінній Eloquent observer, supervisor не перезапустив через misconfigured stopwaitsecs. Ми розбирали такий інцидент на проекті з 500 RPS: діагностика 4 години, фікс — 20 хвилин. Щоб ви не втрачали гроші, пропонуємо послуги бекенд-розробки з акцентом на production-grade надійність — 10+ років досвіду, 50+ проектів, 5 років на ринку. Оцінимо ваш проект за 2 дні.

Які проблеми вирішуємо

N+1 запити: головний вбивця швидкості

N+1 — найпоширеніша причина повільних сторінок у Laravel-додатках. Стандартна історія: сторінка працювала нормально на dev з 10 записами, на production з 10 000 — 8-секундне завантаження.

Laravel Debugbar у dev-оточенні показує кількість запитів. Більше 20 — сигнал для audit.

Model::preventLazyLoading(! app()->isProduction());

Telescope для профілювання: логує всі запити, jobs, mail, notifications з деталізацією. Після впровадження eager loading час завантаження сторінки падає з 8 с до 0.3 с — у 27 разів.

Memory leak у статичних змінних

У Laravel Octane або Swoole додаток тримається в пам’яті між запитами. Статичні змінні не скидаються — призводять до неконтрольованого росту пам’яті. Використовуємо defer-функції та контейнерні біндинги для коректного скидання стану.

Неправильний connection pool

Rails, Laravel, Django відкривають нове з'єднання PostgreSQL на кожен PHP/Python процес. 100 воркерів — 100 з'єднань. PostgreSQL деградує від 200+ активних з'єднань через overhead на управління.

PgBouncer у transaction pooling: 1000 воркерів → 20–50 реальних з'єднань. Це знижує latency на 40% та зменшує витрати на хостинг на 30% — при середній вартості хостингу $2,000/міс економить $600/міс. GIN-індекс для JSONB до 100 разів швидший за B-tree при пошуку.

Як Octane справляється з високим навантаженням?

Laravel Octane (RoadRunner або Swoole) прибирає overhead bootstrap на кожен HTTP-запит. Приріст: 3–8x на синтетичних бенчмарках, 2–4x на реальних додатках. Важливо: не зберігати стан у статичних змінних — застосовуємо це на проектах >1000 RPS.

Як PostgreSQL допомагає уникнути повільних запитів?

Використовуємо composite indexes для WHERE + ORDER BY, partial indexes для фільтрів з високою селективністю, GIN-індекси для JSONB та full-text search. to_tsvector + GIN замість LIKE '%query%' — запобігає seq scan навіть на мільйонах записів. Аналізуємо плани через EXPLAIN ANALYZE та pg_stat_statements.

Як обрати стек для вашого проекту?

Node.js виправданий для realtime: Laravel публікує події в Redis Pub/Sub, Node.js підписується та транслює клієнтам. Go — для goroutines (10k з'єднань на сервер — норма), але розробка повільніша, ніж Laravel.

Чому Redis критичний для продуктивності?

Redis виконує кілька ролей:

Redis Cluster для горизонтального масштабування, Sentinel для автоматичного failover. Замовте консультацію щодо оптимізації Redis для вашого проекту.

Що входить в роботу під ключ

• Архітектурне проектування (документація API, схема БД, діаграма сервісів)
• Реалізація за узгодженим ТЗ з code review
• Налаштування CI/CD (GitHub Actions, Docker), моніторингу (Sentry, Grafana), алертингу
• Навантажувальне тестування (k6, wrk) зі звітом
• Передача вихідних кодів, доступів, інструкція з деплою
• Навчання команди замовника (2–3 сесії)
• Гарантійна підтримка 1 місяць після здачі

Орієнтири по термінах

Вартість розраховується індивідуально після аналізу вимог до навантаження, інтеграцій та бізнес-логіки. Зв'яжіться з нами для безкоштовного аудиту вашого поточного backend — отримайте план оптимізації за 2 дні. Замовте консультацію та дізнайтеся, як знизити витрати на інфраструктуру на 30% без втрати продуктивності.