Микросервис архитектурасында әр қызметтің өз дерекқоры бар. Kafka хабарламаны әлі алмаған жағдайда, Outbox паттерні осы контексте қалай жұмыс істейді?

Outbox үлгісі микросервистің деректерін жазу мен Kafka-ға оқиғаларды жариялаудың атомарлығын қамтамасыз ету үшін пайдаланылады, әрбір қызметтің өз дерекқоры болған кезде.

Идеясы — бір транзакцияда қызметтің күйі өзгерген кезде негізгі бизнес деректерімен бірге арнайы outbox кестесіне оқиға (хабарлама) жазылады. Осылайша, оқиғаны жазу мен бизнес-операция атомарлы — екеуі де сәтті немесе екеуі де қайтарылады.

Бөлек процесс (немесе ағын) outbox кестесінен жаңа жазбаларды оқиды және оларды Kafka-ға жібереді. Егер жіберу сәтсіз болса, жазба кестеде қалады және кейінірек қайталанады. Жіберу сәтті болған кезде, жазба жіберілді деп белгіленеді немесе жойылады.

Осылайша, тіпті Kafka уақытша қол жетімсіз болса да, деректер жоғалмайды, ал қызмет тұтастығы бұзылмайды, себебі оқиға жергілікті сақталады және кейінірек жіберіледі.

Қарапайым схеманың мысалы:

// Транзакцияда
func SaveOrderAndOutbox(db *sql.DB, order Order, event Event) error {
    tx, _ := db.Begin()
    defer tx.Rollback()

    // Тапсырысты сақтау
    _, err := tx.Exec("INSERT INTO orders (...) VALUES (...)", ...)
    if err != nil {
        return err
    }

    // Оқиғаны outbox-қа жазу
    _, err = tx.Exec("INSERT INTO outbox (event_type, payload, processed) VALUES (?, ?, false)", event.Type, event.Payload)
    if err != nil {
        return err
    }

    return tx.Commit()
}

// Бөлек жұмысшы outbox-тан оқиды және Kafka-ға жібереді
func OutboxWorker(db *sql.DB, kafkaProducer KafkaProducer) {
    for {
        rows, _ := db.Query("SELECT id, event_type, payload FROM outbox WHERE processed = false")
        for rows.Next() {
            var id int
            var eventType, payload string
            rows.Scan(&id, &eventType, &payload)

            err := kafkaProducer.Send(eventType, payload)
            if err == nil {
                db.Exec("UPDATE outbox SET processed = true WHERE id = ?", id)
            }
        }
        time.Sleep(time.Second * 5)
    }
}

Осылайша, сенімді хабарламалар жеткізілуі және қызметтер арасындағы деректердің келісімділігі қамтамасыз етіледі.