Go Zero to Hero - Bài 68: Concurrency Pattern: Pipeline

1. Pipeline là gì? Triết lý Dây chuyền Lắp ráp

Pipeline (Đường ống/Dây chuyền) là một chuỗi các công đoạn (Stages) được nối với nhau bằng các ống nước (Channels).

Đặc điểm nhận diện một Pipeline chuẩn mực:

• Độc lập: Mỗi công đoạn là một Goroutine chạy song song.
• Đầu vào / Đầu ra: Đầu ra của công đoạn này (chan) chính là đầu vào của công đoạn tiếp theo.
• Hàm tạo luồng: Mỗi công đoạn thường là một hàm nhận vào một Receive-only channel (<-chan) và trả về một Receive-only channel (<-chan) khác.

Lợi ích tối thượng:

• Streaming Data: Bạn không cần tải toàn bộ 1 triệu dòng log vào RAM rồi mới xử lý. Nước chảy đến đâu, bạn lọc đến đó. RAM luôn ở mức ổn định (Memory Footprint siêu nhỏ).
• Decoupling (Giảm phụ thuộc): Các bước tách biệt hoàn toàn, rất dễ viết Unit Test cho từng công đoạn.

2. Thiết kế Dây chuyền xử lý vé AFC

Hãy xây dựng một hệ thống xử lý giao dịch vé tàu điện qua 3 công đoạn: Sinh dữ liệu (Đầu đọc) ➔ Lọc vé lỗi (Validate) ➔ Đóng gói thông tin (Enrich) ➔ In kết quả (Terminal).

package main

import (
    "fmt"
    "strings"
    "time"
)

// CÔNG ĐOẠN 1: GENERATOR (Nguồn phát)
// Tạo ra dữ liệu thô và đẩy vào ống nước
func readRawEvents(rawEvents []string) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out) // Xong việc phải đóng van
        for _, event := range rawEvents {
            out <- event
        }
    }()
    return out // Trả về đầu ra cho công đoạn sau
}

// CÔNG ĐOẠN 2: FILTER (Bộ lọc)
// Loại bỏ các thẻ rác, thẻ lỗi
func filterValidCards(in <-chan string) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for event := range in {
            // Nghiệp vụ: Chỉ chấp nhận mã thẻ bắt đầu bằng "AFC"
            if strings.HasPrefix(event, "AFC") {
                out <- event
            } else {
                fmt.Println("🚫 Bỏ qua thẻ lỗi/thẻ rác:", event)
            }
        }
    }()
    return out
}

// CÔNG ĐOẠN 3: ENRICH (Làm giàu dữ liệu)
// Đóng gói thêm thông tin trạm và thời gian
func enrichTransaction(in <-chan string) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for cardID := range in {
            // Giả lập việc gọi Cache/DB để lấy thêm thông tin tốn 100ms
            time.Sleep(100 * time.Millisecond)
            out <- fmt.Sprintf("Giao dịch [%s] tại Trạm Trung Tâm", cardID)
        }
    }()
    return out
}

func main() {
    fmt.Println("--- BẬT CẦU DAO DÂY CHUYỀN ---")

    // Dữ liệu thô từ các thiết bị quét trả về
    rawStream := []string{"AFC-1001", "RÁC-002", "AFC-1003", "FAKE-004", "AFC-1005"}

    // LẮP RÁP CÁC ĐƯỜNG ỐNG (Chaining)
    // Code cực kỳ tường minh và đậm chất Clean Code!
    c1 := readRawEvents(rawStream)
    c2 := filterValidCards(c1)
    c3 := enrichTransaction(c2)

    // CÔNG ĐOẠN 4: SINK (Tiêu thụ cuối)
    // Luồng main hứng những giọt nước tinh khiết cuối cùng
    for result := range c3 {
        fmt.Println("✅ Hoàn tất lưu:", result)
    }

    fmt.Println("--- DÂY CHUYỀN NGỪNG HOẠT ĐỘNG ---")
}

Nhìn vào khối lệnh lồng nhau enrichTransaction(filterValidCards(readRawEvents(rawStream))), bạn có thể thấy dòng chảy dữ liệu rõ ràng như ban ngày. Khi luồng rawStream hết nước, c1 đóng. Hàm Filter thấy c1 đóng bèn thoát vòng lặp và đóng c2. Hàm Enrich thấy c2 đóng cũng nghỉ làm và đóng c3. Cuối cùng main kết thúc êm đẹp.

3. Bẫy Rò Rỉ Goroutine (Goroutine Leak) trong Pipeline

Đoạn code trên chạy rất mượt vì chúng ta đã duyệt hết toàn bộ dữ liệu cho đến khi ống cạn. Nhưng điều gì xảy ra nếu Công đoạn cuối cùng muốn dừng lại sớm?

Giả lập trường hợp ở hàm main, sau khi nhận được 2 vé thành công đầu tiên, hệ thống phát hiện lỗi và quyết định break thoát khỏi vòng lặp for range:

1. Hàm main không đọc c3 nữa.
2. Công đoạn 3 (Enrich) bị Khóa (Block) vì cố nhét đồ vào c3 mà không có ai nhận. Nó treo vĩnh viễn!
3. Công đoạn 2 (Filter) cũng bị kẹt lại khi cố nhét đồ vào c2...

Kết quả: Toàn bộ dây chuyền phía sau bị treo cứng trong RAM, gây ra hiện tượng rò rỉ Goroutine (Goroutine Leak).

4. Tuyệt kỹ giải quyết: Pipeline kết hợp Context

Để đảm bảo mọi công đoạn trong Pipeline đều có thể tự hủy ngang chừng một cách an toàn mà không bị rò rỉ bộ nhớ, chúng ta phải áp dụng "nút bấm tự hủy" context.WithCancel đã học ở Bài 65.

Chúng ta nâng cấp hàm công đoạn như sau:

// Truyền ctx vào mọi công đoạn
func filterValidCardsSafe(ctx context.Context, in <-chan string) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for {
            // Dùng select để vừa nghe ống dữ liệu, vừa nghe lệnh bài tử hình
            select {
            case <-ctx.Done():
                // Vị tướng hô "Rút lui", lập tức bỏ việc!
                return 
            case event, ok := <-in:
                if !ok {
                    return // Ống in đã đóng, hết việc
                }
                
                // Cố gắng đẩy dữ liệu vào ống out, 
                // nhưng cũng phải canh chừng lệnh hủy để không bị Block!
                if strings.HasPrefix(event, "AFC") {
                    select {
                    case out <- event:
                    case <-ctx.Done():
                        return
                    }
                }
            }
        }
    }()
    return out
}

Nhờ select { case <-ctx.Done() }, bất kể công đoạn nào bị kẹt, chỉ cần hàm main phát lệnh cancel(), tất cả Goroutines trong dây chuyền sẽ đồng loạt thả tay và tắt đài, trả lại bộ nhớ sạch sẽ cho hệ thống.

💡 Tổng kết

Pipeline là một nghệ thuật chia để trị trong luồng dữ liệu phân tán:

• Kiến trúc: Chia nghiệp vụ phức tạp thành nhiều hàm nhỏ, mỗi hàm chạy một Goroutine và nối với nhau bằng Channel.
• Streaming: Rất phù hợp để xử lý các tệp dữ liệu lớn (CSV, Log file, Message Queue) mà không làm tràn RAM.
• Phòng vệ: Một Pipeline chuẩn mực Backend BẮT BUỘC phải có context kẹp vào từng công đoạn để dọn dẹp Goroutine trong trường hợp dừng khẩn cấp.