Hàng đợi message

Hàng đợi message (RabbitMQ, Amazon SQS) dùng để chia việc: service nhận request đặt việc vào hàng đợi rồi trả lời ngay, còn worker lấy việc ra xử lý theo tốc độ của mình. Khác với Kafka, message trong hàng đợi bị xoá khi đã xử lý xong, và mỗi message được theo dõi riêng: ai đang giữ nó, đã giao bao nhiêu lần, có nên giao lại không (Kafka). Trang này giải thích cách hàng đợi biết một message đã xong, cách chia việc cho consumer nhanh và chậm, cách xử lý message lỗi mãi, thứ tự, và khi consumer chậm hơn producer. Khi lượng việc vừa phải, một bảng trong database làm hàng đợi là đủ và tránh được một hệ thống phải vận hành thêm (Hàng đợi bằng database).

Queue và pub/sub

Có hai cách giao message:

Queue (point-to-point): nhiều consumer cùng đọc một queue, mỗi message chỉ đến một consumer. Đây là cách chia việc cho nhiều worker. Trên lab, hai consumer cùng một queue nhận 52 và 48 trên 100 message.
Pub/sub: mỗi bên quan tâm có một queue riêng, và mỗi message được chép vào mọi queue đó. Đây là cách báo một sự kiện cho nhiều service, như "đơn đã tạo" cho service gửi email và service thống kê. Trên lab, một exchange kiểu fanout gửi 100 message thì queue email và queue thống kê đều nhận đủ 100.

Trong RabbitMQ, producer gửi message vào một exchange, rồi exchange quyết định chép message vào queue nào theo luật của nó:

Exchange	Chép message vào
`direct`	Queue có routing key khớp đúng
`topic`	Queue có mẫu khớp routing key, ví dụ `order.*.created`
`fanout`	Mọi queue gắn vào exchange
`headers`	Queue khớp theo header của message

Với AWS, pub/sub thường làm bằng SNS: một topic SNS đẩy message tới nhiều queue SQS, mỗi service đọc queue của mình.

Ack: khi nào message được coi là xong

Hàng đợi chỉ xoá message khi biết chắc nó đã được xử lý. Có hai cách để biết.

RabbitMQ: ack

Consumer giữ connection với broker, broker đẩy message tới, consumer xử lý xong thì gửi ack. Message đã giao mà chưa ack thì broker vẫn giữ. Nếu connection đứt hay consumer chết, broker tự đưa các message chưa ack về queue để giao cho consumer khác, kèm cờ redelivered.

Chế độ auto-ack (noAck) bỏ bước này. Theo tài liệu RabbitMQ, ở chế độ này message được coi là giao thành công ngay khi gửi đi, tức kiểu "gửi rồi quên". Trên lab, 100 message, consumer bị kill -9 sau khi xử lý 20 message:

Cách ack	Sau khi consumer chết	Kết quả
Auto-ack	Queue trống: broker đã đẩy hết 100 message cho consumer ngay từ đầu	Mất 80 message
Ack sau khi xử lý, prefetch 10	Queue còn 81 message, 10 trong đó có cờ `redelivered`	Không mất message nào, 1 message xử lý hai lần

Message bị xử lý hai lần là message đã xử lý xong nhưng consumer chết trước khi kịp gửi ack. Không có cách nào tránh hoàn toàn trường hợp này, nên consumer phải idempotent (mục dưới).

Phía producer cũng có chuyện tương tự: gửi xong chưa chắc broker đã nhận. Publisher confirm cho producer biết broker đã nhận và ghi message an toàn. Muốn message còn sau khi broker khởi động lại thì cần gửi message persistent vào queue durable. Muốn còn khi một node chết thì dùng quorum queue: queue có replica trên nhiều node, giữ nhất quán bằng Raft, và theo tài liệu là lựa chọn mặc định khi cần queue có replica (Consensus).

SQS: visibility timeout

SQS không giữ connection với consumer. Consumer gọi API để nhận message, và lúc đó SQS ẩn message trong một khoảng gọi là visibility timeout (mặc định 30 giây, tối đa 12 giờ). Consumer xử lý xong thì gọi DeleteMessage. Hết visibility timeout mà message chưa bị xoá thì nó hiện lại và consumer khác nhận được.

Cách này không cần phát hiện consumer chết, nhưng mang theo một cái bẫy: xử lý lâu hơn visibility timeout thì message bị xử lý hai lần, dù không ai chết. Trên lab, visibility timeout 2 giây, xử lý mất 3 giây: consumer thứ hai nhận lại message ở giây thứ 2,0–2,1 trong lúc consumer đầu vẫn đang làm. Cách chống là đặt visibility timeout dài hơn thời gian xử lý lâu nhất, hoặc trong lúc xử lý định kỳ gọi ChangeMessageVisibility để gia hạn. Gia hạn mỗi giây thì trên lab không consumer nào nhận lại.

Tệ hơn, mỗi lần nhận message có một receipt handle mới, và lệnh xoá dùng receipt handle cũ thì theo tài liệu SQS vẫn trả về thành công nhưng message có thể không bị xoá. Consumer chậm tưởng mình đã xoá xong, trong khi message vẫn quay vòng. ElasticMQ, bản giả lập SQS mà lab dùng, thì trả lỗi ReceiptHandleIsInvalid, nên chạy thử trên máy dev không thấy được cái bẫy này.

Chỉnh thời gian xử lý và visibility timeout để thấy message bị xử lý mấy lần:

Xử lý mất 3 s

Visibility timeout 2 s

maxReceiveCount 3

Gia hạn visibility trong lúc xử lý

xử lý xong và xoá đượcxử lý xong nhưng message đã bị người khác nhận lạilỗi hoặc chết giữa chừng

Message được xử lý thành công 3 lần mà vẫn bị chuyển sang dead letter queue sau 6,0 s: lần nào cũng xử lý lâu hơn visibility timeout, message hiện lại trước khi xoá, và lệnh xoá dùng receipt handle cũ nên không có tác dụng. Tăng visibility timeout, hoặc gia hạn visibility trong lúc xử lý.

0,0 s: consumer A nhận message (lần 1).
3,0 s: consumer A xử lý xong nhưng message đã hiện lại lúc 2,0 s và bị nhận lại; lệnh xoá dùng receipt handle cũ, SQS trả thành công nhưng message không bị xoá.
2,0 s: consumer B nhận message (lần 2).
5,0 s: consumer B xử lý xong nhưng message đã hiện lại lúc 4,0 s và bị nhận lại; lệnh xoá dùng receipt handle cũ, SQS trả thành công nhưng message không bị xoá.
4,0 s: consumer C nhận message (lần 3).
7,0 s: consumer C xử lý xong nhưng message đã hiện lại lúc 6,0 s và bị nhận lại; lệnh xoá dùng receipt handle cũ, SQS trả thành công nhưng message không bị xoá.
6,0 s: message hiện lại, đã bị nhận 3 lần = maxReceiveCount, chuyển sang dead letter queue.

Giả định: luôn có consumer rảnh hỏi liên tục, nên message hiện lại là bị nhận ngay; consumer chết giữa chừng ở nửa thời gian xử lý. Lab trên ElasticMQ: visibility 2 s, xử lý 3 s, consumer B nhận lại lúc 2,0–2,1 s.

RabbitMQ cũng có giới hạn tương tự cho consumer giữ message quá lâu mà không ack: consumer timeout, mặc định 30 phút. Quá hạn thì broker đóng channel và message được giao lại. Từ RabbitMQ 4.3, giới hạn này chỉ áp dụng cho quorum queue.

Prefetch: chia việc theo sức của consumer

RabbitMQ chủ động đẩy message cho consumer. Prefetch là số message tối đa một consumer giữ mà chưa ack. Không đặt prefetch thì broker đẩy message chia đều cho các consumer ngay khi chúng tới, bất kể consumer nào nhanh hay chậm.

Trên lab, một consumer nhanh (5 ms mỗi message) và một consumer chậm (50 ms), 200 message:

Không giới hạn prefetch: mỗi consumer nhận 100, consumer nhanh làm xong sớm rồi ngồi không, cả việc mất 5,1 giây.
Prefetch 1: consumer chỉ nhận message mới khi đã ack message trước, nên consumer nhanh tự nhận nhiều hơn (176–178 message), cả việc xong sau 1,1–1,3 giây.

Prefetch nhỏ thì chia việc công bằng, nhưng mỗi message phải chờ ack đi tới broker rồi mới được gửi, nên tốn một vòng mạng mỗi message. Tài liệu RabbitMQ cho rằng prefetch trong khoảng 100–300 thường cho throughput tốt nhất mà không làm consumer quá tải. Với việc nặng và thời gian xử lý dao động nhiều (tạo báo cáo, xử lý video) thì prefetch 1 hợp hơn.

Consumer nhanh: 5 ms mỗi message

Consumer chậm: 50 ms mỗi message

Prefetch 1

0,9 sxử lý hết 200 message

181 / 19message của consumer nhanh / chậm

1,1 s–1,3 slab đo được (consumer nhanh làm 176–178)

Prefetch 1: consumer chỉ nhận thêm khi đã ack, nên consumer nhanh tự nhận nhiều hơn (181 so với 19), xong sau 0,9 s, gần mức chia theo sức (0,9 s). Đổi lại, mỗi message phải chờ một vòng ack tới broker mới được gửi, nên prefetch quá nhỏ làm giảm throughput khi mạng xa.

Mô hình bỏ qua thời gian mạng giữa broker và consumer. Lab chạy với 5 ms và 50 ms; chỉ hai mức prefetch 1 và không giới hạn có số đo.

SQS là kiểu kéo: consumer chỉ nhận message khi gọi ReceiveMessage (tối đa 10 message mỗi lần), nên consumer rảnh tự nhận nhiều hơn, không cần cấu hình gì.

Retry, message độc và dead letter queue

Xử lý lỗi thì message được trả lại để thử lại. Nhưng có những message lỗi mãi: dữ liệu sai định dạng, tham chiếu bản ghi đã bị xoá, hay gây ra bug. Gọi là message độc (poison message). Trả nó về queue vô hạn lần thì nó quay vòng mãi. Trên lab, một message độc trong classic queue được giao lại 5.837–5.985 lần chỉ trong 2 giây, tốn CPU của consumer và có thể chặn các message phía sau.

Cách xử lý là giới hạn số lần giao rồi chuyển message sang dead letter queue (DLQ) để người xem xét:

RabbitMQ quorum queue đếm số lần giao. Từ RabbitMQ 4.0, giới hạn mặc định là 20, đặt bằng x-delivery-limit. Quá giới hạn thì message bị bỏ, hoặc chuyển sang dead letter exchange nếu có cấu hình. Trên lab, x-delivery-limit 3 thì message được giao 4 lần rồi vào DLQ; để mặc định thì giao 21 lần. Message cũng vào dead letter exchange khi bị reject hay nack với requeue = false, khi hết TTL, hay khi queue vượt giới hạn độ dài.
SQS dùng redrive policy: maxReceiveCount là số lần một message được nhận trước khi chuyển sang DLQ. Trên lab, maxReceiveCount 3 thì sau 3 lần nhận mà không xoá, message chuyển sang DLQ.

Vài điểm thực tế:

Phân biệt lỗi tạm thời và lỗi vĩnh viễn. Dịch vụ phía sau timeout thì nên thử lại. Dữ liệu sai thì thử lại bao nhiêu lần cũng vậy, nên đưa thẳng vào DLQ (RabbitMQ: nack với requeue = false).
Thử lại có chờ. Thử lại ngay lập tức khi dịch vụ phía sau đang quá tải chỉ làm nó tệ hơn (Backoff và jitter). RabbitMQ không có sẵn cơ chế chờ khi requeue. Cách hay dùng là gửi message vào một queue chờ có TTL, hết TTL thì message qua dead letter exchange quay về queue chính. Với SQS, có thể đặt visibility timeout của chính message đó bằng thời gian muốn chờ.
DLQ cần cảnh báo và cách đưa message về lại. DLQ không ai xem thì chỉ là chỗ dữ liệu biến mất lặng lẽ.

Cùng các ý này khi dùng database làm hàng đợi ở Retry, backoff và dead letter.

Consumer phải idempotent

Mọi hàng đợi ở trên đều chỉ đảm bảo at-least-once: message có thể được giao nhiều lần. Lab đã gặp đủ các đường dẫn tới chuyện đó: consumer chết sau khi xử lý mà trước khi ack, xử lý lâu hơn visibility timeout, lệnh xoá dùng receipt handle cũ. Tài liệu SQS cũng nói hàng đợi standard có thể giao nhiều hơn một bản của một message. Vì vậy phần xử lý phải cho cùng kết quả khi chạy hai lần: lưu id message đã xử lý trong cùng transaction với kết quả, dùng ràng buộc UNIQUE, hoặc viết thao tác sao cho tự nhiên idempotent (Idempotency).

Thứ tự

Một queue với một consumer xử lý tuần tự thì giữ đúng thứ tự gửi. Có nhiều consumer thì không: hai message liên tiếp được hai consumer xử lý song song, và cái sau có thể xong trước. Message bị requeue thì RabbitMQ đưa nó về vị trí cũ nếu được, nhưng lúc đó message phía sau có thể đã được xử lý rồi. Trên lab, 2 key mỗi key 50 sự kiện, hai consumer, 10% lần xử lý lỗi và requeue: 3–5 sự kiện xử lý xong sau một sự kiện mới hơn của cùng key.

Các cách giữ thứ tự:

Cách	Giữ thứ tự	Đổi lại
Single active consumer (RabbitMQ, `x-single-active-consumer`)	Toàn queue: chỉ một consumer nhận message, các consumer khác chờ làm dự phòng	Không xử lý song song. Trên lab, một consumer nhận cả 108 lần giao, consumer kia không nhận lần nào, và không sự kiện nào sai thứ tự
Chia queue theo key (mỗi nhóm key một queue, mỗi queue một consumer)	Theo key	Phải tự định tuyến, số queue cố định
SQS FIFO, `MessageGroupId`	Theo group	Giới hạn throughput (dưới đây)

SQS FIFO giữ thứ tự theo message group: các message cùng MessageGroupId được giao đúng thứ tự, và theo tài liệu, khi một message của group đang được xử lý (chưa xoá, chưa hết visibility timeout) thì không nhận thêm được message nào khác của group đó. Các group khác vẫn chạy song song. Trên lab, giữ A0 chưa xoá thì không nhận được A1, A2, A3, trong khi toàn bộ group B và C vẫn được nhận; xoá A0 xong mới nhận tiếp A1, A2, A3. Như vậy số group quyết định mức xử lý song song, và một message chậm chặn cả group của nó.

FIFO còn chống trùng khi gửi: gửi lại với cùng MessageDeduplicationId trong vòng 5 phút thì SQS không tạo message mới. Trên lab, gửi B0 hai lần thì group B chỉ có 4 message. Đổi lại là throughput: theo tài liệu, mỗi partition của hàng đợi FIFO giới hạn 300 lệnh mỗi giây cho mỗi loại API, hoặc 3.000 message mỗi giây khi gửi theo lô 10 (chế độ throughput cao cho phép nhiều hơn). Hàng đợi standard gần như không giới hạn nhưng chỉ giữ thứ tự gần đúng.

Khi consumer chậm hơn producer

Hàng đợi sinh ra để chịu đỉnh: producer gửi nhanh hơn consumer xử lý trong một lúc thì message xếp hàng chờ. Nhưng nếu kéo dài thì hàng đợi dài ra mãi. Cần theo dõi hai chỉ số: số message đang chờ và tuổi của message cũ nhất. Con số thứ hai cho biết người dùng đang phải chờ bao lâu, nên hợp để đặt cảnh báo hơn.

Khi hàng đợi dài ra:

Thêm consumer. Hàng đợi chia việc tự do cho bao nhiêu consumer cũng được, khác với Kafka bị giới hạn bởi số partition. Có thể tự động thêm bớt consumer theo độ dài hàng đợi (ví dụ KEDA trên Kubernetes).
Giới hạn độ dài hàng đợi để không hết bộ nhớ hay đĩa của broker. RabbitMQ cho đặt độ dài tối đa của queue, quá thì bỏ message cũ nhất hoặc từ chối nhận message mới. Từ chối nhận là cách báo ngược lại cho producer biết hệ thống đang quá tải.
Prefetch là back-pressure phía consumer: consumer không bị đẩy quá số message nó giữ được.

Với SQS, gọi ReceiveMessage trên hàng đợi rỗng vẫn tính tiền mỗi lần gọi. Long polling (WaitTimeSeconds, tối đa 20 giây) giữ request chờ tới khi có message hoặc hết thời gian. Trên lab, hàng đợi rỗng trong 6 giây: short polling gọi 4.993–5.118 lần và lần nào cũng nhận về rỗng, long polling 5 giây chỉ gọi 2 lần.

RabbitMQ, SQS hay Kafka

	RabbitMQ	SQS	Kafka
Vận hành	Tự chạy hoặc dịch vụ managed	AWS lo toàn bộ, tự mở rộng	Tự chạy hoặc dịch vụ managed, phức tạp nhất
Giao message	Broker đẩy, prefetch	Consumer kéo, visibility timeout	Consumer kéo theo offset
Định tuyến	Linh hoạt (exchange)	Đơn giản, fan-out qua SNS	Theo topic và partition
Sau khi xử lý	Xoá	Xoá	Giữ theo retention, đọc lại được
Thứ tự	Một consumer hoặc single active consumer	FIFO theo message group	Theo partition
Giới hạn đáng nhớ	Bộ nhớ và đĩa của broker	Message tối đa 1 MiB, giữ tối đa 14 ngày (mặc định 4 ngày)	Song song không vượt số partition

Chia việc cho worker (gửi email, tạo ảnh, xử lý đơn) thì dùng hàng đợi. Luồng sự kiện cho nhiều hệ thống đọc, cần đọc lại hay throughput rất lớn thì dùng Kafka. Đang chạy trên AWS và không muốn vận hành broker thì SQS. Cần định tuyến phức tạp, độ trễ thấp, ưu tiên message thì RabbitMQ. RabbitMQ cũng có Stream: log chỉ ghi nối, đọc lại được nhiều lần như Kafka, khi cần cả hai kiểu trong một hệ thống.

Ghi nhớ

Queue: mỗi message đến một consumer, để chia việc. Pub/sub: mỗi bên một queue, mỗi message đến mọi bên.
Message chỉ xong khi consumer ack (RabbitMQ) hay xoá (SQS). Auto-ack mất message khi consumer chết; ack sau khi xử lý thì có thể xử lý lại.
SQS: xử lý lâu hơn visibility timeout thì message bị xử lý hai lần; gia hạn visibility trong lúc xử lý. Xoá bằng receipt handle cũ thì không báo lỗi mà message vẫn còn.
Prefetch nhỏ chia việc theo sức consumer; không giới hạn thì consumer chậm thành điểm nghẽn.
Giới hạn số lần giao rồi đưa message độc vào dead letter queue (quorum queue mặc định 20 lần từ RabbitMQ 4.0, SQS dùng maxReceiveCount); phân biệt lỗi tạm thời và vĩnh viễn; có cảnh báo cho DLQ.
Mọi hàng đợi chỉ at-least-once nên consumer phải idempotent. Nhiều consumer thì mất thứ tự; giữ thứ tự bằng single active consumer, chia queue theo key, hoặc SQS FIFO theo message group.

Nguồn tham khảo

Đã đối chiếu (đọc nguồn ngày 2026-10-28):

Nguồn	Dùng cho phần
RabbitMQ docs, Consumer Acknowledgements and Publisher Confirms	Auto-ack coi message giao xong ngay khi gửi ("fire-and-forget"); tự requeue khi mất connection; message requeue về vị trí cũ nếu được; prefetch 100–300 thường cho throughput tốt nhất; publisher confirm và consumer ack độc lập nhau
RabbitMQ docs, Consumers	Consumer timeout mặc định 30 phút; từ 4.3 chỉ quorum queue hỗ trợ; single active consumer
RabbitMQ docs, Quorum Queues	Queue có replica dựa trên Raft, lựa chọn mặc định khi cần replica; delivery limit mặc định 20 từ 4.0, tắt bằng `x-delivery-limit = -1`
RabbitMQ docs, Dead Letter Exchanges	Bốn trường hợp message vào dead letter exchange
RabbitMQ docs, Streams	Log chỉ ghi nối, đọc lại nhiều lần
AWS docs, SQS: Visibility timeout, Dead-letter queues, FIFO delivery logic, Exactly-once processing, Message quotas, Short and long polling, Standard queues, DeleteMessage	Visibility timeout mặc định 30 s, tối đa 12 giờ; `maxReceiveCount`; không nhận thêm message cùng group khi một message đang xử lý; chống trùng trong 5 phút; FIFO 300 lệnh/s mỗi partition, 3.000 message/s theo lô; giữ mặc định 4 ngày, tối đa 14 ngày; message tối đa 1 MiB; long polling tối đa 20 s; standard queue at-least-once, có thể đảo thứ tự; xoá bằng receipt handle cũ thành công nhưng có thể không xoá

Kiểm bằng lab: fan-out và hai consumer cùng queue, auto-ack và ack sau khi xử lý khi consumer chết, prefetch, message độc trên classic và quorum queue, thứ tự với hai consumer và single active consumer (RabbitMQ 4.1.8); visibility timeout và gia hạn, redrive sang DLQ, FIFO theo message group và chống trùng, short và long polling (ElasticMQ 1.6.14). Từ kiến thức chung, chưa đối chiếu lần này: các loại exchange; fan-out bằng SNS; queue chờ có TTL để retry có độ trễ; độ dài tối đa của queue và cách xử lý khi tràn; KEDA; chế độ throughput cao của FIFO.

Số đo chi tiết

Kết quả chạy lab/hang-doi (RabbitMQ 4.1.8 và ElasticMQ 1.6.14 trong Docker, amqplib 2.2, @aws-sdk/client-sqs 3, Node 24.14, hai lượt)

RabbitMQ:

Phép thử	Kết quả
Consumer 5 ms và 50 ms, 200 message, không giới hạn prefetch	5,1 s; 100 và 100
Như trên, prefetch 1	1,1–1,3 s; 176–178 và 22–24
100 message, kill -9 sau 20, auto-ack	Queue còn 0; mất 80
Như trên, ack sau khi xử lý, prefetch 10	Queue còn 81 (10 có cờ `redelivered`); mất 0, xử lý hai lần 1
Fanout tới hai queue / hai consumer cùng queue, 100 message	100 và 100 / 52 và 48
Message độc, nack và requeue, 2 giây	Classic: 5.837–5.985 lần giao; quorum `x-delivery-limit` 3: 4 lần rồi vào DLQ; quorum mặc định: 21 lần rồi vào DLQ
2 key × 50 sự kiện, 2 consumer, prefetch 1, 10% requeue	Hai consumer: 3–5 sự kiện xong sau sự kiện mới hơn cùng key; single active consumer: 0, một consumer nhận cả 108 lần giao

Lượt chạy đầu đăng ký consumer nhanh trước khi consumer chậm kịp đăng ký, nên với prefetch không giới hạn consumer nhanh nhận hết 200 message; script sau đó đăng ký cả hai consumer rồi mới gửi.

SQS trên ElasticMQ:

Phép thử	Kết quả
Visibility timeout 2 s, xử lý 3 s, không gia hạn	Consumer B nhận lại lúc 2,0–2,1 s (`ApproximateReceiveCount` 2); lệnh xoá của A lỗi `ReceiptHandleIsInvalid` (SQS thật: thành công nhưng có thể không xoá)
Như trên, gia hạn mỗi giây	B không nhận được trong 4 s
Redrive `maxReceiveCount` 3, visibility 1 s	3 lần nhận, vào DLQ sau 3,2 s
FIFO 3 group × 4 message, giữ A0	Nhận B0–B3, C0–C3, không nhận được A1–A3; xoá A0 thì nhận A1, A2, A3; gửi B0 hai lần cùng deduplication id: group B có 4 message
Hàng đợi rỗng, 6 giây	Short polling 4.993–5.118 lần gọi; long polling 5 s: 2 lần

Code đã dùng để chạy thử

Dựng RabbitMQ và ElasticMQ bằng npm run lab -- hang-doi:up, chạy bằng npm run lab -- hang-doi (khoảng 1 phút), xoá bằng npm run lab -- hang-doi:down. Không cần lab:up. Xem Lab.

lab/hang-doi/docker-compose.yml

yml

# Hàng đợi message cho trang messaging/hang-doi.
#   docker compose -f hang-doi/docker-compose.yml up -d --wait     (npm run hang-doi:up)
#   docker compose -f hang-doi/docker-compose.yml down -v          (npm run hang-doi:down)
# lab-rabbitmq: RabbitMQ 4.1, AMQP ở 127.0.0.1:56720, giao diện quản trị ở http://127.0.0.1:15673 (guest/guest)
# lab-elasticmq: ElasticMQ, giả lập API của Amazon SQS, ở http://127.0.0.1:39324
services:
  rabbitmq:
    image: rabbitmq:4.1-management
    container_name: lab-rabbitmq
    ports: ['127.0.0.1:56720:5672', '127.0.0.1:15673:15672']
    # guest chỉ đăng nhập được từ localhost; cổng chỉ mở trên 127.0.0.1 nên cho phép từ ngoài container
    environment:
      RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS: '-rabbit loopback_users []'
    healthcheck:
      test: ['CMD', 'rabbitmq-diagnostics', '-q', 'check_port_connectivity']
      interval: 3s
      timeout: 10s
      retries: 40
    networks: [hang-doi]
  elasticmq:
    image: softwaremill/elasticmq-native:1.6.14
    container_name: lab-elasticmq
    ports: ['127.0.0.1:39324:9324']
    networks: [hang-doi]
networks:
  hang-doi:
    name: lab-hang-doi

lab/hang-doi/rabbit.mjs và rabbit-child.mjs: RabbitMQ

// RabbitMQ 4.1 cho trang messaging/hang-doi.
//   1. Prefetch: 2 consumer, một nhanh (5 ms/message) một chậm (50 ms/message), 200 message; không giới hạn prefetch
//      so với prefetch 1
//   2. Consumer bị kill -9 sau khi xử lý 20/100 message: auto-ack so với ack sau khi xử lý
//   3. Fan-out: một exchange fanout gửi tới hai queue, so với hai consumer cùng một queue
//   4. Message độc: luôn lỗi, nack và requeue; classic queue (không giới hạn) so với quorum queue có giới hạn số lần giao
//   5. Thứ tự theo key với hai consumer, so với single active consumer
import { fork } from 'node:child_process'
import { rabbit, sleep, now, sec } from './common.mjs'

const conn = await rabbit()
const ch = await conn.createChannel()
const fresh = async (q, args = {}) => { await ch.deleteQueue(q).catch(() => {}); await ch.assertQueue(q, { durable: true, arguments: args }) }
const fill = (q, n, f = (i) => String(i)) => { for (let i = 0; i < n; i++) ch.sendToQueue(q, Buffer.from(f(i)), { persistent: true }) }
const children = new Set()

try {
  console.log('\n=== 1. 200 message, consumer nhanh 5 ms và consumer chậm 50 ms mỗi message ===')
  for (const prefetch of [0, 1]) {
    const q = 'lab-work'
    await fresh(q)
    const counts = { nhanh: 0, chậm: 0 }
    const chans = []
    // đăng ký cả hai consumer trước rồi mới gửi message: nếu message có sẵn, consumer đăng ký trước nhận hết
    for (const [name, ms] of [['nhanh', 5], ['chậm', 50]]) {
      const c = await conn.createChannel()
      chans.push(c)
      if (prefetch) await c.prefetch(prefetch)
      let chain = Promise.resolve()
      await c.consume(q, (m) => { chain = chain.then(async () => { await sleep(ms); counts[name]++; c.ack(m) }) })
    }
    const t0 = now()
    fill(q, 200)
    while (counts.nhanh + counts.chậm < 200) await sleep(10)
    console.log(`${prefetch ? `prefetch ${prefetch}` : 'không giới hạn prefetch'}: xong sau ${sec(now() - t0)}; consumer nhanh xử lý ${counts.nhanh}, consumer chậm ${counts.chậm}`)
    for (const c of chans) await c.close()
  }

  console.log('\n=== 2. 100 message, consumer bị kill -9 sau khi xử lý 20 message ===')
  for (const [mode, label] of [['auto', 'Auto-ack (noAck)'], ['manual', 'Ack sau khi xử lý, prefetch 10']]) {
    const q = 'lab-crash'
    await fresh(q)
    fill(q, 100)
    await sleep(300)
    const seen = []
    const child = fork(new URL('./rabbit-child.mjs', import.meta.url), [q, mode])
    children.add(child)
    await new Promise((resolve) => {
      child.on('message', (m) => { seen.push(m); if (seen.length === 20) child.kill('SIGKILL') })
      child.on('exit', resolve)
    })
    await sleep(500)
    const { messageCount } = await ch.checkQueue(q)
    // consumer mới đọc phần còn lại
    const second = []
    const c2 = await conn.createChannel()
    await c2.consume(q, (m) => { second.push({ id: Number(m.content.toString()), redelivered: m.fields.redelivered }); c2.ack(m) })
    let last = second.length, idle = 0
    while (idle < 10) { await sleep(100); if (second.length === last) idle++; else { idle = 0; last = second.length } }
    await c2.close()
    const count = new Map()
    for (const m of [...seen, ...second]) count.set(m.id, (count.get(m.id) ?? 0) + 1)
    const lost = Array.from({ length: 100 }, (_, i) => i).filter((i) => !count.has(i))
    const dup = [...count].filter(([, c]) => c > 1).map(([i]) => i)
    console.log(`${label.padEnd(32)} còn ${messageCount} message trong queue sau khi consumer chết; consumer mới nhận ${second.length} (trong đó ${second.filter((m) => m.redelivered).length} có cờ redelivered); mất ${lost.length}, xử lý hai lần ${dup.length}`)
  }

  console.log('\n=== 3. Fan-out so với hai consumer cùng một queue, 100 message ===')
  {
    await ch.assertExchange('lab-fan', 'fanout', { durable: false })
    for (const q of ['lab-fan-email', 'lab-fan-stats']) { await fresh(q); await ch.bindQueue(q, 'lab-fan', '') }
    for (let i = 0; i < 100; i++) ch.publish('lab-fan', '', Buffer.from(String(i)))
    await fresh('lab-shared')
    fill('lab-shared', 100)
    await sleep(500)
    const n = async (q) => (await ch.checkQueue(q)).messageCount
    const got = [0, 0]
    const cs = []
    for (const k of [0, 1]) { const c = await conn.createChannel(); await c.prefetch(1); cs.push(c); await c.consume('lab-shared', (m) => { got[k]++; c.ack(m) }) }
    await sleep(1500)
    for (const c of cs) await c.close()
    console.log(`Exchange fanout: queue email có ${await n('lab-fan-email')}, queue thống kê có ${await n('lab-fan-stats')}, mỗi queue nhận đủ. Hai consumer cùng một queue: ${got[0]} và ${got[1]}, mỗi message chỉ một consumer nhận`)
    await ch.deleteExchange('lab-fan')
  }

  console.log('\n=== 4. Message độc: consumer luôn lỗi, nack và requeue ===')
  for (const [label, args] of [
    ['Classic queue', { 'x-queue-type': 'classic' }],
    ['Quorum queue, x-delivery-limit 3', { 'x-queue-type': 'quorum', 'x-delivery-limit': 3, 'x-dead-letter-exchange': '', 'x-dead-letter-routing-key': 'lab-dlq' }],
    ['Quorum queue, giới hạn mặc định', { 'x-queue-type': 'quorum', 'x-dead-letter-exchange': '', 'x-dead-letter-routing-key': 'lab-dlq' }]
  ]) {
    const q = 'lab-poison'
    await fresh('lab-dlq')
    await fresh(q, args)
    ch.sendToQueue(q, Buffer.from('message độc'), { persistent: true })
    await sleep(300)
    let deliveries = 0
    const c = await conn.createChannel()
    await c.prefetch(1)
    await c.consume(q, (m) => { deliveries++; c.nack(m, false, true) })
    await sleep(2000)
    await c.close()
    const dlq = (await ch.checkQueue('lab-dlq')).messageCount
    const left = (await ch.checkQueue(q)).messageCount
    console.log(`${label.padEnd(34)} trong 2 giây: giao ${deliveries.toLocaleString('vi-VN')} lần; còn ${left} trong queue, ${dlq} trong dead letter queue`)
  }

  console.log('\n=== 5. 2 key × 50 sự kiện theo thứ tự, gửi xen kẽ; 2 consumer, xử lý 1–10 ms, 10% lần xử lý lỗi và requeue ===')
  for (const [label, args] of [['Hai consumer cùng queue', {}], ['Single active consumer', { 'x-single-active-consumer': true }]]) {
    const q = 'lab-order'
    await fresh(q, args)
    let seq = 0
    for (let s = 0; s < 50; s++) for (let k = 0; k < 2; k++) ch.sendToQueue(q, Buffer.from(JSON.stringify({ key: k, s, seq: seq++ })), { persistent: true })
    await sleep(300)
    const done = []
    const busy = [0, 0]
    const cs = []
    for (const id of [0, 1]) {
      const c = await conn.createChannel(); cs.push(c); await c.prefetch(1)
      await c.consume(q, async (m) => {
        busy[id]++
        await sleep(1 + Math.random() * 9)
        if (Math.random() < 0.1) { c.nack(m, false, true); return }
        done.push(JSON.parse(m.content.toString())); c.ack(m)
      })
    }
    while (done.length < 100) await sleep(20)
    for (const c of cs) await c.close()
    let wrong = 0
    const last = {}
    for (const d of done) { if (last[d.key] !== undefined && d.s < last[d.key]) wrong++; last[d.key] = Math.max(last[d.key] ?? -1, d.s) }
    console.log(`${label.padEnd(26)} ${wrong} sự kiện xử lý xong sau một sự kiện mới hơn của cùng key; số lần nhận của từng consumer: ${busy.join(' và ')}`)
  }
} finally {
  for (const c of children) c.kill('SIGKILL')
  for (const q of ['lab-work', 'lab-crash', 'lab-fan-email', 'lab-fan-stats', 'lab-shared', 'lab-poison', 'lab-dlq', 'lab-order']) await ch.deleteQueue(q).catch(() => {})
  await conn.close()
}

// Consumer RabbitMQ chạy trong process riêng, để lab/hang-doi/rabbit.mjs kill -9 nó giữa chừng.
// Đối số: tên queue, cách ack ('auto': noAck, broker coi là đã giao xong ngay khi gửi; 'manual': ack sau khi xử lý, prefetch 10).
// Mỗi message xử lý 20 ms rồi báo { id, redelivered } cho process cha.
import { rabbit, sleep } from './common.mjs'

const [queue, mode] = process.argv.slice(2)
const conn = await rabbit()
const ch = await conn.createChannel()
if (mode === 'manual') await ch.prefetch(10)
// xử lý tuần tự từng message như một worker thật
let chain = Promise.resolve()
await ch.consume(queue, (m) => {
  chain = chain.then(async () => {
    await sleep(20)
    process.send({ id: Number(m.content.toString()), redelivered: m.fields.redelivered })
    if (mode === 'manual') ch.ack(m)
  })
}, { noAck: mode === 'auto' })
process.on('message', async (m) => { if (m === 'stop') { await chain; await conn.close(); process.exit(0) } })

lab/hang-doi/sqs.mjs: API của SQS trên ElasticMQ

// API của Amazon SQS trên ElasticMQ (bản giả lập chạy trong Docker), cho trang messaging/hang-doi.
//   1. Visibility timeout 2 s, xử lý mất 3 s: consumer khác nhận lại message trong lúc consumer đầu chưa xong;
//      gia hạn visibility mỗi giây thì không
//   2. Message độc với redrive policy maxReceiveCount 3: nhận bao nhiêu lần thì chuyển sang dead letter queue
//   3. FIFO queue: message cùng group theo đúng thứ tự, giữ một message chưa xoá thì cả group bị chặn, group khác vẫn đi;
//      gửi trùng MessageDeduplicationId thì chỉ có một message
//   4. Short polling so với long polling: số lần gọi ReceiveMessage trả về rỗng trong 6 giây
// ElasticMQ là bản giả lập: hành vi lấy từ tài liệu SQS, chi tiết thời gian và giới hạn có thể khác SQS thật.
import {
  CreateQueueCommand, DeleteQueueCommand, SendMessageCommand, ReceiveMessageCommand, DeleteMessageCommand,
  ChangeMessageVisibilityCommand, GetQueueAttributesCommand
} from '@aws-sdk/client-sqs'
import { sqs, sleep, now, sec } from './common.mjs'

const s = sqs()
const created = []
const create = async (name, attrs = {}) => {
  const { QueueUrl } = await s.send(new CreateQueueCommand({ QueueName: name, Attributes: attrs }))
  created.push(QueueUrl)
  return QueueUrl
}
const receive = (QueueUrl, extra = {}) => s.send(new ReceiveMessageCommand({ QueueUrl, MaxNumberOfMessages: 10, MessageSystemAttributeNames: ['ApproximateReceiveCount'], ...extra })).then((r) => r.Messages ?? [])
const suffix = Date.now()

try {
  console.log('\n=== 1. Visibility timeout 2 s, mỗi message xử lý mất 3 s ===')
  for (const extend of [false, true]) {
    const q = await create(`lab-vis-${extend ? 'ext' : 'plain'}-${suffix}`, { VisibilityTimeout: '2' })
    await s.send(new SendMessageCommand({ QueueUrl: q, MessageBody: 'đơn 42' }))
    const t0 = now()
    const [m] = await receive(q)
    const log = [`consumer A nhận lúc 0 s`]
    // consumer A xử lý 3 s; nếu gia hạn thì mỗi giây đặt lại visibility 2 s
    const work = (async () => {
      for (let i = 0; i < 3; i++) {
        await sleep(1000)
        if (extend) await s.send(new ChangeMessageVisibilityCommand({ QueueUrl: q, ReceiptHandle: m.ReceiptHandle, VisibilityTimeout: 2 }))
      }
      await s.send(new DeleteMessageCommand({ QueueUrl: q, ReceiptHandle: m.ReceiptHandle })).catch((e) => log.push(`A xoá lỗi: ${e.name}`))
      log.push(`A xử lý xong và xoá lúc ${sec(now() - t0)}`)
    })()
    // consumer B hỏi liên tục
    let gotB = null
    while (now() - t0 < 4000 && !gotB) { const r = await receive(q); if (r.length) gotB = { at: now() - t0, count: r[0].Attributes.ApproximateReceiveCount }; else await sleep(100) }
    await work
    if (gotB) log.push(`consumer B cũng nhận được lúc ${sec(gotB.at)} (ApproximateReceiveCount = ${gotB.count})`)
    else log.push('consumer B không nhận được trong 4 s')
    console.log(`${extend ? 'Gia hạn visibility mỗi giây' : 'Không gia hạn'}: ${log.join('; ')}`)
  }

  console.log('\n=== 2. Message độc, redrive policy maxReceiveCount 3, visibility timeout 1 s ===')
  {
    const dlq = await create(`lab-dlq-${suffix}`)
    const { Attributes } = await s.send(new GetQueueAttributesCommand({ QueueUrl: dlq, AttributeNames: ['QueueArn'] }))
    const q = await create(`lab-main-${suffix}`, { VisibilityTimeout: '1', RedrivePolicy: JSON.stringify({ deadLetterTargetArn: Attributes.QueueArn, maxReceiveCount: '3' }) })
    await s.send(new SendMessageCommand({ QueueUrl: q, MessageBody: 'message độc' }))
    let receives = 0
    const t0 = now()
    let inDlq = []
    while (now() - t0 < 15000) {
      receives += (await receive(q)).length          // nhận rồi không xoá, như consumer luôn lỗi
      inDlq = await receive(dlq, { VisibilityTimeout: 0 })
      if (inDlq.length) break
      await sleep(200)
    }
    console.log(`Consumer nhận message ${receives} lần (không xoá); message nằm trong dead letter queue sau ${sec(now() - t0)}: ${inDlq.length ? 'có' : 'không'}`)
  }

  console.log('\n=== 3. FIFO queue: 3 group × 4 message ===')
  {
    const q = await create(`lab-${suffix}.fifo`, { FifoQueue: 'true', VisibilityTimeout: '3' })
    for (let i = 0; i < 4; i++) for (const g of ['A', 'B', 'C']) await s.send(new SendMessageCommand({ QueueUrl: q, MessageBody: `${g}${i}`, MessageGroupId: g, MessageDeduplicationId: `${g}${i}` }))
    // gửi lại B0 với cùng deduplication id: không thành message mới
    await s.send(new SendMessageCommand({ QueueUrl: q, MessageBody: 'B0 (gửi lại)', MessageGroupId: 'B', MessageDeduplicationId: 'B0' }))
    const order = []
    // lần nhận đầu: lấy A0 và giữ nó (không xoá), chỉ xoá các message khác
    const first = await receive(q, { MaxNumberOfMessages: 1 })
    const held = first[0]
    order.push(`${held.Body} (giữ, chưa xoá)`)
    const t0 = now()
    let blockedA = true
    while (now() - t0 < 2500) {
      const r = await receive(q)
      for (const m of r) { order.push(m.Body); if (m.Body.startsWith('A')) blockedA = false; await s.send(new DeleteMessageCommand({ QueueUrl: q, ReceiptHandle: m.ReceiptHandle })) }
      if (!r.length) await sleep(100)
    }
    console.log(`Thứ tự nhận trong 2,5 s đầu: ${order.join(', ')}`)
    console.log(`Trong lúc A0 chưa xoá, nhận được message nào khác của group A không: ${blockedA ? 'không' : 'có'}; tổng số message của group B nhận được: ${order.filter((x) => x.startsWith('B')).length} (gửi 5 lần, một lần trùng deduplication id)`)
    await s.send(new DeleteMessageCommand({ QueueUrl: q, ReceiptHandle: held.ReceiptHandle }))
    const rest = []
    for (let i = 0; i < 10; i++) { const r = await receive(q); for (const m of r) { rest.push(m.Body); await s.send(new DeleteMessageCommand({ QueueUrl: q, ReceiptHandle: m.ReceiptHandle })) } if (!r.length && rest.length >= 3) break; await sleep(100) }
    console.log(`Sau khi xoá A0: nhận tiếp ${rest.join(', ')}`)
  }

  console.log('\n=== 4. Queue rỗng, gọi ReceiveMessage liên tục trong 6 giây ===')
  {
    const q = await create(`lab-poll-${suffix}`)
    for (const wait of [0, 5]) {
      let calls = 0
      const t0 = now()
      while (now() - t0 < 6000) { await receive(q, { WaitTimeSeconds: wait }); calls++ }
      console.log(`WaitTimeSeconds = ${wait}${wait ? ' (long polling)' : ' (short polling)'}: ${calls.toLocaleString('vi-VN')} lần gọi, đều trả về rỗng`)
    }
  }
} finally {
  for (const QueueUrl of created) await s.send(new DeleteQueueCommand({ QueueUrl })).catch(() => {})
}

lab/hang-doi/common.mjs

// Hàm dùng chung cho lab/hang-doi: RabbitMQ (amqplib) và ElasticMQ (giả lập SQS, @aws-sdk/client-sqs).
import amqp from 'amqplib'
import { SQSClient } from '@aws-sdk/client-sqs'

export const RABBIT_URL = 'amqp://guest:guest@127.0.0.1:56720'
export const rabbit = () => amqp.connect(RABBIT_URL)
export const sqs = () => new SQSClient({ endpoint: 'http://127.0.0.1:39324', region: 'elasticmq', credentials: { accessKeyId: 'x', secretAccessKey: 'x' } })
export const sleep = (ms) => new Promise((r) => setTimeout(r, ms))
export const now = () => performance.now()
export const sec = (v) => (v / 1000).toFixed(1).replace('.', ',') + ' s'

Hàng đợi message ​

Queue và pub/sub ​

Ack: khi nào message được coi là xong ​

RabbitMQ: ack ​

SQS: visibility timeout ​

Prefetch: chia việc theo sức của consumer ​

Retry, message độc và dead letter queue ​

Consumer phải idempotent ​

Thứ tự ​

Khi consumer chậm hơn producer ​

RabbitMQ, SQS hay Kafka ​

Nguồn tham khảo ​

Số đo chi tiết ​

Code đã dùng để chạy thử ​