Message Queue: Nạp task vào hàng đợi, worker xử lý nền mau lẹ | Blog

Giả sử anh đang viết một API gửi email xác nhận đơn hàng, rồi tạo PDF hóa đơn, rồi gọi webhook bên thứ ba. Nếu cứ xử lý tuần tự ngay trong request, chuyện gì xảy ra?

Response chậm ngất — khách chờ mấy giây chỉ vì một thằng email lì lợm.
Dễ mất việc — process bị restart giữa chừng là mất luôn email, mất luôn hóa đơn.
Không scale được — muốn xử lý nhanh hơn chỉ có cách đợi request tới nhiều hơn.

Giải pháp chuẩn là đưa task vào hàng đợi (queue), trả về 202 Accepted ngay cho client, rồi để một hay nhiều worker xử lý nền. Kiến trúc này tách biệt "nhận yêu cầu" khỏi "tốn thời gian làm".

Viết worker đơn giản trong Go

Cần ví dụ cụ thể, em làm với một hàng đợi trong memory đơn giản (production thì xài Redis Streams, SQS, RabbitMQ tùy nhu cầu):

type Queue struct {
    ch chan Job
}

func NewQueue(size int) *Queue {
    return &Queue{ch: make(chan Job, size)}
}

func (q *Queue) Enqueue(j Job) {
    q.ch <- j // request chỉ cần ném vô đây, trả về ngay
}

func (q *Queue) RunWorkers(n int, fn func(Job) error) {
    for i := 0; i < n; i++ {
        go func() {
            for j := range q.ch {
                if err := fn(j); err != nil {
                    // đừng nuốt lỗi — đẩy qua xử lý retry hoặc DLQ
                    log.Printf("job %d fail: %v", j.ID, err)
                }
            }
        }()
    }
}

Cái make(chan Job, size) với buffer chính là backpressure — khi hàng đợi đầy, caller sẽ phải chờ, không cho load tăng vô hạn làm sập memory.

Retry + backoff: lỗi tạm thời thì thử lại

Webhook của bên thứ ba hay bị 5xx tạm thời. Worker cần retry với exponential backoff + jitter. Code:

func retryWithBackoff(fn func() error, max int) error {
    for attempt := 0; attempt < max; attempt++ {
        err := fn()
        if err == nil {
            return nil
        }
        delay := time.Duration(1<<attempt) * time.Second
        delay += time.Duration(rand.Intn(200)) * time.Millisecond // jitter
        time.Sleep(delay)
    }
    return errors.New("exhausted retries")
}

Em nhấn mạnh jitter — nếu 100 job thất bại cùng lúc và cùng ngủ 2 giây, sau đó toàn bộ "dồn cục" (thundering herd) đánh sập service yếu. Rải ngẫu nhiên chút xíu đỡ hơn nhiều.

Dead-letter queue (DLQ): thất bại cuối cùng đừng mất

Khi retry hết mức vẫn fail, đừng drop việc đó. Đẩy qua hàng đợi "xác chết" để sau này soi nguyên nhân hoặc replay thủ công:

func (w *Worker) handle(j Job) {
    if err := w.process(j); err != nil {
        w.dlq.Enqueue(j) // lưu lại cho người gỡ rối
    }
}

Kinh nghiệm thực tế của em khi build automation: task không có idempotency-key là thảm họa. Worker retry tức là cùng một tác vụ chạy lại nhiều lần. Nếu không đảm bảo chạy lặp không gây tác dụng phụ, anh sẽ gửi email trùng, trừ tiền trùng. Trước khi cho worker retry tự động, phải chắc chắn consumer là idempotent — ví dụ lưu job_id đã xử lý ở DB và bỏ qua nếu gặp lại.

Tóm lại

Request HTTP chỉ nên làm việc nhẹ, cái gì nặng đẩy vô queue.
Dùng nhiều worker để scale theo giá trị thực, không scale theo request.
Luôn retry + backoff + jitter, và có DLQ cho ca bó tay.
Consumer phải idempotent trước khi tự retry.

Queue + worker là nền tảng của mọi hệ thống gửi mail, xử lý ảnh, sync dữ liệu. Nắm được mô hình này thì đọc RabbitMQ, SQS, hay Redis Streams đều thấy quen quen.