Go Distributed Transaction Outbox: 5 Core Patterns for Reliable Event-Driven Architecture

后端开发

Le problème : Les points douloureux déclenchés par des événements

Un système de commande e-commerce refactorisé en architecture orientée événements souffrait fréquemment d'incohérences de données — "commande créée mais inventaire non déduit". L'enquête a révélé : l'envoi de messages et les opérations de base de données ne se trouvant pas dans la même transaction entraînant une perte de messages, les fluctuations de réseau déclenchant une consommation en double, une mauvaise sélection de clé de partition causant des problèmes d'ordre des événements, et des intervalles de balayage de la sortie trop longs provoquant des retards en aval. Ces quatre problèmes cumulatifs ont transformé la "cohérence finale" en "cohérence occasionnelle". Le modèle Transactional Outbox est la solution centrale, garantissant l'atomicité entre les opérations commerciales et la publication des événements.

Concepts clés en un coup d'œil

Concept Description Importance
Boîte de sortie transactionnelle Écrire des événements dans une table de sortie au sein de la même transaction métier, en garantissant l'atomique ⭐⭐⭐⭐⭐
Pilote par événements Découpler les services via des notifications d'événements plutôt que des appels synchrones ⭐⭐⭐⭐⭐
Fiabilité des messages Assurez-vous que les messages ne sont pas perdus, non dupliqués et livrés dans l'ordre ⭐⭐⭐⭐⭐
Consommation idempotente Le consommateur produit le même résultat lors du traitement d'un même message à plusieurs reprises ⭐⭐⭐⭐⭐
CDC Capture des changements de données, surveillance du Binlog de la base de données pour la publication d'événements en temps réel ⭐⭐⭐⭐⭐
Debezium Plateforme CDC open-source prenant en charge la capture des changements MySQL/PostgreSQL ⭐⭐⭐⭐
Message Retry Mécanisme de réessai pour la consommation échouée des messages avec stratégie d'atténuation ⭐⭐⭐⭐
Source d'Événements Utiliser les séquences d'événements comme source de vérité, permettant la reconstruction de l'état et l'audit ⭐⭐⭐

Analyse du problème : 5 grands défis dans la boîte aux lettres transactionnelle

**1. Atomicité des opérations commerciales et de l'envoi de messages : L'approche traditionnelle consiste à écrire dans la base de données (DB) avant d'envoyer un message — deux opérations qui ne peuvent pas garantir l'atomicité. Si l'écriture dans la DB réussit mais l'envoi du message échoue, les services en aval ne reçoivent jamais l'événement ; si l'envoi du message a lieu mais l'écriture dans la DB échoue, cela crée des événements fantômes.

**2. Garantie d'ordre des messages : Les événements pour la même racine d'agrégat doivent être consommés dans l'ordre, mais une mauvaise sélection de la clé de partition Kafka ou un envoi simultané par relais peut entraîner un réordonnancement, conduisant ainsi à l'exécution de logiques métier basées sur un état obsolète en aval.

**3. Implémentation de la consommation idempotente : La retransmission du réseau, l'envoi redondant en relais et les redémarrages des consommateurs provoquent tous une consommation double. Sans idempotence, une même commande pourrait déduire le stock deux fois.

**4. Latence de sondage de la sortie : Le sondage repose sur le balayage périodique de la table de sortie — des intervalles trop longs augmentent la latence, tandis que des intervalles trop courts gaspillent les ressources de la base de données. Sous haute concurrence, le sondage devient un goulot d'étranglement en termes de performance.

**5. Complexité de la configuration CDC : Debezium nécessite le déploiement de Kafka Connect, la configuration des Connecteurs et la gestion des changements de schéma — ce qui entraîne des coûts opérationnels élevés. Les environnements de production doivent également prendre en compte le format Binlog, les GTID et la haute disponibilité.

Pattern 1 : Conception de la table Outbox et écriture transactionnelle

La table Outbox est écrite dans la même transaction de base de données que la table métier, garantissant l'atomicité entre les opérations métier et les enregistrements d'événements. L'état de l'événement commence par PENDING et est envoyé de manière asynchrone par le relais.

package outbox

import (
    "context"
    "database/sql"
    "encoding/json"
    "time"
)

type OutboxEvent struct {
    ID          int64           `json:"id"`
    AggregateID string          `json:"aggregate_id"`
    EventType   string          `json:"event_type"`
    Payload     json.RawMessage `json:"payload"`
    Status      string          `json:"status"`
    Retries     int             `json:"retries"`
    CreatedAt   time.Time       `json:"created_at"`
}

type OutboxRepository struct {
    db *sql.DB
}

func NewOutboxRepository(db *sql.DB) *OutboxRepository {
    return &OutboxRepository{db: db}
}

func (r *OutboxRepository) SaveWithTx(ctx context.Context, tx *sql.Tx, event *OutboxEvent) error {
    query := `INSERT INTO outbox_events (aggregate_id, event_type, payload, status, created_at)
              VALUES (?, ?, ?, 'PENDING', NOW())`
    result, err := tx.ExecContext(ctx, query,
        event.AggregateID, event.EventType, event.Payload)
    if err != nil {
        return err
    }
    event.ID, _ = result.LastInsertId()
    return nil
}

type OrderService struct {
    db     *sql.DB
    outbox *OutboxRepository
}

func (s *OrderService) CreateOrder(ctx context.Context, orderID, userID string, items []string) error {
    tx, err := s.db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    defer tx.Rollback()

    _, err = tx.ExecContext(ctx,
        `INSERT INTO orders (id, user_id, items, status, created_at) VALUES (?, ?, ?, 'CREATED', NOW())`,
        orderID, userID, items)
    if err != nil {
        return err
    }

    payload, _ := json.Marshal(map[string]interface{}{
        "order_id": orderID,
        "user_id":  userID,
        "items":    items,
        "action":   "order_created",
    })
    event := &OutboxEvent{
        AggregateID: orderID,
        EventType:   "order.created",
        Payload:     payload,
    }
    if err := s.outbox.SaveWithTx(ctx, tx, event); err != nil {
        return err
    }

    return tx.Commit()
}

Table DDL de la boîte aux lettres :

CREATE TABLE outbox_events (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    aggregate_id VARCHAR(128) NOT NULL,
    event_type VARCHAR(128) NOT NULL,
    payload JSON NOT NULL,
    status ENUM('PENDING','SENT','FAILED') DEFAULT 'PENDING',
    retries INT DEFAULT 0,
    created_at DATETIME(3) NOT NULL,
    INDEX idx_status_created (status, created_at),
    INDEX idx_aggregate_id (aggregate_id)
) ENGINE=InnoDB;

Pattern 2 : Émetteur de relais par sondage

Le relais de sondage scanne périodiquement les événements de la boîte de sortie avec un statut PENDING, les publie sur Kafka, puis met à jour le statut à SENT. Point clé : utilisez SELECT ... FOR UPDATE SKIP LOCKED pour éviter les envois en double entre plusieurs instances.

package outbox

import (
    "context"
    "database/sql"
    "fmt"
    "log"
    "time"

    "github.com/segmentio/kafka-go"
)

type PollingRelay struct {
    db        *sql.DB
    writer    *kafka.Writer
    batchSize int
    interval  time.Duration
}

func NewPollingRelay(db *sql.DB, kafkaAddr, topic string, batchSize int, interval time.Duration) *PollingRelay {
    return &PollingRelay{
        db: db,
        writer: &kafka.Writer{
            Addr:         kafka.TCP(kafkaAddr),
            Topic:        topic,
            Balancer:     &kafka.LeastBytes{},
            BatchTimeout: 10 * time.Millisecond,
        },
        batchSize: batchSize,
        interval:  interval,
    }
}

func (r *PollingRelay) Start(ctx context.Context) {
    ticker := time.NewTicker(r.interval)
    defer ticker.Stop()
    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            if err := r.pollAndPublish(ctx); err != nil {
                log.Printf("polling relay error: %v", err)
            }
        }
    }
}

func (r *PollingRelay) pollAndPublish(ctx context.Context) error {
    tx, err := r.db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    defer tx.Rollback()

    rows, err := tx.QueryContext(ctx,
        `SELECT id, aggregate_id, event_type, payload, retries
         FROM outbox_events
         WHERE status = 'PENDING' AND retries < 5
         ORDER BY created_at ASC
         LIMIT ? FOR UPDATE SKIP LOCKED`, r.batchSize)
    if err != nil {
        return err
    }
    defer rows.Close()

    var events []OutboxEvent
    for rows.Next() {
        var e OutboxEvent
        if err := rows.Scan(&e.ID, &e.AggregateID, &e.EventType, &e.Payload, &e.Retries); err != nil {
            return err
        }
        events = append(events, e)
    }
    if len(events) == 0 {
        return nil
    }

    var messages []kafka.Message
    for _, e := range events {
        messages = append(messages, kafka.Message{
            Key:   []byte(e.AggregateID),
            Value: e.Payload,
            Headers: []kafka.Header{
                {Key: "event_type", Value: []byte(e.EventType)},
                {Key: "event_id", Value: []byte(fmt.Sprintf("%d", e.ID))},
            },
        })
    }
    if err := r.writer.WriteMessages(ctx, messages...); err != nil {
        for _, e := range events {
            tx.ExecContext(ctx, `UPDATE outbox_events SET retries = retries + 1 WHERE id = ?`, e.ID)
        }
        return fmt.Errorf("kafka write failed: %w", err)
    }

    for _, e := range events {
        if _, err := tx.ExecContext(ctx, `UPDATE outbox_events SET status = 'SENT' WHERE id = ?`, e.ID); err != nil {
            return err
        }
    }
    return tx.Commit()
}

Modèle 3 : Capture de données de changement CDC (Debezium)

Le CDC capture les modifications de la table Outbox en temps réel en surveillant le Binlog de la base de données — pas besoin de sondage, latence réduite. Debezium est une solution CDC de niveau production fonctionnant via Kafka Connect.

Configuration du connecteur MySQL Debezium :

{
  "name": "outbox-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "dbz_pass",
    "database.server.id": "184054",
    "database.server.name": "outbox_server",
    "database.include.list": "order_db",
    "table.include.list": "order_db.outbox_events",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "database.history.kafka.topic": "schema-changes",
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
    "transforms.outbox.route.topic.replacement": "order-events",
    "transforms.outbox.table.field.event.id": "id",
    "transforms.outbox.table.field.event.key": "aggregate_id",
    "transforms.outbox.table.field.event.type": "event_type",
    "transforms.outbox.table.field.event.payload": "payload",
    "transforms.outbox.table.fields.additional.placement": "status:header:eventStatus"
  }
}

Intégration du consommateur Go :

package consumer

import (
    "context"
    "log"

    "github.com/segmentio/kafka-go"
)

type OutboxEventHandler struct {
    reader *kafka.Reader
}

func NewOutboxEventHandler(kafkaAddr, topic, groupID string) *OutboxEventHandler {
    return &OutboxEventHandler{
        reader: kafka.NewReader(kafka.ReaderConfig{
            Brokers:  []string{kafkaAddr},
            Topic:    topic,
            GroupID:  groupID,
            MinBytes: 10e3,
            MaxBytes: 10e6,
        }),
    }
}

func (h *OutboxEventHandler) Start(ctx context.Context) {
    for {
        msg, err := h.reader.ReadMessage(ctx)
        if err != nil {
            if ctx.Err() != nil {
                return
            }
            log.Printf("read message error: %v", err)
            continue
        }
        eventType := ""
        for _, hdr := range msg.Headers {
            if hdr.Key == "event_type" {
                eventType = string(hdr.Value)
                break
            }
        }
        log.Printf("received event: type=%s key=%s", eventType, string(msg.Key))
    }
}

Modèle 4 : Consommation idempotente et suppression des doublons

La consommation idempotente est le filet de sécurité de l'architecture pilotée par événements. La suppression des doublons via une table d'enregistrement de consommation garantit qu'un même événement ne sera jamais traité deux fois.

package consumer

import (
    "context"
    "database/sql"
    "fmt"
)

type IdempotentHandler struct {
    db *sql.DB
}

func NewIdempotentHandler(db *sql.DB) *IdempotentHandler {
    return &IdempotentHandler{db: db}
}

func (h *IdempotentHandler) Handle(ctx context.Context, eventID string, handler func(ctx context.Context) error) error {
    tx, err := h.db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    defer tx.Rollback()

    var status string
    err = tx.QueryRowContext(ctx,
        `SELECT status FROM consume_records WHERE event_id = ? FOR UPDATE`, eventID).Scan(&status)
    if err == nil {
        if status == "PROCESSED" {
            return nil
        }
        return fmt.Errorf("event %s in status %s, skip", eventID, status)
    }
    if err != sql.ErrNoRows {
        return err
    }

    _, err = tx.ExecContext(ctx,
        `INSERT INTO consume_records (event_id, status, created_at) VALUES (?, 'PROCESSING', NOW())`, eventID)
    if err != nil {
        return err
    }

    if err := handler(ctx); err != nil {
        tx.ExecContext(ctx, `UPDATE consume_records SET status = 'FAILED' WHERE event_id = ?`, eventID)
        return err
    }

    _, err = tx.ExecContext(ctx, `UPDATE consume_records SET status = 'PROCESSED' WHERE event_id = ?`, eventID)
    if err != nil {
        return err
    }
    return tx.Commit()
}

Table de consommation :

CREATE TABLE consume_records (
    event_id VARCHAR(128) PRIMARY KEY,
    status ENUM('PROCESSING','PROCESSED','FAILED') DEFAULT 'PROCESSING',
    created_at DATETIME NOT NULL,
    updated_at DATETIME NOT NULL
) ENGINE=InnoDB;

Modèle 5 : Framework de sortie de production (avec surveillance)

Une boîte de sortie de niveau production nécessite : des vérifications de santé, la collecte de métriques, une fermeture gracieuse, des files d'attente de lettres mortes et des alertes. Ce framework intègre tous les modèles mentionnés ci-dessus.

package outbox

import (
    "context"
    "database/sql"
    "log"
    "sync"
    "time"

    "github.com/prometheus/client_golang/prometheus"
    "github.com/segmentio/kafka-go"
)

type OutboxFramework struct {
    db      *sql.DB
    writer  *kafka.Writer
    relay   *PollingRelay
    handler *IdempotentHandler
    cancel  context.CancelFunc
    wg      sync.WaitGroup

    eventsPublished prometheus.Counter
    eventsFailed    prometheus.Counter
    relayLatency    prometheus.Histogram
}

func NewOutboxFramework(db *sql.DB, kafkaAddr, topic string) *OutboxFramework {
    f := &OutboxFramework{
        db: db,
        writer: kafka.NewWriter(kafka.WriterConfig{
            Brokers:      []string{kafkaAddr},
            Topic:        topic,
            Balancer:     &kafka.LeastBytes{},
            BatchTimeout: 10 * time.Millisecond,
        }),
        relay:   NewPollingRelay(db, kafkaAddr, topic, 100, 500*time.Millisecond),
        handler: NewIdempotentHandler(db),
    }

    f.eventsPublished = prometheus.NewCounter(prometheus.CounterOpts{
        Name: "outbox_events_published_total",
        Help: "Total number of outbox events published",
    })
    f.eventsFailed = prometheus.NewCounter(prometheus.CounterOpts{
        Name: "outbox_events_failed_total",
        Help: "Total number of outbox events failed",
    })
    f.relayLatency = prometheus.NewHistogram(prometheus.HistogramOpts{
        Name:    "outbox_relay_latency_seconds",
        Help:    "Latency from event creation to publish",
        Buckets: prometheus.DefBuckets,
    })
    prometheus.MustRegister(f.eventsPublished, f.eventsFailed, f.relayLatency)
    return f
}

func (f *OutboxFramework) Start() {
    ctx, cancel := context.WithCancel(context.Background())
    f.cancel = cancel

    f.wg.Add(1)
    go func() {
        defer f.wg.Done()
        f.relay.Start(ctx)
    }()

    f.wg.Add(1)
    go func() {
        defer f.wg.Done()
        f.monitorPendingEvents(ctx)
    }()

    log.Println("outbox framework started")
}

func (f *OutboxFramework) monitorPendingEvents(ctx context.Context) {
    ticker := time.NewTicker(10 * time.Second)
    defer ticker.Stop()
    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            var pending int
            f.db.QueryRowContext(ctx,
                `SELECT COUNT(*) FROM outbox_events WHERE status = 'PENDING'`).Scan(&pending)
            if pending > 1000 {
                log.Printf("ALERT: %d pending outbox events, possible relay lag", pending)
            }
        }
    }
}

func (f *OutboxFramework) Shutdown() {
    f.cancel()
    f.wg.Wait()
    f.writer.Close()
    log.Println("outbox framework shutdown complete")
}

Guide des pièges à éviter

❌ Écrivez d'abord la base de données, puis envoyez le message, deux opérations sans garantie de transaction ✅ Utilisez la table Outbox pour écrire des événements dans la même transaction, en garantissant l'atomicité

❌ Relais de sondage sans verrouillage, plusieurs instances envoient des doublons ✅ Utilisez FOR UPDATE SKIP LOCKED pour une consommation d'exclusion mutuelle sans verrouillage

❌ Les clés aléatoires des messages Kafka provoquant un réordonnancement des événements ✅ Utilisez aggregate_id comme clé de partition pour garantir l'ordre pour la même racine d'agrégat

❌ Consommateur sans idempotence, la consommation en double provoque des erreurs métier ✅ Table d'enregistrement de consommation + Handler idempotent garantit que chaque événement est traité une seule fois

❌ La table de la boîte de sortie croît indéfiniment, la performance des requêtes se détériore ✅ Archiver régulièrement les événements SENT, les migrer vers la table d'historique après 7 jours

Dépannage des erreurs

Symptôme d'erreur Cause possible Solution
Événements en attente s'accumulant Le relais n'a pas été démarré ou Kafka est inaccessible Vérifiez l'état de la routine de relais et la connexion Kafka
Le consommateur reçoit des événements en double L'envoi a réussi mais la mise à jour de l'état a échoué Vérifiez la logique de validation de la transaction, assurez-vous que l'envoi et la mise à jour de l'état sont atomiques
Commande de consommation d'événement incorrecte Clé de partition ne utilisant pas aggregate_id Unifier l'utilisation de l'ID de racine d'agrégat comme clé de message Kafka
Le connecteur Debezium s'est arrêté Format de binlog non ROW ou droits insuffisants Vérifiez que binlog_format=ROW et accordez les privilèges de REPLICATION
Table d'impotentiel bloquée Consommation concurrente du même événement avec FOR UPDATE Utiliser un index unique + INSERT IGNORE au lieu de SELECT FOR UPDATE
Latence de sondage trop élevée Taille du lot trop petite ou intervalle trop long Augmenter batch_size à 200+ et raccourcir l'intervalle à 200 ms
Ralentissement de la requête de la table Outbox Grand volume de données sans index Ajouter un index composite (status, created_at), archiver périodiquement
Timeout lors de l'envoi d'un message Kafka Surcharge du cluster Kafka ou fluctuations du réseau Augmenter le WriteTimeout, activer les reprises et le producteur idempotent
Gestion de l'explosion des enregistrements dans la table Enregistrements expirés non nettoyés Suppression périodique des enregistrements PROCESSED plus anciens de 7 jours
Retard CDC de plusieurs minutes Mode de capture Debezium mal configuré Utilisez schema_only pour éviter la capture complète, vérifiez la rétention des Binlog

Optimisation avancée

1. Multi-Tenant Outbox : Ajoutez un champ tenant_id à la table Outbox, les relais sont envoyés par le shard du tenant pour empêcher les grands tenants de bloquer les petits.

2. Compression des événements : Utilisez la compression gzip pour le champ Payload — les corps d'événements volumineux (par exemple, les détails de commande) peuvent atteindre 70 % de compression, réduisant ainsi la bande passante Kafka et les coûts de stockage.

3. File d'attente prioritaire : Ajoutez un champ de priorité à la table de la Boîte de sortie. Les événements de haute priorité (paiement réussi) sont envoyés en premier, tandis que les événements de faible priorité (notifications) sont reportés.

**4. Dégradation d'écriture : Lorsque Kafka est indisponible, la table Outbox sert de tampon persistant. Le relais se dégrade automatiquement en mode de stockage local et rejoue les messages lorsque Kafka est rétabli.

**5. Registry des schémas d'événements : Utilisez Confluent Schema Registry pour gérer les versions des schémas d'événements. Les consommateurs désérialisent en fonction de la version, empêchant ainsi les modifications de schéma de provoquer des échecs de consommation.

Analyse comparative

Dimension Sortie de boîte Polling CDC (Debezium) MQ Transactionnel Événements Saga
Latence Moyenne (100ms-1s) Faible (<100ms) Faible (<50ms) Moyenne
Complexité de mise en œuvre Faible Élevée Moyenne Élevée
Coût opérationnel Faible Haut (Kafka Connect) Moyen Haut
Dépendance de la base de données Forte (pression de sonde) Faible (surveillance Binlog) Aucune Moyenne
Ordre des messages ✅ Contrôle de la clé de partition ✅ Journal binaire ordonné ✅ Messages de transaction ordonnés ⚠️ Exige une conception supplémentaire
Support de l'idempotence ⚠️ Implémenté manuellement ⚠️ Implémenté manuellement ✅ Présent dans le MQ ⚠️ Implémenté manuellement
Cas d'utilisation Petite à moyenne échelle, adoption rapide Grande échelle, exigences de faible latence Écosystème RocketMQ Orchestration de transactions longues

Résumé et Perspectives

La Transactional Outbox est la pierre angulaire de la fiabilité de l'architecture orientée événements, résolvant le problème d'atomicité entre les opérations métier et la publication d'événements. L'approche par sondage est simple à mettre en œuvre et convient pour une adoption rapide ; l'approche CDC offre une latence plus faible pour les scénarios à grande échelle. Les deux nécessitent une consommation idempotente pour garantir la cohérence finale. Les tendances futures incluent : la surveillance des changements de base de données basée sur eBPF remplaçant l'analyse Binlog, les bus d'événements Serverless simplifiant le relais Outbox, et le routage des messages et la détection d'anomalies pilotés par l'IA. Maîtriser ces 5 modèles fondamentaux permet de construire des architectures orientées événements fiables de niveau production.

Outils en ligne recommandés

Essayez ces outils exécutés localement dans le navigateur — aucune inscription requise →

#事务性发件箱#分布式事务#Outbox模式#消息可靠性#2026#后端开发