Что такое паттерн Outbox?

Паттерн Outbox

Outbox паттерн — это архитектурный паттерн, используемый в распределенных системах для обеспечения надежной доставки сообщений при асинхронной коммуникации. Паттерн решает проблему потери данных, когда операция в базе данных успешна, но сообщение не отправляется в message broker.

Проблема, которую решает Outbox

Сценарий без Outbox (уязвимость):

1. Приложение 1 обновляет базу данных (заказ создан)
2. Приложение 1 отправляет событие в message broker (RabbitMQ, Kafka)
3. ❌ Сбой перед отправкой или во время отправки
4. Результат: БД обновлена, но событие потеряно
5. Приложение 2 не узнает о новом заказе

Обеспечение атомарности:

Outbox гарантирует, что если транзакция в БД успешна, то сообщение гарантированно будет доставлено.

Как работает Outbox

Архитектура паттерна

┌─────────────────────────────────────────┐
│ Приложение (Пример: Order Service)      │
└────────────────┬────────────────────────┘
                 │
         ┌───────┴───────┐
         │               │
    ┌────▼─────┐   ┌─────▼──────┐
    │  orders  │   │   outbox   │
    │  table   │   │   table    │
    └──────────┘   └──────┬─────┘
         ▲                │
         │                │
  ┌──────┴────────────────▼────────┐
  │ ACID Transaction (одна транзакция)
  │  - INSERT into orders
  │  - INSERT into outbox
  └────────────────────────────────┘
         │
         │
    ┌────▼──────────────┐
    │ Outbox Poller     │ (фоновый процесс)
    │ (читает outbox)   │
    └────┬─────────────┘
         │
  ┌──────▼──────────────┐
  │  Message Broker    │
  │ (RabbitMQ, Kafka)  │
  └────┬───────────────┘
       │
  ┌────▼─────────────────┐
  │ Other Services       │
  │ (получают события)   │
  └──────────────────────┘

Реализация Outbox в Java

1. Создание Outbox таблицы

CREATE TABLE outbox (
    id UUID PRIMARY KEY,
    aggregate_id UUID NOT NULL,
    aggregate_type VARCHAR(255) NOT NULL,
    event_type VARCHAR(255) NOT NULL,
    payload JSON NOT NULL,
    created_at TIMESTAMP NOT NULL,
    published_at TIMESTAMP,
    published BOOLEAN DEFAULT FALSE
);

CREATE INDEX idx_outbox_published ON outbox(published, created_at);

2. Domain Event класс

public abstract class DomainEvent {
    private UUID aggregateId;
    private LocalDateTime createdAt;
    private String eventType;
    
    public DomainEvent(UUID aggregateId) {
        this.aggregateId = aggregateId;
        this.createdAt = LocalDateTime.now();
        this.eventType = this.getClass().getSimpleName();
    }
    
    public UUID getAggregateId() {
        return aggregateId;
    }
    
    public LocalDateTime getCreatedAt() {
        return createdAt;
    }
    
    public String getEventType() {
        return eventType;
    }
}

public class OrderCreatedEvent extends DomainEvent {
    private UUID orderId;
    private String customerName;
    private BigDecimal totalAmount;
    
    public OrderCreatedEvent(UUID orderId, String customerName, BigDecimal totalAmount) {
        super(orderId);
        this.orderId = orderId;
        this.customerName = customerName;
        this.totalAmount = totalAmount;
    }
    
    public UUID getOrderId() {
        return orderId;
    }
    
    public String getCustomerName() {
        return customerName;
    }
    
    public BigDecimal getTotalAmount() {
        return totalAmount;
    }
}

3. Outbox репозиторий

@Repository
public class OutboxRepository {
    private final JdbcTemplate jdbcTemplate;
    private final ObjectMapper objectMapper;
    
    @Autowired
    public OutboxRepository(JdbcTemplate jdbcTemplate, ObjectMapper objectMapper) {
        this.jdbcTemplate = jdbcTemplate;
        this.objectMapper = objectMapper;
    }
    
    public void save(DomainEvent event) {
        String eventPayload = objectMapper.writeValueAsString(event);
        
        String sql = "INSERT INTO outbox (id, aggregate_id, aggregate_type, " +
                "event_type, payload, created_at, published) " +
                "VALUES (?, ?, ?, ?, ?, ?, false)";
        
        jdbcTemplate.update(sql,
            UUID.randomUUID(),
            event.getAggregateId(),
            "Order",
            event.getEventType(),
            eventPayload,
            event.getCreatedAt()
        );
    }
    
    public List<OutboxEvent> findUnpublished() {
        String sql = "SELECT id, aggregate_id, event_type, payload, created_at " +
                "FROM outbox WHERE published = false ORDER BY created_at ASC";
        
        return jdbcTemplate.query(sql, (rs, rowNum) -> new OutboxEvent(
            UUID.fromString(rs.getString("id")),
            UUID.fromString(rs.getString("aggregate_id")),
            rs.getString("event_type"),
            rs.getString("payload"),
            rs.getTimestamp("created_at").toLocalDateTime()
        ));
    }
    
    public void markAsPublished(UUID eventId) {
        String sql = "UPDATE outbox SET published = true, published_at = NOW() " +
                "WHERE id = ?";
        jdbcTemplate.update(sql, eventId);
    }
}

public class OutboxEvent {
    private UUID id;
    private UUID aggregateId;
    private String eventType;
    private String payload;
    private LocalDateTime createdAt;
    
    public OutboxEvent(UUID id, UUID aggregateId, String eventType, String payload, LocalDateTime createdAt) {
        this.id = id;
        this.aggregateId = aggregateId;
        this.eventType = eventType;
        this.payload = payload;
        this.createdAt = createdAt;
    }
    
    // getters
}

4. Order Service с Outbox

@Service
public class OrderService {
    private final OrderRepository orderRepository;
    private final OutboxRepository outboxRepository;
    
    @Autowired
    public OrderService(OrderRepository orderRepository, OutboxRepository outboxRepository) {
        this.orderRepository = orderRepository;
        this.outboxRepository = outboxRepository;
    }
    
    @Transactional
    public Order createOrder(CreateOrderRequest request) {
        UUID orderId = UUID.randomUUID();
        
        // Создаем заказ в базе данных
        Order order = new Order(
            orderId,
            request.getCustomerName(),
            request.getTotalAmount()
        );
        orderRepository.save(order);
        
        // Сохраняем событие в Outbox (в той же транзакции)
        DomainEvent event = new OrderCreatedEvent(
            orderId,
            request.getCustomerName(),
            request.getTotalAmount()
        );
        outboxRepository.save(event);
        
        // Если оба INSERT успешны, транзакция коммитится
        // Если что-то сломается, оба откатываются
        return order;
    }
}

5. Outbox Poller (фоновый процесс)

@Component
public class OutboxPoller {
    private final OutboxRepository outboxRepository;
    private final MessagePublisher messagePublisher;
    private static final Logger logger = LoggerFactory.getLogger(OutboxPoller.class);
    
    @Autowired
    public OutboxPoller(OutboxRepository outboxRepository, MessagePublisher messagePublisher) {
        this.outboxRepository = outboxRepository;
        this.messagePublisher = messagePublisher;
    }
    
    @Scheduled(fixedRate = 1000) // Каждую секунду
    public void pollAndPublish() {
        List<OutboxEvent> unpublishedEvents = outboxRepository.findUnpublished();
        
        for (OutboxEvent event : unpublishedEvents) {
            try {
                // Отправляем в message broker
                messagePublisher.publish(event);
                
                // Отмечаем как опубликованное
                outboxRepository.markAsPublished(event.getId());
                
                logger.info("Published event: {} for aggregate: {}",
                    event.getEventType(), event.getAggregateId());
            } catch (Exception e) {
                logger.error("Failed to publish event: {}", event.getId(), e);
                // Повторная попытка в следующем цикле
            }
        }
    }
}

6. Message Publisher

@Service
public class MessagePublisher {
    private final RabbitTemplate rabbitTemplate;
    private static final String EXCHANGE = "orders.events";
    
    @Autowired
    public MessagePublisher(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }
    
    public void publish(OutboxEvent event) {
        String routingKey = event.getEventType().toLowerCase();
        rabbitTemplate.convertAndSend(
            EXCHANGE,
            routingKey,
            event.getPayload()
        );
    }
}

Гарантии Outbox паттерна

✅ At-Least-Once Delivery:
   - Событие в Outbox гарантирует, что оно будет отправлено
   - Может быть отправлено несколько раз (idempotency требуется)

✅ Atomicity:
   - Создание сущности и запись события атомарны
   - Если одно не удалось, откатывается всё

✅ Eventual Consistency:
   - Другие сервисы в конечном итоге получат событие
   - Может быть задержка, но консистентность гарантирована

Обработка сообщений идемпотентно

@Service
public class OrderEventListener {
    private final ProcessedEventRepository processedEventRepository;
    private final NotificationService notificationService;
    
    @RabbitListener(queues = "order-events.queue")
    public void handleOrderCreated(OrderCreatedEvent event) {
        // Проверяем, не обработано ли уже это событие
        if (processedEventRepository.exists(event.getAggregateId())) {
            return; // Уже обработано, игнорируем
        }
        
        // Обрабатываем событие
        notificationService.sendConfirmationEmail(
            event.getCustomerName(),
            event.getTotalAmount()
        );
        
        // Отмечаем как обработанное
        processedEventRepository.save(event.getAggregateId());
    }
}

Альтернативы Outbox

1. Change Data Capture (CDC)

Добавить триггер в БД, который следит за изменениями
Использовать инструменты типа Debezium

2. Event Sourcing

Вместо хранения состояния, хранишь последовательность событий
Любое изменение = новое событие
Дороже в реализации, но полная история

Преимущества и недостатки

Преимущества:

• Гарантирует доставку сообщений
• Простая реализация
• Не требует изменения БД архитектуры

Недостатки:

• Дополнительная таблица (storage overhead)
• Нужна фоновая задача (poller)
• Возможны дубликаты сообщений (нужна идемпотентность)
• Может быть задержка в доставке

Выводы

• Outbox паттерн решает проблему потери событий
• Атомарность: событие и данные сохраняются вместе
• Poller: фоновый процесс, который отправляет события
• Идемпотентность: обработчик должен уметь обрабатывать дубликаты
• At-Least-Once: гарантирует минимум одну доставку
• Используется в микросервисных архитектурах для надежной коммуникации