Đồng Bộ Tồn Kho Thời Gian Thực: Kafka, Debezium CDC & Redis Cluster Chống Overselling

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

Answer-first: Đồng bộ tồn kho thời gian thực áp dụng mô hình kiến trúc Speed & Truth: PostgreSQL đóng vai trò là Chân lý Dữ liệu (Truth Layer) phát các sự kiện thay đổi từ Write-Ahead Log (WAL) qua Debezium CDC vào Apache Kafka (phân vùng nghiêm ngặt theo sku_id), trong khi Redis Cluster đóng vai trò là Tầng Tốc độ (Speed Layer) thực thi trừ tồn kho nguyên tử (atomic reservation) bằng Redis Lua script kết hợp Idempotency Key và Hash Tags {sku_id}. Thiết kế này triệt tiêu hoàn toàn rủi ro bán vượt mức (overselling), loại bỏ lỗi nghẽn khóa dòng SQL (SELECT FOR UPDATE), và giải quyết triệt để vấn đề Dual-Write.

📦 Bài viết này thuộc chuyên đề kiến trúc e-commerce quy mô lớn. Xem trọn bộ tại Series Điều Phối Đơn Hàng & Tồn Kho Phân Tán.

1. Thách Thức Kỹ Thuật: Nghịch Lý Dual-Write & Sự Sụp Đổ Của Database Locking

Khi phục vụ một sự kiện Flash Sale quy mô hàng trăm ngàn lượt người dùng đổ vào tranh mua vài ngàn sản phẩm trong vòng 10 giây đầu tiên, kiến trúc CRUD nguyên khối truyền thống sẽ nhanh chóng rơi vào trạng thái tê liệt.

1.1. Thảm Họa Của Cơ Chế Khóa Dòng SQL (SELECT ... FOR UPDATE)

Trong các hệ thống e-commerce cổ điển, để ngăn chặn việc bán quá số lượng kho (overselling), nhà phát triển thường sử dụng transaction có khóa dòng:

BEGIN TRANSACTION;
-- Khóa cứng dòng tồn kho của sản phẩm
SELECT stock FROM inventory WHERE sku_id = 'SKU-IPHONE-16' FOR UPDATE;

-- Nếu stock >= quantity, thực hiện trừ kho và tạo đơn hàng
UPDATE inventory SET stock = stock - 1 WHERE sku_id = 'SKU-IPHONE-16';
INSERT INTO orders (order_id, sku_id, quantity) VALUES (...);
COMMIT;

Khi có 5.000 requests/giây cùng tranh nhau mua SKU này, database engine (MySQL hoặc PostgreSQL) buộc phải xếp hàng 5.000 giao dịch này tuần tự phía sau một dòng khóa vật lý duy nhất. Thời gian chờ giữ khóa (lock wait timeout) bùng nổ, connection pool của ứng dụng bị cạn kiệt chỉ trong 2 giây, kéo theo hàng loạt lỗi HTTP 504 Gateway Timeout trên toàn bộ hệ thống bán hàng.

1.2. Nghịch Lý Dual-Write (Ghi Song Trùng)

Một phản xạ tự nhiên của kỹ sư là: “Hãy trừ tồn kho trên Redis trước để lấy tốc độ, rồi sau đó lưu vào SQL database!”

// ANTI-PATTERN: Dual-write gây lệch dữ liệu chết người
func Checkout(sku string, qty int) error {
    // Bước 1: Trừ kho trên Redis
    if err := redisClient.DecrBy(ctx, "stock:"+sku, qty); err != nil {
        return err
    }
    // Bước 2: Lưu đơn hàng vào PostgreSQL
    if err := postgresDB.SaveOrder(ctx, order); err != nil {
        // Chuyện gì xảy ra nếu tiến trình bị kill ở đây, hoặc mạng bị ngắt?
        // Redis đã bị trừ, nhưng PostgreSQL không có đơn hàng -> MẤT TỒN KHO ẢO!
        return err
    }
    return nil
}

Theo Định lý CAP và bản chất của hệ thống phân tán, hai hệ thống lưu trữ độc lập (Redis và PostgreSQL) không thể đảm bảo tính toàn vẹn giao dịch nguyên tử nếu không sử dụng giao thức Two-Phase Commit (2PC) vốn cực kỳ chậm chạp. Nếu mạng bị phân mảnh hoặc Pod ứng dụng bị OOMKilled ngay sau bước trừ Redis, dữ liệu giữa hai bên sẽ bị phân kỳ (data drift) vĩnh viễn.

2. Kiến Trúc Tốc Độ & Sự Thật (Speed & Truth Architecture)

Để giải quyết triệt để bài toán này, các hệ sinh thái thương mại điện tử hàng đầu áp dụng mô hình Speed & Truth:

flowchart TD
    Client["Ứng dụng Khách Hàng / Mobile App"]
    
    subgraph SpeedLayer ["Tầng Tốc Độ (Speed Layer - In-Memory)"]
        RedisCluster[("Redis Cluster")]
        LuaReserve["Lua Script Trừ Kho Nguyên Tử + Idempotency Key"]
        RedisCluster --- LuaReserve
    end
    
    subgraph OrderIngress ["Tầng Tiếp Nhận Giao Dịch"]
        APIGateway["API Gateway (Envoy)"]
        OrderService["Order Service (Go)"]
    end

    subgraph EventBackbone ["Xương Sống Sự Kiện (Event Backbone)"]
        KafkaOrderTopic[["Kafka Topic: order.reserved (Partitioned by SKU)"]]
        DebeziumCDC["Debezium CDC Engine"]
        KafkaSyncTopic[["Kafka Topic: inventory.cdc.events"]]
    end

    subgraph TruthLayer ["Tầng Chân Lý (Truth Layer - Durable SQL)"]
        PGPrimary[("PostgreSQL Primary (WAL Engine)")]
        InventoryWorker["Inventory Settlement Worker (Go)"]
    end

    Client -->|"1. Đặt hàng (POST /checkout)"| APIGateway
    APIGateway --> OrderService
    OrderService -->|"2. Thực thi Lua Script"| RedisCluster
    
    RedisCluster -->|"3a. Đủ Tồn Kho (Token ID)"| KafkaOrderTopic
    OrderService -->|"3b. Phản hồi đặt hàng thành công"| Client
    KafkaOrderTopic -->|"4. Consume Batch"| InventoryWorker
    InventoryWorker -->|"5. BATCH UPDATE"| PGPrimary
    
    RedisCluster -.->|"Lỗi INSUFFICIENT_STOCK"| OrderService
    OrderService -.->|"Trả về HTTP 409 Sold Out"| Client

    PGPrimary -.->|"6. Logical Decoding WAL"| DebeziumCDC
    DebeziumCDC -->|"7. Stream Event"| KafkaSyncTopic

2.1. Phân Tách Trách Nhiệm Giữa Hai Tầng

1. Tầng Chân Lý (Truth Layer - PostgreSQL):
   • Lưu trữ bản ghi pháp lý cuối cùng của tồn kho vật lý và đơn hàng.
   • Không trực tiếp phục vụ các truy vấn kiểm tra tồn kho tần suất cao trong giờ cao điểm.
   • Nhận các lệnh ghi cập nhật tồn kho theo lô (batch updates) từ hàng đợi tin nhắn Kafka.
2. Xương Sống Sự Kiện (Event Backbone - Debezium CDC & Apache Kafka):
   • Đọc trực tiếp Write-Ahead Log (WAL) của PostgreSQL mà không gây tải truy vấn lên CPU database.
   • Phân vùng (partition) topic theo chính xác sku_id. Mọi thay đổi của một mặt hàng được gom vào một partition duy nhất, đảm bảo tính thứ tự tuần tự tuyệt đối (Strict Ordering).
3. Tầng Tốc Độ (Speed Layer - Redis Cluster):
   • Lưu trữ trạng thái số dư tồn kho khả dụng tức thời.
   • Xử lý các phép trừ tồn kho nguyên tử bằng mã Lua Script với độ trễ dưới 1 mili-giây (sub-millisecond).

3. Trừ Tồn Kho Nguyên Tử & Lũy Đẳng (Atomic & Idempotent) Trên Redis Cluster

Khi sử dụng Redis Cluster, hai vấn đề kỹ thuật lớn nhất cần phải giải quyết là:

1. Lỗi Cross-Slot (CROSSSLOT Keys in request don't hash to the same slot): Các lệnh Lua script thao tác trên nhiều key sẽ văng lỗi nếu các key đó rơi vào các Redis hash slot khác nhau.
2. Trùng Lặp Tin Nhắn (Message Duplication): Do Kafka cam kết phân phối tin nhắn theo cơ chế Ít nhất một lần (At-least-once delivery), nếu consumer xử lý lại một message, hệ thống có thể bị trừ kho hai lần nếu thiếu tính lũy đẳng.

3.1. Kỹ Thuật Hash Tag {sku_id} Định Tuyến Cùng Hash Slot

Bằng cách bọc mã định danh SKU vào trong cặp ngoặc nhọn {}, Redis Cluster sẽ chỉ băm chuỗi ký tự bên trong cặp ngoặc này để xác định Hash Slot (từ 0 đến 16383). Nhờ vậy, key tồn kho và key token lũy đẳng của cùng một SKU chắc chắn nằm trên cùng một node vật lý:

• Key tồn kho: stock:{SKU-IPHONE-16}
• Key lũy đẳng: idempotent:{SKU-IPHONE-16}:ord_998811

3.2. Mã Nguồn Redis Lua Script Chuẩn Production

-- ========================================================================
-- LUA SCRIPT: TRỪ TỒN KHO NGUYÊN TỬ VÀ KIỂM TRA LŨY ĐẲNG
-- KEYS[1]: Key Tồn kho (Ví dụ: "stock:{SKU-IPHONE-16}")
-- KEYS[2]: Key Lũy đẳng đơn hàng (Ví dụ: "idempotent:{SKU-IPHONE-16}:ORD-7788")
-- ARGV[1]: Số lượng cần trừ (Quantity)
-- ARGV[2]: Thời gian sống của Idempotency Key tính bằng giây (TTL, ví dụ 86400)
-- ========================================================================

-- Bước 1: Kiểm tra xem đơn hàng này đã từng được xử lý hay chưa
if redis.call("EXISTS", KEYS[2]) == 1 then
    return {err = "ORDER_ALREADY_PROCESSED"}
end

-- Bước 2: Kiểm tra số dư tồn kho hiện tại
local current_stock = tonumber(redis.call("GET", KEYS[1]) or "0")
local requested_qty = tonumber(ARGV[1])

if current_stock < requested_qty then
    return {err = "INSUFFICIENT_STOCK"}
end

-- Bước 3: Thực hiện trừ tồn kho nguyên tử
redis.call("DECRBY", KEYS[1], requested_qty)

-- Bước 4: Đánh dấu đơn hàng đã được giữ chỗ và set TTL 24 giờ để tự động dọn dẹp RAM
redis.call("SET", KEYS[2], "RESERVED", "EX", tonumber(ARGV[2]))

-- Trả về số dư còn lại sau khi trừ
return current_stock - requested_qty

4. Xử Lý Hot SKU: Kỹ Thuật Virtual Inventory Buckets

Khi một sản phẩm siêu hot (như vé xem ca nhạc hoặc iPhone giá sốc 1K) được mở bán, hàng trăm ngàn lượt request sẽ dồn vào đúng một key Redis duy nhất. Vì Redis xử lý lệnh đơn luồng (Single-Threaded Event Loop), một CPU core của Redis Node chứa slot đó sẽ bị quá tải 100%, trong khi các core khác và các node khác trong cụm hoàn toàn nhàn rỗi.

Giải pháp kiến trúc: Băm nhỏ tồn kho ảo (Virtual Inventory Buckets).

flowchart LR
    TotalStock["10.000 Sản Phẩm Cần Bán"] --> Split["Chia Đều Thành 10 Buckets"]
    Split --> B0["stock:{SKU}_b0 (1.000 cái) -> Node 1"]
    Split --> B1["stock:{SKU}_b1 (1.000 cái) -> Node 2"]
    Split --> B2["stock:{SKU}_b2 (1.000 cái) -> Node 3"]
    Split --> B3["... Buckets Khác ... -> Node 4..N"]

    ClientReq["Yêu Cầu Mua Hàng"] --> RandomHash["Hash(UserID + Timestamp) % 10"]
    RandomHash --> B0
    RandomHash --> B1
    RandomHash --> B2

Khi người dùng gửi request, hệ thống sẽ chọn ngẫu nhiên một bucket để trừ:

• Giảm tải CPU contention xuống 1/10.
• Nếu bucket được chọn bị hết hàng tạm thời (INSUFFICIENT_STOCK), client sẽ tự động thử lại (retry) sang bucket kế tiếp trong vòng vài mili-giây.

5. Mã Nguồn Production: Go High-Throughput Inventory Consumer & Lua Client

Dưới đây là mã nguồn Go hoàn chỉnh của worker xử lý tồn kho, tích hợp Kafka Consumer Group, thực thi Lua script nguyên tử, cơ chế Exponential Backoff và Dead-Letter Queue (DLQ) khi gặp sự cố bất thường:

package inventory

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"time"

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

// ReservationMessage định nghĩa cấu trúc thông điệp từ Kafka
type ReservationMessage struct {
	OrderID     string `json:"order_id"`
	SKUID       string `json:"sku_id"`
	Quantity    int    `json:"quantity"`
	RequestedAt int64  `json:"requested_at"`
}

type InventoryWorker struct {
	redisClient *redis.ClusterClient
	kafkaReader *kafka.Reader
	dlqWriter   *kafka.Writer
	luaSHA      string
}

const reservationLuaScript = `
if redis.call("EXISTS", KEYS[2]) == 1 then
    return {err = "ORDER_ALREADY_PROCESSED"}
end
local current_stock = tonumber(redis.call("GET", KEYS[1]) or "0")
local requested_qty = tonumber(ARGV[1])
if current_stock < requested_qty then
    return {err = "INSUFFICIENT_STOCK"}
end
redis.call("DECRBY", KEYS[1], requested_qty)
redis.call("SET", KEYS[2], "RESERVED", "EX", tonumber(ARGV[2]))
return current_stock - requested_qty
`

func NewInventoryWorker(redisAddrs []string, kafkaBrokers []string, topic, groupID string) (*InventoryWorker, error) {
	rdb := redis.NewClusterClient(&redis.ClusterOptions{
		Addrs:        redisAddrs,
		DialTimeout:  2 * time.Second,
		ReadTimeout:  1 * time.Second,
		WriteTimeout: 1 * time.Second,
		PoolSize:     200,
	})

	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	if err := rdb.Ping(ctx).Err(); err != nil {
		return nil, fmt.Errorf("không thể kết nối tới Redis Cluster: %w", err)
	}

	// Tải trước Lua script vào Redis script cache để gọi qua EVALSHA (tối ưu băng thông)
	sha, err := rdb.ScriptLoad(ctx, reservationLuaScript).Result()
	if err != nil {
		return nil, fmt.Errorf("lỗi tải Lua script: %w", err)
	}

	reader := kafka.NewReader(kafka.ReaderConfig{
		Brokers:        kafkaBrokers,
		Topic:          topic,
		GroupID:        groupID,
		MinBytes:       10e3, // 10KB
		MaxBytes:       10e6, // 10MB
		CommitInterval: 1 * time.Second,
	})

	dlqWriter := &kafka.Writer{
		Addr:     kafka.TCP(kafkaBrokers...),
		Topic:    topic + ".dlq",
		Balancer: &kafka.LeastBytes{},
	}

	return &InventoryWorker{
		redisClient: rdb,
		kafkaReader: reader,
		dlqWriter:   dlqWriter,
		luaSHA:      sha,
	}, nil
}

// StartProcessingLoop lắng nghe và xử lý sự kiện đặt hàng từ Kafka
func (w *InventoryWorker) StartProcessingLoop(ctx context.Context) {
	log.Println("Bắt đầu vòng lặp xử lý tồn kho thời gian thực...")

	for {
		select {
		case <-ctx.Done():
			log.Println("Đang dừng worker gracefully...")
			w.kafkaReader.Close()
			w.dlqWriter.Close()
			return
		default:
			msg, err := w.kafkaReader.FetchMessage(ctx)
			if err != nil {
				if errors.Is(err, context.Canceled) {
					return
				}
				log.Printf("Lỗi đọc Kafka message: %v", err)
				time.Sleep(100 * time.Millisecond)
				continue
			}

			if err := w.processReservation(ctx, msg.Value); err != nil {
				log.Printf("Xử lý đơn hàng thất bại: %v. Đang đẩy vào Dead-Letter Queue...", err)
				w.sendToDLQ(ctx, msg.Key, msg.Value, err)
			}

			// Commit offset sau khi xử lý thành công
			if err := w.kafkaReader.CommitMessages(ctx, msg); err != nil {
				log.Printf("Lỗi commit Kafka offset: %v", err)
			}
		}
	}
}

func (w *InventoryWorker) processReservation(ctx context.Context, data []byte) error {
	var req ReservationMessage
	if err := json.Unmarshal(data, &req); err != nil {
		return fmt.Errorf("lỗi giải mã JSON: %w", err)
	}

	// Sử dụng Hash Tag {} để đảm bảo hai key cùng rơi vào một hash slot
	stockKey := fmt.Sprintf("stock:{%s}", req.SKUID)
	idempotentKey := fmt.Sprintf("idempotent:{%s}:%s", req.SKUID, req.OrderID)
	ttlSeconds := 86400 // 24 giờ

	// Thực thi qua EvalSha để đạt hiệu năng tối đa
	res, err := w.redisClient.EvalSha(ctx, w.luaSHA, []string{stockKey, idempotentKey}, req.Quantity, ttlSeconds).Result()
	if err != nil {
		// Nếu script bị mất khỏi cache Redis, tải lại và thực thi qua Eval thông thường
		if isNoScriptErr(err) {
			res, err = w.redisClient.Eval(ctx, reservationLuaScript, []string{stockKey, idempotentKey}, req.Quantity, ttlSeconds).Result()
		}
		if err != nil {
			return fmt.Errorf("lỗi thực thi trừ kho Redis: %w", err)
		}
	}

	log.Printf("[THÀNH CÔNG] Giữ chỗ SKU %s cho Đơn %s. Số dư còn lại: %v", req.SKUID, req.OrderID, res)
	return nil
}

func (w *InventoryWorker) sendToDLQ(ctx context.Context, key, val []byte, cause error) {
	dlqPayload := map[string]interface{}{
		"original_payload": string(val),
		"error_reason":     cause.Error(),
		"failed_at":        time.Now().UTC().Format(time.RFC3339),
	}
	bytes, _ := json.Marshal(dlqPayload)

	_ = w.dlqWriter.WriteMessages(ctx, kafka.Message{
		Key:   key,
		Value: bytes,
	})
}

func isNoScriptErr(err error) bool {
	return err != nil && (err.Error() == "NOSCRIPT No matching script. Please use EVAL.")
}

6. Kịch Bản Đối Soát & Phục Hồi Sau Thảm Họa (Disaster Recovery)

Một hệ thống phân tán không bao giờ hoàn hảo 100%. Các sự cố mạng, restart đột ngột của Redis pod, hoặc tràn bộ nhớ đệm có thể làm lệch số dư giữa Redis và PostgreSQL.

6.1. Tiến Trình Đối Soát Bất Đồng Bộ (Reconciliation Daemon)

Định kỳ mỗi 30 phút, một cronjob ngầm sẽ quét số liệu giữa hai tầng:

1. Tính toán tổng tồn kho thực tế trong PostgreSQL: $$\text{Stock}_{\text{Actual}} = \text{Total Inventory} - \sum \text{Orders (Pending/Shipped)}$$
2. Đọc số dư hiện tại từ Redis: Stock_Redis = GET stock:{sku_id}.
3. Nếu phát hiện sai lệch $|\text{Stock}{\text{Actual}} - \text{Stock}{\text{Redis}}| > 0$:
   • Phát cảnh báo PagerDuty mức vàng.
   • Sử dụng lệnh nguyên tử SET stock:{sku_id} Stock_Actual để đồng bộ lại tầng Speed Layer.

6.2. Kịch Bản Tái Tạo Toàn Bộ Cache (Cold-Start Cache Rebuilding)

Nếu cụm Redis Cluster bị sập hoàn toàn và mất toàn bộ dữ liệu RAM:

1. Đóng cổng API Gateway để tạm ngưng tiếp nhận traffic mua hàng mới (chuyển sang màn hình bảo trì 60 giây).
2. Chạy script khôi phục: Quét toàn bộ bảng inventory trên PostgreSQL bằng Streaming Cursor, tính toán trừ đi các đơn hàng đang nằm chờ trong Kafka topic chưa được flush xuống database.
3. Sử dụng Redis Pipeline nạp 5 triệu SKU vào Redis Cluster trong vòng 45 giây.
4. Mở lại API Gateway cho người dùng bình thường.

7. Bảng So Sánh Các Mô Hình Đồng Bộ Tồn Kho

8. Câu Hỏi Thường Gặp (FAQ)

9. Tài Liệu Kỹ Thuật Liên Quan

• Kiến Trúc Shopee Flash Sale: Tìm hiểu cách Shopee xử lý hàng triệu TPS trong đêm 11.11 tại Kiến Trúc Shopee Flash Sale: Chống Tải Đột Biến & Redis Cache-Ahead.
• Tối Ưu Hiệu Năng Go pprof: Hướng dẫn chẩn đoán cấp phát bộ nhớ và CPU cho Kafka workers tại Hướng Dẫn Go pprof CPU & Bộ Nhớ Trên Production.
• Phát Hiện Goroutine Leak: Xử lý rò rỉ goroutine trong các service xử lý stream sự kiện tại Phát Hiện và Xử Lý Goroutine Leak trên Go Services.
• Mở Rộng Cơ Sở Dữ Liệu MySQL: Kỹ thuật sharding và tối ưu ghi dữ liệu lớn tại MySQL Horizontal Scaling: Vitess & GORM Sharding.
• Series Điều Phối Đơn Hàng: Xem toàn bộ 10 phần chuyên sâu tại Series Kiến Trúc Điều Phối Đơn Hàng & Tồn Kho Phân Tán.