Consumer idempotent: nhận trùng message mà không ghi hai lần (Kafka, RabbitMQ) Trung cấp

Vấn đề

Consumer nhận OrderPaid, cộng 12.500 vào số dư merchant và commit database. Process chết trước khi báo broker (Kafka: commit offset; RabbitMQ: basic.ack). Broker giao lại cùng event, handler cộng thêm lần nữa, số dư thành 25.000.

Không message nào mất, không có lỗi SQL. Đây là at-least-once đúng như thiết kế: ghi database và báo broker là hai nơi lưu tiến độ, không có transaction chung. Cách xử lý là làm side effect idempotent.

Chọn cách chống trùng

Cách	Hoạt động	Khi nào hợp
Inbox table (bài này)	Insert (consumer, message_id) unique cùng transaction với side effect	Mặc định: mọi side effect trong một database
Idempotent tự nhiên	SET status = 'paid', upsert theo khóa nghiệp vụ	Thao tác “đặt giá trị”, không phải “cộng thêm”
Versioned update	UPDATE ... WHERE version = $expected	Event có thứ tự theo aggregate
Idempotency key ở downstream	Gửi message ID cho API ngoài (cổng thanh toán)	Side effect không nằm trong DB của bạn

Nếu thao tác tự nhiên đã idempotent thì không cần inbox. “Cộng 12.500” thì không.

Inbox table

Event ID phải ổn định qua mọi lần giao: producer sinh khi phát event (UUID hoặc ID từ outbox), consumer không tự sinh. Primary key trên (consumer_name, message_id) để mỗi consumer group có sổ riêng.

CREATE TABLE inbox (
  consumer_name text        NOT NULL,
  message_id    text        NOT NULL,
  processed_at  timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (consumer_name, message_id)
);

Một transaction: insert inbox và chỉ cập nhật nghiệp vụ nếu insert thực sự thêm dòng.

BEGIN;
WITH ins AS (
  INSERT INTO inbox (consumer_name, message_id)
  VALUES ('accounting', 'event-42')
  ON CONFLICT DO NOTHING
  RETURNING 1
)
UPDATE ledger SET balance = balance + 12500
 WHERE merchant_id = 1 AND EXISTS (SELECT 1 FROM ins);
COMMIT;

Đã chạy trên PostgreSQL 17 (database scratch): lần đầu UPDATE 1, lần giao lại cùng event-42 cho UPDATE 0, số dư cuối cùng 12500. Ứng dụng đọc số dòng bị ảnh hưởng (hoặc kiểm tra kết quả insert) để log “duplicate skipped”, rồi ack bình thường.

Unique constraint là thứ chặn race giữa hai process. SELECT kiểm tra rồi mới INSERT không đủ: hai worker cùng thấy “chưa có”. Insert marker và business write phải ở cùng transaction, vì tách ra thì rollback một nửa làm redelivery bỏ qua việc chưa làm, hoặc làm lại việc đã làm.

Giao lần hai gặp conflict trong inbox, ledger không đổi

Phóng to sơ đồ

Broker → Consumerevent-42

◀ Trước▶ PhátTiếp ▶↺ Từ đầuBước 1/4

Giao event-42 lần đầu (redelivered = false trên RabbitMQ; Kafka chưa commit offset).

Góc nhìn Kafka và RabbitMQ

	Kafka	RabbitMQ
Khoảng hở	DB commit xong, chưa commit offset	DB commit xong, chưa basic.ack (hoặc channel đóng)
Dấu hiệu giao lại	Không có cờ; thấy lại offset đã xử lý	Cờ redelivered = true trên delivery
Key dedupe	Event ID trong payload/header (không dùng offset vì replay hoặc đổi topic)	Message ID hoặc event ID trong payload
Xem thêm	Commit offset	Ack và prefetch

Cờ redelivered của RabbitMQ chỉ nói “message này từng được giao và requeue”. Dùng nó để tối ưu: khi false, có thể bỏ qua một lần kiểm tra tốn kém; khi true, bắt buộc kiểm tra inbox. Không dùng nó để thay inbox: một job được publish hai lần là hai message khác nhau, cả hai đều có redelivered = false.

Kafka exactly-once chỉ bao trùm luồng đọc Kafka rồi ghi về Kafka. Ghi PostgreSQL, gửi email hay gọi REST vẫn cần idempotency ở đích. Xem Kafka transactions.

Retention của bảng dedupe

Inbox lớn dần, và xóa quá sớm làm duplicate lọt qua. Quy tắc: giữ ít nhất bằng khoảng thời gian dài nhất mà một message có thể bị giao lại hoặc replay.

Nguồn trùng	Cửa sổ cần phủ
Crash, rebalance, redelivery	Vài phút tới vài giờ
Outbox relay publish lại	Thời gian relay có thể chết và phục hồi
Replay từ dead queue hoặc reset offset Kafka	Thời gian bạn cho phép replay, thường là retention của topic

Cách dọn:

-- đơn giản: xóa theo lô
DELETE FROM inbox WHERE processed_at < now() - interval '7 days';

Bảng lớn nên partition theo processed_at rồi DROP PARTITION, rẻ hơn DELETE nhiều và không để lại dead tuple. Con số 7 ngày ở trên là ví dụ, hãy chọn theo cửa sổ replay của bạn. Sau khi xóa, event cũ hơn cửa sổ nếu bị replay sẽ áp dụng lại, nên cần đối soát định kỳ.

Demo đã chạy

npx --prefix examples ts-node examples/src/idempotent-consumer/consumer-demo.ts

unsafeHandle gọi hai lần với cùng message cho số dư 25000. Sau reset, handleInOneTransaction trả applied lần đầu, duplicate lần sau, số dư giữ 12500. Demo dùng Map và Set trong RAM nên không có atomicity hay unique constraint thật; SQL ở trên mới là bản bảo vệ thật.

Để kiểm chứng thiết kế thật: hai process cùng xử lý một eventId và assert chỉ một áp dụng; kill process sau DB commit, trước ack, rồi kiểm tra số dư và inbox.

Ví dụ thực tế

Kế toán nhận OrderPaid

Service kế toán nhận khoảng 200 event/giây và cộng số dư chờ thanh toán của merchant.

• Producer sinh eventId (UUID) khi phát event qua outbox.
• Consumer ghi inbox và ledger trong một transaction; ledger lưu eventId làm reference đối soát.
• Rolling deploy gây rebalance: một số event đọc lại, log “duplicate skipped”, số dư không đổi.
• Inbox giữ 14 ngày (dài hơn retention topic), partition theo ngày.
• Đối soát hằng ngày vì idempotency không phát hiện event bị thiếu, hay hai event ID khác nhau cho cùng một nghiệp vụ.

Khi nào dùng / không dùng

• Dùng inbox: side effect dạng cộng, trừ, tạo bản ghi, và message có thể giao lại (hầu như luôn đúng).
• Không cần inbox: thao tác đã idempotent (upsert theo khóa nghiệp vụ, đặt trạng thái).
• Không đủ: side effect ngoài database (email, HTTP). Dùng idempotency key ở đích hoặc outbox cho bước đó.
• Chọn key: orderId quá thô nếu một order có payment và refund; dùng event ID.

🎮 Thử tay

• Scale Lab: notifications: retry làm gửi trùng, bật chống gửi trùng ở worker.

Checklist production

1. message_id do producer sinh, ổn định qua redelivery; không dùng offset hay UUID sinh ở consumer.
2. Primary key hoặc unique trên (consumer_name, message_id).
3. Marker và business write cùng transaction.
4. Ack (hoặc commit offset) sau khi transaction commit.
5. Duplicate được ack như thành công, kèm log và metric.
6. Retention inbox lớn hơn cửa sổ redelivery và replay; có job dọn hoặc drop partition.
7. Lỗi tạm thời thì retry; poison message đi dead-letter, không commit bừa để lag đẹp.
8. Integration test crash sau commit, trước ack.

Sai lầm thường gặp

Sai lầm thường gặp

Ack hoặc commit offset trước khi DB commit. Crash ở giữa làm mất việc mà broker không biết. Xác nhận sau khi ghi bền.

Sai lầm thường gặp

SELECT kiểm tra rồi INSERT không có unique. Hai worker cùng thấy “chưa có”. Để database enforce bằng constraint.

Sai lầm thường gặp

Dùng cờ redelivered hoặc offset làm khóa dedupe. Cờ không bắt được publish hai lần; offset đổi khi replay hay đổi topic. Dùng event ID ổn định.

Tóm tắt

Tóm tắt

• At-least-once nghĩa là message có thể tới lại; side effect phải chịu được điều đó.
• Inbox với unique key và một transaction là cách mặc định cho side effect trong database.
• RabbitMQ redelivered chỉ là gợi ý; Kafka không có cờ tương đương. Cả hai dedupe bằng event ID.
• Chọn retention inbox theo cửa sổ redelivery và replay, không theo cảm tính.

Tự kiểm tra: (1) Crash sau DB commit, trước ack thì chuyện gì xảy ra? (2) Vì sao Set trong RAM không đủ? (3) Một order có payment và refund: dedupe theo orderId có đúng không?

Tham khảo

• Message Delivery Semantics — Apache Kafka 4.1
• Consumer Acknowledgements and Publisher Confirms — RabbitMQ
• INSERT — PostgreSQL 17
• Consuming messages — KafkaJS