TopolearnDistributed messaging always produces duplicates and reordering in some failure; robust systems are designed for it.
With publisher confirms and manual consumer acks, RabbitMQ provides at-least-once delivery. Without confirms, or with auto-ack, it is at-most-once. Exactly-once delivery doesn't exist across a network: a confirm or an ack can be lost after the work happened, so the message is sent or processed again.
Duplicates arise when a publisher retries after a lost confirm, or when a consumer crashes or times out after doing work but before acking. Make processing idempotent:
- Give each message a unique
message_idset by the producer. - Record processed IDs in the same database transaction as the side effect, and skip IDs already seen.
- Or design operations to be naturally idempotent: "set status to paid" instead of "add 10 to balance".
Ordering: a single queue is FIFO, and with one consumer messages are processed in publish order (from a single publisher on one channel). With several consumers, prefetch and requeues, processing order is not guaranteed: a requeued message comes back later than its successors. When order per entity matters, use single active consumer (x-single-active-consumer) on a queue, or partition by key with the consistent hash exchange or super streams so each partition has one consumer.
Throughput tuning trades against these properties: larger prefetch, batching confirms and multiple consumers increase throughput but widen the window for duplicates and reordering. The transactional outbox pattern (write the message to your database first, publish from there) keeps database state and published events consistent.
Example
import json, sqlite3, pika
db = sqlite3.connect("app.db")
db.execute("CREATE TABLE IF NOT EXISTS processed (id TEXT PRIMARY KEY)")
db.execute("CREATE TABLE IF NOT EXISTS payments (order_id INT, amount INT)")
def on_message(ch, method, props, body):
try:
with db: # one transaction
db.execute("INSERT INTO processed VALUES (?)", (props.message_id,))
msg = json.loads(body)
db.execute("INSERT INTO payments VALUES (?, ?)", (msg["order"], msg["amount"]))
except sqlite3.IntegrityError:
print("duplicate", props.message_id, "skipped") # already processed
ch.basic_ack(method.delivery_tag) # ack after commit (or skip)
conn = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
ch = conn.channel()
ch.queue_declare("payments", durable=True,
arguments={"x-single-active-consumer": True}) # strict order
ch.basic_consume("payments", on_message)
ch.start_consuming()
# producer: properties=pika.BasicProperties(message_id="pay-1001", ...)Remember
- Confirms + manual acks = at-least-once; exactly-once must be built by you
- Deduplicate by message_id in the same transaction as the side effect
- Order holds per queue with one consumer; use single active consumer or partitioning
Watch out for
- Assuming messages are processed in order with several competing consumers.