Message Queues Trong Production: Từ Địa Ngục Synchronous Đến Thiên Đường Async

@app.post("/send-newsletter")
def send_newsletter():
    users = db.query("SELECT * FROM users WHERE subscribed = true")
    for user in users:
        send_email(user.email, newsletter_content)
    return {"message": "Done"}

# Producer (API) - Trả về ngay lập tức
@app.post("/send-newsletter")
def send_newsletter():
    users = db.query("SELECT * FROM users WHERE subscribed = true")

    # Queue tất cả email jobs (chỉ mất 3 giây)
    for user in users:
        queue.push({
            "type": "send_email",
            "to": user.email,
            "template": newsletter_content
        })

    return {"status": "queued", "total": len(users)}
    # Response ngay lập tức!

# Consumer (Background Worker) - Xử lý từ từ
def process_message(message):
    if message["type"] == "send_email":
        send_email(message["to"], message["template"])
        message.ack()  # Xác nhận đã xử lý

import pika
import json

class QueueProducer:
    def __init__(self):
        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters('localhost')
        )
        self.channel = self.connection.channel()
        self.channel.queue_declare(queue='tasks', durable=True)

    def push(self, message: dict):
        self.channel.basic_publish(
            exchange='',
            routing_key='tasks',
            body=json.dumps(message),
            properties=pika.BasicProperties(
                delivery_mode=2,  # Persistent
            )
        )

    def close(self):
        self.connection.close()

producer = QueueProducer()
producer.push({"type": "send_email", "to": "user@example.com"})

import pika
import json

class QueueConsumer:
    def __init__(self):
        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters('localhost')
        )
        self.channel = self.connection.channel()
        self.channel.queue_declare(queue='tasks', durable=True)
        self.channel.basic_qos(prefetch_count=1)  # One message at a time

    def start_consuming(self):
        self.channel.basic_consume(
            queue='tasks',
            on_message_callback=self.process_message
        )
        print("Waiting for messages...")
        self.channel.start_consuming()

    def process_message(self, ch, method, properties, body):
        message = json.loads(body)
        try:
            if message["type"] == "send_email":
                send_email(message["to"], message["template"])
            ch.basic_ack(delivery_tag=method.delivery_tag)  # Xác nhận thành công
        except Exception as e:
            print(f"Error: {e}")
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)  # Retry sau

consumer = QueueConsumer()
consumer.start_consuming()

from fastapi import FastAPI

app = FastAPI()
producer = QueueProducer()

@app.post("/send-newsletter")
async def send_newsletter():
    users = db.query("SELECT * FROM users WHERE subscribed = true")

    for user in users:
        producer.push({
            "type": "send_email",
            "to": user.email,
            "user_id": user.id
        })

    return {
        "status": "queued",
        "total": len(users),
        "estimated_time": "~2 minutes"
    }
    # Response trong 2 giây (thay vì 30 phút đồng bộ)

@app.post("/process-files")
async def process_files(files: list):
    job_id = str(uuid.uuid4())

    for file in files:
        producer.push({
            "type": "process_file",
            "job_id": job_id,
            "file_path": file.path,
            "file_size": file.size
        })

    return {
        "job_id": job_id,
        "status": "processing",
        "files": len(files)
    }

# SAI (không idempotent)
def process_order(order_id):
    charge_credit_card(order_id, 100)  # Tính tiền 2 lần nếu chạy 2 lần!

# ĐÚNG (idempotent)
def process_order(order_id):
    if db.order_already_charged(order_id):
        return  # Đã xử lý rồi

    charge_credit_card(order_id, 100)
    db.mark_order_charged(order_id)

import time

class RetryableConsumer:
    MAX_RETRIES = 3
    DELAYS = [60, 300, 900]  # 1 phút, 5 phút, 15 phút

    def process_with_retry(self, message, delivery_tag):
        for attempt in range(self.MAX_RETRIES + 1):
            try:
                self.process_message(message)
                self.channel.basic_ack(delivery_tag=delivery_tag)
                return
            except TemporaryError as e:
                if attempt < self.MAX_RETRIES:
                    delay = self.DELAYS[attempt]
                    print(f"Attempt {attempt + 1} failed, retrying in {delay}s: {e}")
                    time.sleep(delay)
                else:
                    # Hết số lần retry, chuyển sang DLQ
                    self.send_to_dlq(message)
                    self.channel.basic_ack(delivery_tag=delivery_tag)
            except PermanentError:
                self.send_to_dlq(message)
                self.channel.basic_ack(delivery_tag=delivery_tag)
                return

# Khai báo DLQ
channel.queue_declare(queue='tasks_dlq', durable=True)

# Producer cho DLQ
def send_to_dlq(message):
    dlq_message = {
        **message,
        "error": str(e),
        "failed_at": datetime.now().isoformat(),
        "retry_count": attempt + 1
    }

    channel.basic_publish(
        exchange='',
        routing_key='tasks_dlq',
        body=json.dumps(dlq_message),
        properties=pika.BasicProperties(
            delivery_mode=2
        )
    )

# Producer
queue.push({"type": "resize_image", "path": "/uploads/photo.jpg"})

# Consumer - một worker xử lý
def resize_image(message):
    resize(message["path"])
    message.ack()

# Producer
kafka.send("user.events", {"user_id": 123, "event": "page_view", "page": "/home"})

# Consumer 1 - Analytics
kafka.subscribe("user.events")
def track_analytics(event):
    analytics_db.increment(event["event"])

# Consumer 2 - Recommendations
kafka.subscribe("user.events")
def update_recommendations(event):
    rec_engine.track(event["user_id"], event["page"])

# API Server - FastAPI
from fastapi import FastAPI
import pika
import json

app = FastAPI()
connection = pika.BlockingConnection(
    pika.ConnectionParameters('rabbitmq-cluster')
)
channel = connection.channel()

@app.post("/api/send-newsletter")
def send_newsletter(payload: NewsletterRequest):
    users = get_subscribed_users()

    for user in users:
        channel.basic_publish(
            exchange='email.tasks',
            routing_key='email.send',
            body=json.dumps({
                "user_id": user.id,
                "email": user.email,
                "template": payload.template_id
            }),
            properties=pika.BasicProperties(
                delivery_mode=2,
                priority=1
            )
        )

    return {"queued": len(users)}

# Worker - Background Service
import pika
import json

class EmailWorker:
    def __init__(self, worker_id):
        self.worker_id = worker_id
        self.connection = pika.BlockingConnection(...)
        self.channel = self.connection.channel()
        self.channel.basic_qos(prefetch_count=10)

    def start(self):
        self.channel.basic_consume(
            queue='email.send',
            on_message_callback=self.handle_email
        )
        self.channel.start_consuming()

    def handle_email(self, ch, method, properties, body):
        data = json.loads(body)
        try:
            send_email_smtp(data["email"], data["template"])
            ch.basic_ack(method.delivery_tag)
            logger.info(f"Worker {self.worker_id}: Sent to {data['email']}")
        except smtplib.SMTPServerDisconnected:
            ch.basic_nack(method.delivery_tag, requeue=True)
        except Exception as e:
            logger.error(f"Failed: {e}")
            ch.basic_nack(method.delivery_tag, requeue=False)  # → DLQ

from prometheus_client import Gauge, Counter

queue_depth = Gauge('queue_depth', 'Messages waiting', ['queue'])
processed = Counter('messages_processed_total', 'Messages processed', ['queue', 'status'])
dlq_count = Gauge('dlq_count', 'Messages in DLQ', ['queue'])

# Worker health check
@app.get("/health")
def health():
    return {
        "queue_depth": queue_depth,
        "dlq_count": dlq_count,
        "workers_active": len(active_workers)
    }

def autoscale():
    depth = get_queue_depth('emails')
    workers = get_worker_count()

    if depth > 500 and workers < 10:
        launch_worker()
    elif depth < 50 and workers > 2:
        terminate_worker()

# Rules:
# Depth < 50: 2 workers
# Depth 50–200: 5 workers
# Depth > 500: 10 workers

@timeout(seconds=300)
def process_task(message):
    # Xử lý ở đây
    pass

# Không queue (operation 5ms)
def get_user(id):
    return db.get(id)

# Queue (operation 5 phút)
def process_video(path):
    queue.push({"type": "process_video", "path": path})