Dựng RabbitMQ bằng Docker với management UI, cài thư viện pika, rồi viết một mẫu Work Queue hoàn chỉnh (producer + consumer) áp dụng durable, persistent và manual ack. Cuối cùng là ví dụ Pub/Sub fanout và vài bài tập tự làm.
Image rabbitmq:3-management đã tích hợp giao diện quản trị web. Mở 2 cổng: 5672 (AMQP cho ứng dụng) và 15672 (web UI).
# khởi động broker + management UI docker run -d --name rabbitmq \ -p 5672:5672 \ -p 15672:15672 \ rabbitmq:3-management
Mở http://localhost:15672 và đăng nhập với user/pass mặc định guest / guest. Tại đây xem được exchange, queue, message rate, binding... rất tiện để quan sát trong lúc học.
~/Downloads/cdc-lab đã có sẵn service rabbitmq. Bạn có thể docker compose up -d trong đó thay vì chạy container riêng — tránh trùng cổng 5672/15672.pip install pika
pika là client AMQP chính thức cho Python. Hai script bên dưới đặt cạnh nhau trong một thư mục: producer.py gửi việc, consumer.py nhận và xử lý.
Work Queue phân phối công việc nặng cho nhiều worker. Producer đẩy task vào một queue durable; nhiều consumer cùng lấy ra xử lý song song, prefetch=1 để chia tải công bằng.
producer.py — kết nối, khai báo queue durable, publish message persistent:
# producer.py import sys, pika conn = pika.BlockingConnection(pika.ConnectionParameters("localhost")) ch = conn.channel() # queue bền: tồn tại qua restart broker ch.queue_declare(queue="task_queue", durable=True) message = " ".join(sys.argv[1:]) or "Hello World" ch.basic_publish( exchange="", # default exchange routing_key="task_queue", # = tên queue body=message, properties=pika.BasicProperties(delivery_mode=2), # persistent ) print("đã gửi:", message) conn.close()
consumer.py — đặt prefetch, callback xử lý, manual ack:
# consumer.py import time, pika conn = pika.BlockingConnection(pika.ConnectionParameters("localhost")) ch = conn.channel() ch.queue_declare(queue="task_queue", durable=True) # mỗi consumer chỉ giữ 1 message chưa ack -> chia tải đều ch.basic_qos(prefetch_count=1) def callback(channel, method, props, body): text = body.decode() print("nhận:", text) time.sleep(text.count(".")) # giả lập công việc nặng print("xong") channel.basic_ack(delivery_tag=method.delivery_tag) # ack sau khi xong ch.basic_consume(queue="task_queue", on_message_callback=callback) print("đang chờ message. Ctrl+C để thoát") ch.start_consuming()
python consumer.py — có 2 worker.python producer.py task A... (số dấu chấm = giây xử lý).Khác Work Queue (mỗi message một consumer xử lý), Pub/Sub gửi mỗi message tới mọi queue. Dùng fanout exchange + queue tạm (exclusive) cho từng subscriber.
# publisher: gửi vào fanout exchange ch.exchange_declare(exchange="events", exchange_type="fanout") ch.basic_publish(exchange="events", routing_key="", body="user signed up") # subscriber: queue tạm riêng, nhận mọi message broadcast res = ch.queue_declare(queue="", exclusive=True) qname = res.method.queue ch.queue_bind(queue=qname, exchange="events") ch.basic_consume(queue=qname, on_message_callback=callback, auto_ack=True) ch.start_consuming()
Chạy nhiều subscriber cùng lúc: mỗi cái có queue tạm riêng nên tất cả đều nhận bản sao của mọi event.
logs_topic. Một consumer bind *.error để chỉ nhận log lỗi, một consumer khác bind # để nhận tất cả. Publish các log với routing key như app.info, db.error, api.warning và kiểm chứng định tuyến.task_queue với x-dead-letter-exchange trỏ tới dlx và một task_failed queue bind vào DLX. Trong consumer, nếu body chứa "fail" thì basic_nack(requeue=False) — kiểm tra message rơi vào task_failed trên management UI.retry_queue có x-message-ttl=10000 và DLX trỏ ngược về task_queue. Đọc header x-death để đếm số lần thử; quá 3 lần thì đẩy hẳn sang dead_queue.