Bài 12: Message Queue & Xử Lý Bất Đồng Bộ

Bài 11 kết luận rằng at-least-once cộng xử lý idempotent là tổ hợp khả thi duy nhất. Message queue là nơi tổ hợp đó không còn là lựa chọn mà là điều kiện bắt buộc: khi một job có thể được giao lại bất cứ lúc nào, consumer phải idempotent — nếu không, mỗi lần redeliver là một side-effect nữa.

Lab của bài này dựng producer/consumer thật trên Redis Streams. Hai phép đo đáng chú ý. Thứ nhất, tăng số consumer từ 1 lên 4 chỉ làm nhanh 2,68 lần chứ không phải 4 — vì nút cổ chai đã chuyển từ consumer sang chính Redis. Thứ hai, ack sai thời điểm làm 352 job biến mất không dấu vết: số job khôi phục được là 0 thay vì 352.

12.1 Vì sao chuyển sang bất đồng bộ

12.2 Ngữ nghĩa ack — và 352 job biến mất

Toàn bộ độ tin cậy của một message queue nằm ở thời điểm gửi ack. Ack nghĩa là "job này xong rồi, đừng giao lại nữa" — nên gửi nó quá sớm là tự tay bỏ mất khả năng khôi phục.

Cơ chế đứng sau chuyện đó có tên: khi broker giao một job, nó không xoá job đi mà chuyển sang danh sách treo (pending) và bấm giờ. Nếu ack tới trước khi hết giờ, job coi như xong. Nếu không — worker chết, hoặc chỉ đơn giản là quá chậm — broker kết luận worker đã hỏng và giao job cho người khác. Khoảng thời gian chờ đó gọi là visibility timeout, và nó chính là thứ biến "worker chết" từ mất dữ liệu thành một lần giao lại.

Lab đo được thiệt hại. Consumer đọc trước 500 job mỗi lần (prefetch 500), mỗi job xử lý 5 ms — nên một lô mất khoảng 2,5 giây. Ta docker kill consumer sau 1,2 giây, tức là giữa lô:

// BLOCK 1000: wait up to 1s if the stream is empty, instead of spinning and burning CPU.
const res = await redis.cmd('XREADGROUP', 'GROUP', GROUP, CONSUMER,
  'COUNT', String(BATCH), 'BLOCK', '1000', 'STREAMS', STREAM, '>');

for (const [id, fields] of entries) {
  const f = {};
  for (let i = 0; i < fields.length; i += 2) f[fields[i]] = fields[i + 1];

  // Dedup on event_id BEFORE causing any side effect. SET NX is atomic, so if two
  // workers receive the same event only one wins (the precondition of at-least-once).
  const fresh = await redis.cmd(
    'SET', `lab:q:seen:${f.event_id}`, '1', 'NX', 'EX', '3600');
  if (fresh !== 'OK') {
    await redis.cmd('XACK', STREAM, GROUP, id);   // already processed -> ack and skip
    continue;
  }

  await doWork(f);

  // ACK LAST. This single line decides the reliability of the whole system.
  await redis.cmd('XACK', STREAM, GROUP, id);
}

12.3 Poison message và DLQ — và một điều cần nói cho đúng

Poison message là job không bao giờ xử lý được: payload sai định dạng, tham chiếu tới bản ghi đã bị xoá, hoặc chạm đúng một bug. Nó thất bại, được giao lại, thất bại lại — mãi mãi.

Cách xử lý tiêu chuẩn là DLQ — viết tắt của dead letter queue, "hàng đợi thư chết". Nó chỉ là một queue thứ hai: sau khi một job đã thất bại quá số lần cho phép, thay vì để nó quay lại mãi, ta chuyển nó sang queue đó rồi ack ở queue chính. Job không mất, nhưng cũng không còn làm phiền luồng bình thường — nó nằm chờ một con người nhìn tới.

Lab bơm 5.000 job, cứ 500 job chèn một job độc (tổng 10 job độc), một consumer:

12.4 Thứ tự và consumer group — thêm consumer là phá thứ tự

Trước hết là cái tên trong tiêu đề. Consumer group là cơ chế để nhiều consumer cùng đọc một queue mà không giẫm lên nhau: broker nhớ từng job đã giao cho consumer nào, và mỗi job chỉ giao cho đúng một thành viên trong nhóm. Đó là lý do lệnh ở mục 12.2 là XREADGROUP chứ không phải XREAD — thêm một consumer vào nhóm là thêm năng lực xử lý, không phải nhân đôi công việc. Đổi lại chính là vấn đề của mục này.

Thứ tự chỉ được đảm bảo trong phạm vi một partition (hoặc một stream key). Ngay khi bạn thêm consumer thứ hai để scale, các job được xử lý song song và không còn thứ tự toàn cục nào.

Số đo scaling ở lab cho thấy chính xác điều đó — và một điều nữa không ai mong đợi:

Gấp bốn số consumer chỉ nhanh lên 2,68 lần. Và cột "tốc độ mỗi consumer" giải thích lý do: nó tụt từ 11.338 xuống ~7.600. Nút cổ chai đã chuyển từ consumer sang Redis — Redis xử lý lệnh trên một luồng duy nhất, nên bốn consumer đang tranh nhau một tài nguyên tuần tự.

12.5 Backpressure và queue depth

Queue depth là chỉ số sức khoẻ số một của hệ bất đồng bộ, và lý do nằm ở Bài 1: nếu tốc độ vào $\lambda$ lớn hơn tốc độ ra $\mu$ thì depth tăng không giới hạn, và theo Little's Law thời gian chờ cũng vậy. Không có mức tải nào là "an toàn" nếu $\lambda > \mu$ — chỉ có thời gian trước khi hậu quả xuất hiện.

Tự làm lại các phép đo

cd blog/sysdesign/sysdesign-lab
docker compose --profile queue up -d redis

# ./tools/queue-test.sh COUNT N_CONSUMER [WORK_MS] [POISON_EVERY] [MAX_ATTEMPTS] \
#                       [IDEMPOTENT] [ACK_MODE] [DUR] [BATCH]

# --- Scaling: 20,000 jobs, 1 / 2 / 4 consumers. Read `msActive`, not `rate`. ---
./tools/queue-test.sh 20000 1 0 0 0 0 after 15000   # msActive 1764ms · 11,338/s
./tools/queue-test.sh 20000 2 0 0 0 0 after 15000   # msActive ~1201ms · total 16,652/s
./tools/queue-test.sh 20000 4 0 0 0 0 after 15000   # msActive ~658ms  · total 30,375/s

# --- Poison messages: 5,000 jobs, one poison job every 500 ---
./tools/queue-test.sh 5000 1 0 500 0 0 after 15000  # pending=10, stuck forever
./tools/queue-test.sh 5000 1 0 500 1 0 after 15000  # dlq=10 · pending=0

# --- Acking at the wrong moment: kill the consumer MID-BATCH of 500 jobs ---
docker compose exec -T redis redis-cli DEL lab:jobs
docker compose run --rm --no-deps -e ROLE=producer -e COUNT=3000 queueworker queue.js
docker compose run --rm --name qkill --no-deps -e ROLE=consumer -e WORK_MS=5 -e BATCH=500 \
  -e ACK_MODE=on-receive -e DURATION_MS=30000 queueworker queue.js &
sleep 1.2 && docker kill qkill
docker compose exec -T redis redis-cli XPENDING lab:jobs g1 | head -1
# prints 0  -> the 352 unprocessed jobs are gone for good
# Switch ACK_MODE to `after` and repeat: it prints 352, all recoverable

Tóm lại

Bất đồng bộ tách độ trễ người dùng khỏi việc nặng và hấp thụ được burst, nhưng nó đổi hình dạng của vấn đề chứ không xoá vấn đề: đỉnh tải thành tồn đọng, và tồn đọng có cái giá riêng.

Toàn bộ độ tin cậy nằm ở thời điểm ack. Đo thật với prefetch 500 và worker bị kill giữa lô: ack-khi-nhận cho 0 job khôi phục được, ack-sau-khi-xong cho 352. Số job mất tối đa đúng bằng kích thước prefetch — nên prefetch là tham số độ tin cậy, không chỉ là tham số hiệu năng. Và vì at-least-once nghĩa là job có thể xử lý hai lần, consumer bắt buộc phải idempotent (Bài 11).

Về poison message, mình sửa lại một điều thường được nói sai: với Redis Streams nó không chặn message phía sau — 4.990 job khác vẫn chạy bình thường, message hỏng chỉ nằm lại trong danh sách treo. Điều đó làm thiệt hại âm thầm hơn, nên với broker kiểu này thứ bắt buộc phải có là alert theo độ dài danh sách treo, không chỉ là DLQ.

Cuối cùng, gấp bốn số consumer chỉ nhanh lên 2,68 lần — lần thứ tư trong series nút cổ chai dịch sang thành phần dùng chung ngay khi ta scale. Cách kiểm tra rẻ nhất là nhìn tốc độ mỗi worker: nếu nó tụt khi bạn thêm worker, bạn đang chia lại một chiếc bánh cố định.

Bài 13 đi vào van xả cuối cùng trong danh sách ở mục 12.5: rate limiting. Đó là cơ chế để hệ thống nói không một cách có kiểm soát — thay vì nhận hết rồi vỡ, hoặc nhận hết rồi hứa một tồn đọng sáu giờ.