Phần trước dựng bản đồ năm khái niệm. Bài này đi vào hai lời gọi bạn sẽ viết nhiều nhất — basicPublish và basicConsume — và theo dõi một thông điệp qua từng trạng thái của nó.
basicPublish có năm tham số, và ai cũng bỏ qua hai cái
ch.basicPublish(exchange, routingKey, mandatory, props, body);
| Tham số | Ý nghĩa | Hay bị bỏ qua vì |
|---|---|---|
exchange |
Gửi vào exchange nào. "" là default exchange |
Ví dụ nào cũng để rỗng |
routingKey |
Khoá để binding so khớp | Với default exchange thì nó là tên hàng đợi |
mandatory |
Báo lại nếu không queue nào nhận | Mặc định false — im lặng |
props |
Siêu dữ liệu đi kèm | Truyền null cho nhanh |
body |
Mảng byte. RabbitMQ không quan tâm bên trong là gì | — |
Hai cái bị bỏ qua — mandatory và props — chính là hai cái cứu bạn lúc gỡ lỗi lúc 2 giờ sáng.
Mười thuộc tính đi kèm, và chúng tới nơi nguyên vẹn
var props = new AMQP.BasicProperties.Builder()
.messageId("don-001").correlationId("phien-42")
.contentType("application/json").contentEncoding("utf-8")
.deliveryMode(2).priority(0).timestamp(new Date())
.appId("cua-hang").type("DonMoi")
.headers(Map.of("nguon", "web", "lan-thu", 1))
.build();
Consumer nhận được đúng những gì đã gửi:
messageId=don-001 correlationId=phien-42 contentType=application/json
deliveryMode=2 appId=cua-hang type=DonMoi headers={nguon=web, lan-thu=1}
redelivered=false
deliveryMode=2 nghĩa là persistent. headers là chỗ để trace id đi theo thông điệp qua ranh giới broker — bài 31. correlationId là chỗ để nối yêu cầu với trả lời. Không cái nào RabbitMQ tự dùng, trừ deliveryMode, priority và expiration; số còn lại là để bạn đọc.
Vòng đời: Ready → Unacked → xong
Gửi 5 thông điệp rồi hỏi broker ở bốn thời điểm:
| Thời điểm | messages_ready |
messages_unacknowledged |
messages |
|---|---|---|---|
| Chưa có consumer | 5 | 0 | 5 |
| Consumer đã nhận, chưa ack | 0 | 5 | 5 |
| Consumer đứt kết nối | 5 | 0 | 5 |
| Sau khi ack đủ 5 | 0 | 0 | 0 |
messages là ready + unacked; nhìn nhầm cột ready rồi kết luận consumer đã xử lý xong là sai — chúng có thể đang bị giữ trong một consumer treo.
Dòng thứ ba là điều quan trọng nhất: consumer chết giữa chừng thì không mất thông điệp nào. Broker chờ ack, không thấy connection nữa thì trả hết về hàng đợi.
Cờ redelivered nói lên điều gì — và không nói lên điều gì
Sau khi 5 thông điệp quay lại và được consumer thứ hai xử lý:
sau khi ack đủ 5 : ready=0 | số cái có cờ redelivered=5/5
Cả năm đều mang cờ. Nghe thì tiện: cứ thấy redelivered=true là biết đây là lần giao thứ hai trở đi.
Nhưng đừng xây logic nghiệp vụ trên nó. Cờ này chỉ nói "thông điệp này có thể đã được giao trước đó", không nói đã giao mấy lần, cũng không đảm bảo lần trước consumer đã xử lý tới đâu. Muốn biết chắc thì phải làm consumer lũy đẳng — bài 22.
mandatory: bắt lại thông điệp không có đường đi
Phần trước cho thấy thông điệp gửi vào exchange không có binding khớp sẽ biến mất không dấu vết. Bật mandatory là có dấu vết:
mandatory=false : broker trả về 0 thông điệp (im lặng nuốt mất)
mandatory=true : broker trả về 1 thông điệp
code=312 text=NO_ROUTE routingKey=khong-ai-bind
Cách dùng: bật mandatory=true và gắn một ReturnListener.
ch.addReturnListener((code, text, ex, key, props, body) ->
log.error("Không có đường đi: {} {} routingKey={}", code, text, key));
ch.basicPublish("don-hang", "khong-ai-bind", true, null, body);
Không có listener mà bật mandatory thì thông điệp vẫn bị trả về rồi vẫn bị vứt — chỉ khác là bạn đã trả tiền cho một vòng mạng thừa. Bật cờ thì phải bắt.
basicConsume: bốn tham số
ch.basicConsume(queue, autoAck, consumerTag, deliverCallback, cancelCallback);
autoAck=true là ack ngay lúc broker giao, trước khi bạn xử lý xong — mất thông điệp nếu tiến trình chết giữa chừng, và bảng ở trên sẽ không bao giờ có cột unacked. Bài 16 đo phần thiệt hại.
consumerTag là tên bạn tự đặt cho consumer. Bỏ trống thì client tự sinh chuỗi ngẫu nhiên, và khi rabbitmqctl list_consumers hiện amq.ctag-4kPz… thì không ai đoán nổi đó là dịch vụ nào. Đặt tên đi.
cancelCallback chạy khi hàng đợi bị xoá dưới chân consumer. Phần lớn mã ví dụ để trống nó, và rồi consumer im lặng ngừng làm việc mà không ai biết.
Thử ba mươi giây
# gửi 3 thông điệp rồi xem hai cột đó tách nhau thế nào
docker exec -u rabbitmq rmq rabbitmqctl -q list_queues \
name messages_ready messages_unacknowledged messages
Chạy lệnh này trong lúc ứng dụng đang bận. Nếu messages_unacknowledged lớn và không tụt xuống, bạn có một consumer đang treo — và không lệnh nào khác chỉ ra điều đó rõ bằng nó.
Bài sau: ba đường nhìn vào broker — giao diện quản trị, rabbitmqctl và HTTP API — và lệnh kiểm tra nào thật sự nói lên rằng broker đã sẵn sàng.
