Разработка real-time приложений с использованием Change Streams MongoDB

Change Streams в экосистеме MongoDB

MongoDB Change Streams — API для подписки на поток изменений в коллекциях, базах или всём кластере (начиная с replica set или sharded cluster). Вместо polling по updatedAt или tailing oplog вручную приложение получает упорядоченные события insert, update, replace, delete с полным или delta документом после изменения.

Это основа для real-time дашбордов, инвалидации кэша, синхронизации search index (Elasticsearch, Atlas Search), event-driven микросервисов, audit log и push-уведомлений пользователям. Change Streams опираются на WiredTiger oplog и inherit его guarantees ordering в пределах одного shard key для sharded collections.

Как устроен поток событий

Клиент открывает cursor через collection.watch() или db.watch() с pipeline агрегации для фильтрации. Каждое событие содержит:

• operationType — insert, update, replace, delete, invalidate и др.
• fullDocument — при updateLookup или при insert/replace;
• documentKey — _id изменённого документа;
• clusterTime — логическая метка для resume;
• updateDescription — changedFields и removedFields для update.

Resume token позволяет переподключиться после обрыва без пропуска или дублирования (при корректной обработке idempotency на consumer). Без сохранения token consumer после restart начнёт с «now» и потеряет промежуток.

Pipeline-фильтрация на сервере

$match на operationType, поля fullDocument.status или tenantId снижает трафик на consumer. Фильтруйте как можно ближе к источнику — не тащите весь oplog в приложение ради одного поля.

Архитектура real-time приложения

Типичная схема: MongoDB → Change Stream consumer (Node, Python, Go worker) → message bus (Kafka, Redis Streams, NATS) или WebSocket gateway → клиенты. Альтернатива — serverless trigger через Atlas Trigger или Change Stream Handler в том же процессе API для простых кейсов.

WebSocket и SSE

API-сервер держит подписки клиентов по room/userId. Worker читает change stream, маршрутизирует событие в нужные room. Важно: fan-out на тысячи клиентов не держите в одном процессе без horizontal scale gateway.

Idempotency и ordering

Consumer может получить duplicate при reconnect. Храните processed event id (hash clusterTime + documentKey + operationType) в Redis или локальной таблице с TTL. Бизнес-обработчик должен быть идемпотентным: «обновить search index» безопасно повторить, «списать баланс» — нет без dedup.

Backpressure

При burst записей в MongoDB consumer не успевает — lag растёт, memory pressure. Решения: batch обработка, отдельная очередь с consumer group, масштабирование workers по partition key (shard _id range).

Требования, ограничения и версии

• Replica set обязателен; standalone не поддерживает change streams.
• Sharded cluster: stream на sharded collection соблюдает порядок только для документов с одним shard key value; глобальный total order не гарантирован.
• Transaction events видны как отдельные операции или через full transaction в зависимости от версии.
• Pre/post images (change stream pre/post images) дают до/после для update — полезно для audit, требует включения на коллекции.
• DDL (drop collection) генерирует invalidate — consumer должен пересоздать stream.

На Atlas доступны дополнительные интеграции; self-hosted требует мониторинга oplog window: если consumer отстаёт дольше, чем хранится oplog, resume невозможен без full sync.

Безопасность и эксплуатация

Change stream cursor использует права чтения на базу. Выделите role readChangeStream + read на нужные коллекции для service account worker, не admin. TLS между worker и MongoDB обязателен в prod.

Метрики для алертов:

• lag между clusterTime события и временем обработки;
• rate ошибок resume (ExpiredChangeStreamEvent);
• размер очереди downstream;
• restarts consumer pod без сохранённого token.

Тестируйте failover: primary step-down не должен убивать stream навсегда — драйвер переподключается к новому primary с resume token.

Практический сценарий: live-дашборд заказов

Коллекция orders с полями status, warehouseId, updatedAt. Pipeline: $match status in [processing, shipped]. Worker пишет в Redis pub/sub channel warehouse:{id}. Frontend подписан на WebSocket и обновляет таблицу без refresh. При update status на delivered событие уходит только подписчикам нужного склада.

Дополнительно: тот же stream кормит Elasticsearch для полнотекстового поиска; idempotency по orderId + updatedAt version field предотвращает stale overwrite.

Change Streams vs альтернативы

Polling — просто, но latency и load на DB. Triggers в приложении — легко пропустить путь записи. Debezium CDC — мощнее для multi-DB, но тяжелее в ops. Change Streams — sweet spot внутри MongoDB-only landscape с минимальным glue.

Если real-time слой критичен для продукта, закладывайте его в архитектуру с первого релиза: resume tokens, idempotency, мониторинг oplog lag — не «добавим потом».

Разбор интеграции Change Streams в ваш стек — на консультации по real-time MongoDB.

Надёжный consumer Change Streams

Change Streams удобны для fan-out событий, но в проде важны resume token, обработка ошибок и backpressure. Храните token в надёжном store; после рестарта consumer должен продолжить с того же места, а не «с сейчас».

Не тащите тяжёлую бизнес-логику прямо в колбэк стрима: кладите события в очередь и обрабатывайте воркерами. Иначе один медленный handler остановит чтение.

• Фильтры pipeline на нужные operationType и поля
• Идемпотентность по идентификатору события
• Алерты на лаг consumer и потери соединения
• Тест failover replica set: стрим должен пережить primary stepDown

Границы применимости

Change Streams — не замена полноценной event-driven архитектуре. Для очень высоких write TPS и сложного фан-аута часто выгоднее явный outbox. Используйте стримы там, где MongoDB уже source of truth и нужен относительно простой real-time слой.

Записаться на бесплатную консультацию.