Kiến trúc Microservices Ngân hàng: Go, Saga & Event Sourcing

🇬🇧 Read the English version of this article on tanhdev.com

1. Giới thiệu: Phá vỡ Core Banking truyền thống

Trả lời nhanh: Một kiến trúc microservices ngân hàng hiện đại sẽ thay thế các sổ cái nguyên khối cũ (như T24 hoặc Flexcube) bằng cách sử dụng Go để định tuyến giao dịch với thông lượng cao. Hệ thống đạt được tính nhất quán phân tán mà không cần dùng two-phase commit (2PC) bằng cách kết hợp Event Sourcing (luồng sự kiện bất biến của sổ cái), Điều phối Saga (dùng Temporal hoặc Dapr), mẫu Transactional Outbox và ràng buộc unique của PostgreSQL để đảm bảo API luôn idempotent.

Trong nhiều thập kỷ, các ngân hàng phụ thuộc vào các hệ thống core nguyên khối như Temenos T24 hay Oracle FLEXCUBE. Dù ổn định, nhưng các hệ thống này lại tạo ra nút thắt cổ chai nghiêm trọng cho ngân hàng số hiện đại. Chúng được thiết kế để xử lý lô (batch processing) qua đêm, chứ không phải cho các giao dịch toàn cầu, thời gian thực theo hướng API-first.

Việc chuyển đổi sang kiến trúc microservices vào năm 2026 đòi hỏi phải phá vỡ các giới hạn này:

• Giới hạn mở rộng: Hệ thống nguyên khối mở rộng theo chiều dọc (phần cứng đắt đỏ), trong khi microservices mở rộng theo chiều ngang.
• Chu kỳ phát hành: Core cũ đòi hỏi những đợt release khổng lồ, rủi ro cao hàng quý. Microservices cho phép triển khai độc lập liên tục.
• Khóa dữ liệu: Cơ sở dữ liệu tập trung trong hệ thống nguyên khối tạo ra xung đột khóa nghiêm trọng trong các sự kiện có tần suất cao (như xử lý ngày trả lương).

Bằng cách tận dụng runtime xử lý đồng thời cực tốt của Go và kiến trúc hướng sự kiện phân tán, chúng ta tối ưu hóa hệ thống để đạt tốc độ ghi cơ sở dữ liệu <10ms ở mức 10.000 TPS, đảm bảo khả năng mở rộng và chịu lỗi.

2. Phân tách Domain: Lập bản đồ các Context trong Core Banking

Trả lời nhanh: Việc phân tách một hệ thống core banking đòi hỏi phải xác định các bounded context hoạt động độc lập. Các domain chính là Accounts (CASA), Payments, Ledgers và Notifications, cho phép mở rộng độc lập và sở hữu dữ liệu riêng biệt.

Để chuyển đổi thành công bằng mẫu Strangler Fig, bạn phải thiết lập một Anti-Corruption Layer (ACL) để biên dịch các model cũ thành các bounded context hiện đại.

Dưới đây là cách các domain cốt lõi tương tác:

graph TD
    API[API Gateway] --> Accounts[Accounts Service - CASA]
    API --> Payments[Payments Routing Service]
    Payments --> Ledger[Ledger Service]
    Accounts --> Ledger
    Ledger --> Notifications[Notification Service]
    
    subgraph Legacy Core
        ACL[Anti-Corruption Layer]
        T24[Temenos T24]
        ACL --> T24
    end
    
    Ledger -.Sync.-> ACL

Mỗi service sở hữu cơ sở dữ liệu riêng. Ledger Service không bao giờ truy vấn trực tiếp cơ sở dữ liệu Accounts; thay vào đó, nó đăng ký nhận các sự kiện thay đổi trạng thái bất biến.

3. Event Sourcing: Thiết kế Sổ cái kép Bất biến (Immutable Double-Entry Ledger)

Trả lời nhanh: Sổ cái kép bất biến đảm bảo tuân thủ kiểm toán bằng cách ghi lại các sự kiện tài chính thay vì sửa đổi các trường số dư. Bằng cách sử dụng PostgreSQL với Optimistic Concurrency Control (OCC) và unique index trên (stream_id, version), hệ thống đảm bảo tính nhất quán tuần tự tuyệt đối.

Ràng buộc cốt lõi của bất kỳ hệ thống tài chính nào là không bao giờ lưu trữ số dư làm bản ghi chính. Lưu trữ một cột balance có thể bị sửa đổi sẽ dẫn đến lỗi lost updates và hỏng hóc dữ liệu không thể cứu vãn. Thay vào đó, bạn phải lưu trữ các giao dịch (Event Sourcing).

DDL cho Mô hình Ghi trên PostgreSQL

Schema này ép buộc sử dụng Optimistic Concurrency Control (OCC) cho luồng sự kiện:

CREATE TABLE ledger_streams (
    stream_id UUID PRIMARY KEY,
    version BIGINT NOT NULL,
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE ledger_events (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    stream_id UUID NOT NULL REFERENCES ledger_streams(stream_id),
    version BIGINT NOT NULL,
    event_type VARCHAR(100) NOT NULL,
    payload JSONB NOT NULL,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    CONSTRAINT uq_stream_version UNIQUE (stream_id, version)
);

Transaction Ghi OCC bằng Go

Khi thêm một sự kiện mới, hệ thống sẽ kiểm tra expected_version để ngăn ngừa race conditions.

// Go repository query dùng pgx/v5
tx, err := pool.Begin(ctx)
if err != nil {
    return err
}
defer tx.Rollback(ctx)

// 1. Xác minh và cập nhật phiên bản
res, err := tx.Exec(ctx, `
    UPDATE ledger_streams 
    SET version = $1, updated_at = NOW() 
    WHERE stream_id = $2 AND version = $3`, 
    expectedVersion+1, streamID, expectedVersion)
if err != nil {
    return err
}
if res.RowsAffected() == 0 {
    return ErrConcurrencyConflict // Phiên bản đã thay đổi kể từ lúc đọc
}

// 2. Chèn sự kiện
_, err = tx.Exec(ctx, `
    INSERT INTO ledger_events (stream_id, version, event_type, payload) 
    VALUES ($1, $2, $3, $4)`, 
    streamID, expectedVersion+1, eventType, payloadJson)
if err != nil {
    return err
}

return tx.Commit(ctx)

Để tối ưu hóa runtime Go cho tốc độ ghi <10ms ở mức 10.000 TPS, chúng tôi sử dụng connection pool PgBouncer ở chế độ transaction, ổ lưu trữ NVMe và tắt synchronous_commit = off (nếu quy tắc nghiệp vụ cho phép độ trễ nhỏ khi sập hệ thống).

Schema Sổ cái kép cấp thương mại (Production-Grade Double-Entry Schema)

Trong khi cơ chế Event Sourcing ghi nhận các chuyển đổi trạng thái, một hệ thống sổ cái kép thực sự yêu cầu các ràng buộc xác minh số dư nghiêm ngặt trên các tài khoản tài sản (asset), nợ phải trả (liability), vốn chủ sở hữu (equity), doanh thu (revenue) và chi phí (expense). Trong nguyên tắc kế toán kép, mỗi giao dịch tài chính phải có ít nhất một khoản ghi nợ (debit) và một khoản ghi có (credit), đồng thời tổng số tiền ghi nợ phải bằng chính xác tổng số tiền ghi có.

Dưới đây là mã DDL PostgreSQL cấp thương mại cho hệ thống sổ cái kép chuẩn hóa và hiệu năng cao. Schema này sử dụng các ràng buộc kiểm tra (check constraints) để bắt buộc các giá trị không âm và hàm trigger để đảm bảo tính bất biến của số dư ở cấp độ giao dịch.

-- Bắt buộc các phân loại tài khoản
CREATE TYPE account_class AS ENUM ('ASSET', 'LIABILITY', 'EQUITY', 'REVENUE', 'EXPENSE');
CREATE TYPE entry_direction AS ENUM ('DEBIT', 'CREDIT');

-- Bảng Accounts cốt lõi đại diện cho hệ thống tài khoản
CREATE TABLE accounts (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    code VARCHAR(50) UNIQUE NOT NULL,
    name VARCHAR(255) NOT NULL,
    class account_class NOT NULL,
    currency CHAR(3) NOT NULL DEFAULT 'USD',
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);

-- Journal Entries đóng vai trò là vỏ bọc giao dịch (phần tiêu đề)
CREATE TABLE journal_entries (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    reference_id UUID UNIQUE, -- Tham chiếu giao dịch bên ngoài (liên kết idempotency)
    narration TEXT NOT NULL,
    posted_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);

-- Ledger Lines đại diện cho các bút toán riêng lẻ (dòng) trong một journal entry
CREATE TABLE ledger_lines (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    journal_entry_id UUID NOT NULL REFERENCES journal_entries(id) ON DELETE CASCADE,
    account_id UUID NOT NULL REFERENCES accounts(id),
    amount NUMERIC(20, 4) NOT NULL CHECK (amount > 0),
    direction entry_direction NOT NULL,
    -- Cột chuỗi băm đơn giản để kiểm toán phát hiện giả mạo
    line_hash BYTEA,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);

-- Tạo chỉ mục hỗn hợp để đối soát dòng sổ cái và tính toán số dư tài khoản nhanh chóng
CREATE INDEX idx_ledger_lines_account_posted ON ledger_lines(account_id, created_at DESC);
CREATE INDEX idx_ledger_lines_journal_entry ON ledger_lines(journal_entry_id);

Để áp dụng tính bất biến của sổ cái kép (tổng nợ bằng tổng có trong một giao dịch), chúng ta sử dụng cơ chế kiểm tra ràng buộc trì hoãn (deferred constraint check) hoặc trigger cấp độ câu lệnh. Dưới đây là hàm trigger PostgreSQL thực hiện xác minh tính toàn vẹn của số dư trước khi commit transaction:

CREATE OR REPLACE FUNCTION verify_journal_entry_balance()
RETURNS TRIGGER AS $$
DECLARE
    v_debit_sum NUMERIC(20, 4);
    v_credit_sum NUMERIC(20, 4);
BEGIN
    -- Tính tổng nợ (debit) và có (credit) cho journal entry hiện tại
    SELECT 
        COALESCE(SUM(amount) FILTER (WHERE direction = 'DEBIT'), 0),
        COALESCE(SUM(amount) FILTER (WHERE direction = 'CREDIT'), 0)
    INTO v_debit_sum, v_credit_sum
    FROM ledger_lines
    WHERE journal_entry_id = NEW.journal_entry_id;

    IF v_debit_sum <> v_credit_sum THEN
        RAISE EXCEPTION 'Double-entry balance mismatch for journal entry %. Total debits (%) must equal total credits (%).', 
            NEW.journal_entry_id, v_debit_sum, v_credit_sum;
    END IF;

    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

-- Áp dụng trigger dưới dạng constraint trigger được thực thi ở cuối câu lệnh hoặc giao dịch
CREATE CONSTRAINT TRIGGER trg_enforce_double_entry
AFTER INSERT OR UPDATE ON ledger_lines
DEFERRABLE INITIALLY DEFERRED
FOR EACH ROW
EXECUTE FUNCTION verify_journal_entry_balance();

Vòng lặp xác minh giao dịch (Transaction Verification Loops)

Trong các hệ thống phân tán hiệu năng cao, cơ sở dữ liệu thỉnh thoảng có thể gặp lỗi hỏng dữ liệu ngầm (silent data corruption), hoặc các lỗi ở cấp độ ứng dụng có thể vượt qua các kiểm tra ràng buộc (ví dụ: trong quá trình di chuyển dữ liệu hoặc can thiệp thủ công). Để giảm thiểu rủi ro này, các kiến trúc core banking hiện đại triển khai các Vòng lặp xác minh giao dịch bất đồng bộ (Transaction Verification Loops).

Vòng lặp xác minh giao dịch là một dịch vụ chạy nền bằng Go, liên tục kiểm toán sổ cái theo từng phân đoạn (ví dụ: các lô hàng giờ) để xác minh tính bất biến toán học và chữ ký mật mã của các bản ghi sổ cái.

Logic của Vòng lặp xác minh:

1. Kiểm toán tính bất biến toán học (Mathematical Invariant Audit): Vòng lặp thực hiện các truy vấn phân tích hiệu năng cao để đảm bảo rằng tất cả các dòng ledger_lines được nhóm theo journal_entry_id đều có tổng bằng 0 (tương ứng với nợ - có = 0).
2. Kiểm toán chuỗi băm (Hash-Chain Auditing): Để ngăn chặn hành vi sửa đổi dữ liệu trái phép (ví dụ: một tác nhân nội bộ thay đổi số tiền giao dịch trong SQL), mỗi dòng sổ cái lưu trữ một mã băm mật mã của nội dung dòng đó kết hợp với mã băm của dòng trước đó (mã line_hash của dòng N là SHA256(amount + direction + account_id + journal_entry_id + line_hash_of_N-1)). Vòng lặp xác minh sẽ duyệt qua chuỗi băm này để kiểm tra tính toàn vẹn.
3. Vòng lặp đối soát (Reconciliation Loop): Thực hiện đối soát số dư của các dòng sổ cái với các bản ghi bên ngoài (như báo cáo cổng thanh toán hoặc luồng sự kiện Kafka).

Dưới đây là một đoạn mã Go triển khai truy vấn kiểm toán toán học được chạy bởi vòng lặp xác minh:

type ReconciliationFailure struct {
    JournalEntryID uuid.UUID
    DebitSum       decimal.Decimal
    CreditSum      decimal.Decimal
}

func RunVerificationLoop(ctx context.Context, db *pgxpool.Pool, interval time.Duration) {
    ticker := time.NewTicker(interval)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            failures, err := auditLedgerInvariants(ctx, db)
            if err != nil {
                log.Printf("Verification loop query failure: %v", err)
                continue
            }
            if len(failures) > 0 {
                for _, f := range failures {
                    log.Printf("CRITICAL: Ledger imbalance detected at journal entry %s: debits=%s, credits=%s", 
                        f.JournalEntryID, f.DebitSum.String(), f.CreditSum.String())
                    // Kích hoạt cảnh báo, đóng băng các tài khoản bị ảnh hưởng, và liên hệ kỹ sư trực ca (on-call)
                }
            }
        }
    }
}

func auditLedgerInvariants(ctx context.Context, db *pgxpool.Pool) ([]ReconciliationFailure, error) {
    // Kiểm toán các giao dịch trong 2 giờ qua để phát hiện các commit bị trễ
    query := `
        SELECT 
            journal_entry_id,
            COALESCE(SUM(amount) FILTER (WHERE direction = 'DEBIT'), 0) AS debit_sum,
            COALESCE(SUM(amount) FILTER (WHERE direction = 'CREDIT'), 0) AS credit_sum
        FROM ledger_lines
        WHERE created_at >= NOW() - INTERVAL '2 hours'
        GROUP BY journal_entry_id
        HAVING COALESCE(SUM(amount) FILTER (WHERE direction = 'DEBIT'), 0) <> 
               COALESCE(SUM(amount) FILTER (WHERE direction = 'CREDIT'), 0)
    `
    rows, err := db.Query(ctx, query)
    if err != nil {
        return nil, err
    }
    defer rows.Close()

    var failures []ReconciliationFailure
    for rows.Next() {
        var f ReconciliationFailure
        if err := rows.Scan(&f.JournalEntryID, &f.DebitSum, &f.CreditSum); err != nil {
            return nil, err
        }
        failures = append(failures, f)
    }
    return failures, nil
}

Bằng cách tách biệt vòng lặp xác minh này khỏi luồng ghi chính (hot write path), hệ thống duy trì độ trễ commit giao dịch cực thấp trong khi vẫn đảm bảo bằng chứng toán học và mật mã về tính chính xác của sổ cái chỉ trong vòng vài giây.

4. Mẫu Transactional Outbox: Ngăn chặn lỗi Dual-Write

Trả lời nhanh: Mẫu Transactional Outbox giải quyết bài toán dual-write bằng cách chèn dữ liệu nghiệp vụ và thông điệp sự kiện vào cơ sở dữ liệu trong cùng một transaction cục bộ. Các Go worker (hoặc công cụ CDC) sau đó sẽ quét bảng outbox để đảm bảo tin nhắn được đẩy lên Kafka ít-nhất-một-lần (at-least-once).

Nếu một service trừ tiền trong database nhưng thất bại khi đẩy sự kiện MoneyDeducted lên Kafka do lỗi mạng, hệ thống sẽ bị bất nhất vĩnh viễn.

Kiến trúc Triển khai

sequenceDiagram
    participant App as Go Service
    participant DB as PostgreSQL
    participant Worker as Outbox Relay
    participant Broker as Kafka
    
    App->>DB: BEGIN TX
    App->>DB: INSERT ledger_events
    App->>DB: INSERT outbox_events
    App->>DB: COMMIT TX
    Worker->>DB: Poll/CDC outbox_events
    Worker->>Broker: Publish Message
    Worker->>DB: Mark as processed

Go Polling Relay với FOR UPDATE SKIP LOCKED

Để quét an toàn các sự kiện outbox từ nhiều instance Go chạy song song mà không bị deadlock, chúng tôi dùng FOR UPDATE SKIP LOCKED của PostgreSQL.

func PollOutbox(ctx context.Context, db *pgxpool.Pool, producer sarama.SyncProducer) error {
    tx, err := db.Begin(ctx)
    if err != nil {
        return err
    }
    defer tx.Rollback(ctx)

    // Chỉ khóa các dòng trả về, bỏ qua các dòng đã bị khóa bởi worker khác
    rows, err := tx.Query(ctx, `
        SELECT id, aggregate_type, event_type, payload 
        FROM outbox_events 
        WHERE processed_at IS NULL 
        ORDER BY created_at ASC 
        LIMIT 50 
        FOR UPDATE SKIP LOCKED`)
    if err != nil {
        return err
    }
    defer rows.Close()

    var eventIDs []uuid.UUID
    for rows.Next() {
        var id uuid.UUID
        var aggType, eventType string
        var payload []byte
        
        if err := rows.Scan(&id, &aggType, &eventType, &payload); err != nil {
            return err
        }
        
        // Push lên Kafka
        _, _, err = producer.SendMessage(&sarama.ProducerMessage{
            Topic: aggType,
            Key:   sarama.StringEncoder(id.String()),
            Value: sarama.ByteEncoder(payload),
        })
        if err != nil {
            return fmt.Errorf("lỗi đẩy tin nhắn: %w", err)
        }
        eventIDs = append(eventIDs, id)
    }

    if len(eventIDs) > 0 {
        _, err = tx.Exec(ctx, `
            UPDATE outbox_events SET processed_at = NOW() WHERE id = ANY($1)`, eventIDs)
        if err != nil {
            return err
        }
    }
    return tx.Commit(ctx)
}

5. Điều phối Saga (Saga Orchestration): Temporal vs. Dapr

Trả lời nhanh: Saga Orchestration điều phối các giao dịch tài chính phân tán bằng các hành động bồi hoàn (compensation) khi xảy ra lỗi. Temporal cung cấp một engine thực thi chuyên dụng, siêu bền bỉ cho các workflow dài hạn, trong khi Dapr Workflows nhúng Durable Task Framework dưới dạng sidecar gọn nhẹ.

Two-Phase Commit (2PC) sẽ khóa các cơ sở dữ liệu và bóp nghẹt thông lượng. Chúng ta phải dùng Saga để đảm bảo tính Nhất quán Cuối cùng (Eventual Consistency - xem hướng dẫn đầy đủ tại Orchestrated Saga Pattern với Temporal).

So sánh các Orchestrator

Quyết định Kiến trúc (Tiêu chuẩn Core Banking 2026):

• Temporal là bắt buộc nếu bạn cần dấu vết kiểm toán theo chuẩn PCI-DSS lâu dài, lưu trữ lịch sử ra S3 và xử lý các workflow nhiều ngày phức tạp (ví dụ: giải ngân thế chấp). Lưu ý Temporal có giới hạn cứng về lịch sử sự kiện (51.200 event), buộc phải dùng Continue-As-New cho các sổ cái tài chính chạy vô hạn.
• Dapr Workflows tối ưu hơn cho các Saga ngắn hạn (như chuyển tiền chéo service) nếu bạn đã và đang dùng Dapr cho routing và pub/sub (hiểu rõ Đánh đổi Tính Nhất Quán của Dapr State Store là điều tối quan trọng ở đây).

Cấu trúc mã Workflow Temporal bằng Go

Temporal thực thi các hoạt động bồi hoàn (compensations) một cách tự nhiên. Trong Go, bạn xây dựng một danh sách (slice) các hàm bồi hoàn và kích hoạt chúng thông qua từ khóa defer nếu workflow thất bại. Đặc biệt, khi một workflow bị hủy bỏ (cancelled), context gốc (ctx) cũng sẽ bị hủy, khiến các hoạt động bồi hoàn trong block defer thất bại do context đã hết hiệu lực. Để khắc phục điều này, chúng ta cần tạo một context không kết nối thông qua workflow.NewDisconnectedContext(ctx) để chạy các activity bồi hoàn một cách an toàn.

func FinancialTransferSaga(ctx workflow.Context, req TransferRequest) (err error) {
    options := workflow.ActivityOptions{
        StartToCloseTimeout: time.Minute,
        RetryPolicy: &temporal.RetryPolicy{MaximumAttempts: 3},
    }
    ctx = workflow.WithActivityOptions(ctx, options)

    var compensations []func(workflow.Context)
    
    // Trì hoãn việc thực thi bồi hoàn bằng cách sử dụng một context ngắt kết nối
    defer func() {
        if err != nil {
            newCtx, _ := workflow.NewDisconnectedContext(ctx)
            for _, comp := range compensations {
                comp(newCtx)
            }
        }
    }()

    // Bước 1: Trừ tiền (Deduct)
    err = workflow.ExecuteActivity(ctx, DeductFundsActivity, req).Get(ctx, nil)
    if err != nil {
        return err
    }
    compensations = append(compensations, func(c workflow.Context) {
        workflow.ExecuteActivity(c, RefundFundsActivity, req).Get(c, nil)
    })

    // Bước 2: Ghi có (Credit)
    err = workflow.ExecuteActivity(ctx, CreditTargetActivity, req).Get(ctx, nil)
    if err != nil {
        return err
    }

    return nil
}

6. Thiết kế API Thanh toán Idempotent bằng Go

Trả lời nhanh: Một API thanh toán idempotent đảm bảo rằng các request giống nhau chỉ được xử lý đúng một lần, ngăn ngừa tình trạng trừ tiền kép. Điều này được thực hiện qua mẫu Key-Check-Execute sử dụng Redis để khóa theo TTL và PostgreSQL unique index để chống trùng vĩnh viễn.

Khi Kafka gửi lại tin nhắn, hoặc client gửi lại request khi bị timeout, API phải an toàn khi được gọi nhiều lần.

1. Check: Client gửi kèm một header Idempotency-Key.
2. Lock: API Go cố gắng lấy một lock trong Redis bằng Lua script (SET NX).
3. Database Constraint: Để an toàn vĩnh viễn, khóa idempotency được chèn vào bảng processed_transactions trong PostgreSQL với ràng buộc UNIQUE. Nếu có một request khác cố gắng chèn cùng khóa, PostgreSQL sẽ từ chối.

7. Khả năng Quan sát: OpenTelemetry trong Sổ cái Phân tán

Trả lời nhanh: Để theo dõi các giao dịch tài chính xuyên suốt các microservice, OpenTelemetry (OTel) context propagation phải được tiêm (inject) vào các header HTTP, thông điệp Kafka, và workflow Temporal, đảm bảo khả năng kiểm toán end-to-end và theo dõi độ trễ.

Trong Go, khi sử dụng thư viện segmentio/kafka-go, không có sẵn các wrapper OTel chuẩn. Chúng ta phải tự xây dựng một TextMapCarrier để ánh xạ các trường context OTel vào kafka.Header.

type KafkaHeaderCarrier struct {
    Headers *[]kafka.Header
}

func (c *KafkaHeaderCarrier) Get(key string) string {
    for _, h := range *c.Headers {
        if h.Key == key {
            return string(h.Value)
        }
    }
    return ""
}

func (c *KafkaHeaderCarrier) Set(key, value string) {
    *c.Headers = append(*c.Headers, kafka.Header{
        Key:   key,
        Value: []byte(value),
    })
}

func (c *KafkaHeaderCarrier) Keys() []string {
    keys := make([]string, len(*c.Headers))
    for i, h := range *c.Headers {
        keys[i] = h.Key
    }
    return keys
}

Bằng cách truyền carrier này một cách tường minh khi đẩy và nhận tin nhắn, transaction ID sẽ chảy liên tục qua toàn bộ kiến trúc, cung cấp dữ liệu cực kỳ quan trọng để giải quyết sự cố.

FAQ