Transactional Outbox таблица PostgreSQL
Transactional Outbox — это шаблон для надёжной передачи событий из базы данных в брокер сообщений (Kafka, RabbitMQ, NATS и т. д.). Он помогает избежать проблем с «двойной записью» (Dual Write Problem), обеспечивая атомарность базы данных и брокера сообщений.
Структура таблицы outbox
Стандартная таблица Transactional Outbox содержит следующие ключевые поля:
|
1 2 3 4 5 6 7 8 9 10 11 |
CREATE TABLE outbox ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), -- Уникальный идентификатор события aggregate_id UUID NOT NULL, -- Идентификатор сущности (например, заказа) aggregate_type TEXT NOT NULL, -- Тип сущности (например, "Order") event_type TEXT NOT NULL, -- Тип события (например, "ORDER_CREATED") payload JSONB NOT NULL, -- Данные события в формате JSON created_at TIMESTAMP DEFAULT NOW(), -- Время создания события processed_at TIMESTAMP NULL, -- Время обработки события (NULL = ещё не обработано) status TEXT DEFAULT 'PENDING' CHECK (status IN ('PENDING', 'PROCESSED', 'FAILED')) -- Статус обработки ); |
Объяснение полей:
| 📌 Поле | 📜 Описание |
|---|---|
id |
Уникальный ID события (UUID). |
aggregate_id |
ID связанной сущности (например, заказа, пользователя). |
aggregate_type |
Тип агрегата (например, «Order», «User»). |
event_type |
Тип события (например, «ORDER_CREATED», «USER_REGISTERED»). |
payload |
Данные события в формате JSON (гибкость хранения данных). |
created_at |
Время создания записи. |
processed_at |
Время обработки (NULL = ещё не обработано). |
status |
Состояние обработки (PENDING, PROCESSED, FAILED). |
Как работает Transactional Outbox?
1️⃣ Сервис записывает событие в outbox в рамках основной транзакции.
2️⃣ Фоновый «Outbox Processor» читает новые события (PENDING).
3️⃣ Отправляет их в брокер (Kafka, RabbitMQ, NATS и т. д.).
4️⃣ После успешной отправки обновляет статус на PROCESSED.
Обновление статуса после обработки
После успешной отправки сообщения в брокер нужно обновить статус:
|
1 2 3 4 |
UPDATE outbox SET status = 'PROCESSED', processed_at = NOW() WHERE id = 'some-event-id'; |
FAILED:|
1 2 3 4 |
UPDATE outbox SET status = 'FAILED' WHERE id = 'some-event-id'; |
Очистка старых записей
Чтобы база не разрасталась, можно удалять старые записи:
|
1 2 |
DELETE FROM outbox WHERE processed_at < NOW() - INTERVAL '30 days'; |
ИЛИ использовать партиционирование по дате (created_at), чтобы эффективно очищать старые данные.
Вывод
🔹 Transactional Outbox — это надёжный способ доставки событий.
🔹 Запись в outbox идёт в той же транзакции, что и основная БД, исключая расхождения.
🔹 Фоновый процессор гарантированно отправит события в брокер сообщений.