Bài 7: Pipeline & Transaction - Tối ưu Network RTT và đảm bảo tính nguyên vẹn

Mục tiêu bài học:

• Nhận diện nút thắt cổ chai (bottleneck) do Network Round-Trip Time (RTT) khi gọi hàng ngàn lệnh tuần tự.

• Sử dụng Pipeline để gom nhóm hàng loạt lệnh vào một request duy nhất.

• Sử dụng TxPipeline và lệnh WATCH để đảm bảo tính nguyên vẹn dữ liệu (Atomicity) theo cơ chế Khóa lạc quan (Optimistic Locking).

1. Network RTT và Giải pháp Pipeline

Redis xử lý các lệnh trên bộ nhớ (RAM) với tốc độ microsecond. Tuy nhiên, thời gian để gói tin đi từ Application Server đến Redis Server qua mạng (RTT) thường mất vài millisecond. Nếu bạn chạy vòng lặp for 1000 lần để ghi 1000 keys, bạn phải chịu chi phí mạng 1000 lần.

Pipeline giải quyết bằng cách: Application gom toàn bộ 1000 lệnh vào bộ đệm, gửi 1 lần qua mạng. Redis xử lý xong trả về 1 mảng 1000 kết quả.

Go

func usePipeline(rdb *redis.Client) {
	// Khởi tạo Pipeline
	pipe := rdb.Pipeline()

	// Ghi các lệnh vào pipeline (chưa gửi đi)
	// Lưu ý: Các hàm gọi qua pipe không trả về lỗi ngay lập tức
	for i := 0; i < 100; i++ {
		key := fmt.Sprintf("batch_user:%d", i)
		pipe.Set(ctx, key, "value", 0)
	}

	// Thực thi toàn bộ pipeline (Gửi 1 lần duy nhất)
	_, err := pipe.Exec(ctx)
	if err != nil {
		fmt.Println("Lỗi khi chạy pipeline:", err)
	} else {
		fmt.Println("Đã ghi 100 keys qua Pipeline thành công!")
	}
}

2. Transaction (TxPipeline)

Pipeline thông thường chỉ gom nhóm gói tin mạng. Nó không đảm bảo tính nguyên vẹn (Atomicity). Nếu có các client khác chèn lệnh vào giữa lúc Redis đang xử lý Pipeline của bạn, các lệnh đó vẫn được chạy xen kẽ.

Để đảm bảo các lệnh của bạn chạy liên tục không bị gián đoạn bởi bất kỳ client nào khác, hãy dùng TxPipeline (tương đương với chuỗi lệnh MULTI ... EXEC trong Redis).

Go

func useTxPipeline(rdb *redis.Client) {
	txPipe := rdb.TxPipeline()

	// Các lệnh trong TxPipeline sẽ được Redis thực thi như một khối duy nhất
	txPipe.Incr(ctx, "counter")
	txPipe.Expire(ctx, "counter", time.Hour)

	// Gửi MULTI, INCR, EXPIRE, EXEC trong 1 request
	_, err := txPipe.Exec(ctx)
	if err != nil {
		fmt.Println("Transaction thất bại:", err)
	}
}

3. Lệnh WATCH (Optimistic Locking)

Trong mô hình giao dịch của SQL, bạn có thể khóa một dòng dữ liệu (Row Lock) để sửa. Redis không có cơ chế khóa tự động đó, mà dùng Optimistic Locking thông qua lệnh WATCH.

Quy trình:

1. WATCH một key.

2. Đọc giá trị hiện tại.

3. Tính toán giá trị mới ở Application.

4. Gửi kết quả mới qua Transaction. Nếu key đó bị một client khác thay đổi trong khoảng thời gian giữa bước 1 và bước 4, Transaction sẽ thất bại (lỗi redis.TxFailedErr), và bạn cần retry.

Ứng dụng thực tế: Chuyển tiền (Mô phỏng Race Condition)

Go

func transferMoney(rdb *redis.Client, from string, to string, amount int) error {
	// Hàm thực thi transaction, go-redis sẽ tự động retry nếu WATCH phát hiện thay đổi
	txf := func(tx *redis.Tx) error {
		// 1. Lấy số dư tài khoản người gửi (Đã tự động được WATCH bởi tx.Watch)
		balanceStr, err := tx.Get(ctx, from).Result()
		if err != nil && err != redis.Nil {
			return err
		}

		balance := 0
		if balanceStr != "" {
			fmt.Sscanf(balanceStr, "%d", &balance)
		}

		if balance < amount {
			return fmt.Errorf("không đủ số dư")
		}

		// 2. Bắt đầu đưa các lệnh thay đổi vào TxPipeline
		_, err = tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
			pipe.DecrBy(ctx, from, int64(amount))
			pipe.IncrBy(ctx, to, int64(amount))
			return nil
		})
		return err
	}

	// Khởi chạy Watch với key cần giám sát
	// go-redis tự động gọi WATCH "from" và chạy hàm txf
	err := rdb.Watch(ctx, txf, from)
	
	if err == redis.TxFailedErr {
		fmt.Println("Giao dịch bị xung đột do có người khác cập nhật số dư cùng lúc, cần thử lại!")
	}
	return err
}