В мире событийно-управляемой архитектуры часто возникает иллюзия, что для создания микросервиса обязательно нужно подключать Apache Flink, Kafka Streams или Spark. Однако правда заключается в том, что базовый шаблон производителя и потребителя (Basic Producer/Consumer — BPC) покрывает потребности огромного количества бизнес-сценариев.

BPC-микросервис — это, по сути, элегантно простой цикл: получить событие → применить бизнес-логику → опубликовать результат. В нём нет встроенных механизмов для водяных знаков, сложного планирования событий или распределённых журналов изменений. Но именно эта простота делает его невероятно гибким инструментом в руках грамотного архитектора.

Это десятая глава цикла. Она опирается на главу V (state stores, rebalance), главу VII (внешние vs внутренние state stores) и главу IX (FaaS как форма простоты). Здесь мы разберём, где BPC выигрывает у «тяжёлых» фреймворков, а где — проигрывает.

IАнатомия BPC: минимализм как преимущество

В основе BPC лежит использование стандартных клиентских библиотек брокера событий. Весь бизнес-процесс ограниченного контекста инкапсулирован в одном приложении. Это снижает порог входа, позволяет использовать любой язык программирования и избавляет команду от необходимости поддерживать кластеры сложных потоковых фреймворков.

// Базовый цикл BPC-микросервиса
while (running) {
    ConsumerRecords<String, Event> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, Event> record : records) {
        // Применяем кастомную бизнес-логику
        OutputEvent output = processEvent(record.value());
        
        // Публикуем результат
        producer.send(new ProducerRecord<>("output-topic", output));
    }
    
    // Коммитим offset только ПОСЛЕ успешной обработки
    consumer.commitSync();
}

Обратите внимание: commitSync() вызывается после обработки всего батча. Это ключевое отличие BPC от наивной реализации «poll → commit → process», которая при сбое теряет данные.

IIПаттерн «Коляска» (Sidecar) для Legacy-систем

Представьте старый монолит, который не может напрямую работать с потоками событий, но которому критически важны данные из них (например, фронтенд интернет-магазина, которому нужны актуальные цены и остатки).

Паттерн Sidecar позволяет запустить BPC-микросервис в одном контейнере с монолитом. «Коляска» потребляет события из брокера и тихо обновляет локальную БД монолита (или кэш), не требуя изменения ни строчки кода в самом наследии.

// Sidecar: читает поток цен, пишет в БД монолита
while (running) {
    ConsumerRecords<String, PriceEvent> records = consumer.poll(...);
    
    try (Connection conn = legacyDb.getConnection()) {
        for (ConsumerRecord<String, PriceEvent> record : records) {
            PriceEvent event = record.value();
            
            // Upsert в таблицу монолита
            PreparedStatement stmt = conn.prepareStatement(
                "INSERT INTO prices (product_id, price) VALUES (?, ?) " +
                "ON CONFLICT (product_id) DO UPDATE SET price = EXCLUDED.price"
            );
            stmt.setString(1, event.getProductId());
            stmt.setBigDecimal(2, event.getPrice());
            stmt.executeUpdate();
        }
        consumer.commitSync();
    } catch (SQLException e) {
        // Offset НЕ коммитим — повторим batch
        log.error("Failed to update legacy DB", e);
    }
}

IIIGating: состояние без строгого порядка

Многие бизнес-процессы требуют наличия нескольких фактов, но порядок их поступления не имеет значения. Классический пример — книгоиздание.

Чтобы отправить книгу в печать, нужны три события: «Текст готов», «Обложка утверждена», «Цена установлена». BPC-сервис материализует эти факты в локальную таблицу. Как только все три ячейки заполнены (независимо от того, в каком порядке они пришли), сервис генерирует событие «Готово к печати».

-- Таблица состояния Gating
CREATE TABLE book_readiness (
    book_id UUID PRIMARY KEY,
    text_ready BOOLEAN DEFAULT FALSE,
    cover_approved BOOLEAN DEFAULT FALSE,
    price_set BOOLEAN DEFAULT FALSE,
    ready_for_print BOOLEAN DEFAULT FALSE,
    updated_at TIMESTAMP DEFAULT NOW()
);

-- Обработка события
UPDATE book_readiness
SET text_ready = TRUE
WHERE book_id = :bookId;

-- Проверка готовности
UPDATE book_readiness
SET ready_for_print = TRUE
WHERE book_id = :bookId
  AND text_ready = TRUE
  AND cover_approved = TRUE
  AND price_set = TRUE
  AND ready_for_print = FALSE
RETURNING book_id;
-- Если вернулась строка — публикуем событие ReadyForPrint

Обратите внимание на AND ready_for_print = FALSE: это защита от повторной публикации события при повторной обработке того же факта (идемпотентность).

IVКогда «тяжёлую» работу делает уровень данных

Зачем писать сложную логику агрегации в коде микросервиса, если это может сделать специализированная БД? Если ваш сценарий включает геопространственный поиск (например, поиск ближайших кафе на основе событий геолокации) или классификацию через ML-модель, BPC выступает лишь в роли «маршрутизатора».

// BPC как маршрутизатор к внешнему ML-сервису
while (running) {
    ConsumerRecords<String, ImageEvent> records = consumer.poll(...);
    
    for (ConsumerRecord<String, ImageEvent> record : records) {
        ImageEvent event = record.value();
        
        // Отправляем в ML-сервис
        ClassificationResult result = mlService.classify(event.getImageUrl());
        
        // Сохраняем результат
        producer.send(new ProducerRecord<>(
            "image-classifications",
            new ClassificationEvent(event.getImageId(), result.getLabel())
        ));
    }
    consumer.commitSync();
}

Вся сложность инкапсулирована в инфраструктуре данных: ML-сервис, vector DB, геопространственный индекс. BPC здесь — тонкий слой оркестрации.

VНезависимое масштабирование обработки и данных

Объём событий часто зависит от времени суток (циклы сна и бодрствования пользователей), а вот размер состояния (например, количество уникальных пользователей) остаётся постоянным.

BPC позволяет элегантно решить эту проблему: ночью вы масштабируете количество инстансов обработки до нуля (экономя деньги), но внешнее хранилище состояний (например, DynamoDB или Redis) продолжает хранить профили всех пользователей. Утром BPC-инстансы просто просыпаются и продолжают читать состояния.

VIГибридные приложения: аутсорсинг стрим-процессинга

Что делать, если вам нужно сделать сложный JOIN трёх потоков в реальном времени, но ваш основной язык программирования не имеет хорошей библиотеки для этого?

Вы используете гибридный паттерн. BPC-микросервис запускает внешнее задание в потоковом фреймворке (например, в KSQL или Flink), который выполняет «грязную» работу по объединению данных. Результат сбрасывается в промежуточный поток событий, который ваш BPC-микросервис забирает и использует для обслуживания REST API или загрузки в витрину данных.

┌─────────────┐     ┌──────────────┐     ┌─────────────┐
│  Raw Events │────>│  Flink Job   │────>│  Joined     │
│  (3 streams)│     │  (JOIN)      │     │  Stream     │
└─────────────┘     └──────────────┘     └──────┬──────┘
                                                 │
                                                 ▼
                                          ┌─────────────┐
                                          │ BPC Service │
                                          │ (serve API) │
                                          └─────────────┘

Flink решает узкую задачу, которая плохо ложится на BPC. BPC решает всё остальное: обслуживание API, запись в витрину, интеграцию с другими сервисами.

VIIИдемпотентность в BPC

BPC-микросервис работает в режиме at-least-once: если обработка batch упала до коммита offset, тот же batch будет обработан снова. Значит, шаги должны быть идемпотентны.

Три подхода

1. Idempotency key. Использовать event_id или бизнес-ключ как idempotency key. Сервис хранит таблицу обработанных ID и пропускает дубликаты.
2. Conditional writes. В PostgreSQL — INSERT ... ON CONFLICT DO UPDATE. В DynamoDB — attribute_not_exists(event_id).
3. Естественная идемпотентность. Если бизнес-операция идемпотентна по природе (SET price = 100, а не price += 100) — дубликаты не страшны.

// Дедупликация через таблицу обработанных событий
Connection conn = db.getConnection();
conn.setAutoCommit(false);
try {
    // Проверка дубликата и бизнес-операция в одной транзакции
    PreparedStatement check = conn.prepareStatement(
        "INSERT INTO processed_events (event_id) VALUES (?) " +
        "ON CONFLICT DO NOTHING RETURNING event_id"
    );
    check.setString(1, event.getId());
    ResultSet rs = check.executeQuery();
    
    if (!rs.next()) {
        // Уже обработано — пропускаем
        conn.commit();
        continue;
    }
    
    // Бизнес-операция
    updateBusinessState(conn, event);
    conn.commit();
} catch (SQLException e) {
    conn.rollback();
    // Offset НЕ коммитим
}

VIIIGraceful shutdown

Что происходит, когда BPC-контейнер получает SIGTERM при деплое? Если просто выключиться — broker будет ждать session.timeout.ms, прежде чем передаст partitions другому потребителю. Это создаёт искусственные задержки в обработке.

public class BpcService {
    private volatile boolean running = true;
    
    public void run() {
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            running = false;
        }));
        
        try {
            while (running) {
                ConsumerRecords<String, Event> records = consumer.poll(...);
                processRecords(records);
                consumer.commitSync();
            }
        } finally {
            // Ключевое: закрыть consumer → отказаться от partitions
            consumer.close(Duration.ofSeconds(10));
            producer.flush();
            producer.close(Duration.ofSeconds(10));
            db.close();
        }
    }
}

Правило: при любых сомнениях всегда очищайте соединения перед завершением. Это ускоряет rebalance и снижает нагрузку на broker.

IXRebalance в BPC

BPC-микросервис на контейнере — это тоже consumer group. При деплое, при падении инстанса — rebalance. Все проблемы из главы V применимы и здесь.

// Настройка consumer для минимизации rebalance
props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
    "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");
props.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG,
    "bpc-instance-" + System.getenv("HOSTNAME"));
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "10000");

XObservability BPC-микросервиса

BPC в production без мониторинга — чёрный ящик. Что обязательно нужно измерять:

XIBPC vs Kafka Streams vs FaaS: decision tree

Decision tree

Нужны watermarks / event time / windows?
    ├── Да → Kafka Streams или Flink
    └── Нет ↓

Нужны сложные joins нескольких потоков?
    ├── Да → Kafka Streams или Flink (или гибридный BPC + Flink)
    └── Нет ↓

Нужна детерминированная обработка с exactly-once?
    ├── Да → Kafka Streams (exactly_once_v2)
    └── Нет ↓

Нагрузка переменная, нужна экономия при простое?
    ├── Да → FaaS (глава IX)
    └── Нет ↓

BPC на контейнере + внешнее state

XIIОграничения BPC: когда простота не спасает

Базовый шаблон не был бы универсальным, если бы не имел своих границ. Отказываясь от «тяжеловесных» и «легковесных» фреймворков, вы берёте на себя ответственность за реализацию сложных паттернов самостоятельно.

Кроме того, в BPC-паттерне внешние хранилища состояний используются чаще внутренних. Это означает, что сетевые задержки (latency) при чтении/записи в БД становятся вашим главным врагом производительности, и архитектуру нужно проектировать с учётом кэширования и пакетной обработки.

Когда BPC — плохой выбор

• Streaming joins трёх и более потоков — Kafka Streams или Flink справятся лучше.
• Скользящие окна с event time — нужны watermarks, которых в BPC нет.
• Exactly-once с транзакциями — Kafka Streams предоставляет из коробки.
• Очень высокая нагрузка (> 100k msg/s) — фреймворк с in-memory state эффективнее.
• Interactive Queries — нет встроенной поддержки чтения state извне.

XIIIЧеклисты и связь с циклом

Связь с предыдущими главами

Чеклист BPC-микросервиса

• Offset коммитится ПОСЛЕ успешной обработки batch
• Обработка batch в транзакции с дедупликацией (idempotency key + processed_events)
• Graceful shutdown: SIGTERM → drain → close consumer → commit offsets
• Cooperative sticky assignor в конфигурации consumer
• Static group membership (group.instance.id) для предсказуемых деплоев
• State вынесен во внешнее хранилище (Redis, PostgreSQL, DynamoDB)
• Кэширование часто читаемых данных, чтобы не бить в БД на каждом событии
• Batch processing для снижения числа round-trip к внешнему state
• DLQ для событий, которые не удаётся обработать после N ретраев
• Метрики: consumer lag, processing time p99, error rate, dlq_rate, commit latency, rebalance count
• Три обязательных алерта: consumer lag, error rate, DLQ non-empty
• Decision tree применён: не нужен ли Kafka Streams / Flink / FaaS
• Documentation: какие сценарии покрыты, какие — нет

Ключевые выводы главы

1. BPC покрывает 80% бизнес-сценариев. Простой цикл poll → process → produce + внешнее state — этого достаточно для большинства задач.
2. Sidecar — интеграция legacy-систем без модификации их кода.
3. Gating — состояние без строгого порядка: достаточно факта наличия, а не времени.
4. Тяжёлая работа — на уровень данных. BPC — тонкий слой оркестрации, ML и vector DB делают остальное.
5. Масштабирование до нуля — главное преимущество внешнего state. Контейнеры stateless, state в Redis/DynamoDB.
6. Гибридный BPC + Flink — когда нужен сложный join, но не хочется переписывать весь пайплайн.
7. Идемпотентность обязательна. At-least-once — реальность, шаги должны быть идемпотентны.
8. Graceful shutdown ускоряет rebalance. Consumer.close() перед завершением.
9. Rebalance-стратегии те же, что в Kafka Streams: cooperative sticky, static membership.
10. Observability: три обязательных алерта (lag, error rate, DLQ).
11. Decision tree: BPC vs Kafka Streams vs FaaS — по потребности в watermarks, joins, EOS, переменной нагрузке.

Глава X цикла «Событийно-управляемая архитектура». Источники: Apache Kafka, Confluent, практические кейсы production-систем, опыт FinCodeCow.