پردازش‌های هوش مصنوعی ماهیت سنگین، وابستگی شدید به سخت‌افزار (GPU/NPU) و زمان اجرای متغیر دارند. الگوی سنتی Synchronous HTTP/gRPC که در معماری‌های متداول کاربرد دارد، در مواجهه با وظایف AI با شکست مواجه می‌شود. در این مقاله تخصصی، نحوه طراحی، پیاده‌سازی و مقیاس‌دهی یک سیستم صف‌بندی وظایف (Task Queue System) برای بار کاری AI با استفاده از RabbitMQ را به‌صورت عمیق بررسی خواهیم کرد.

چرا الگوی همزمان (Synchronous) در پردازش AI شکست می‌خورد؟

در یک معماری همزمان، کلاینت درخواست HTTP ارسال کرده و تا زمان پایان پردازش و دریافت پاسخ، اتصال (Connection) را باز نگه می‌دارد. این الگو برای بارهای کاری هوش مصنوعی به دلایل زیر غیرقابل استفاده است:

1. بلوکه شدن نخ‌ها (Thread Starvation): اگر زمان استنتاج (Inference) یک مدل مانند Stable Diffusion یا Llama-3 بین ۵ تا ۴۰ ثانیه طول بکشد، نخ‌های وب‌سرور (مانند Kestrel در .NET یا Gunicorn در Python) به سرعت اتمام یافته و وب‌سرور از دسترس خارج می‌شود.

2. پیش‌آمد زمان اتمام اتصال (HTTP Timeout): در پردازش‌های دسته‌ای (Batch Processing) یا مدل‌های سنگین، لایه‌های واسط نظیر Nginx، Cloudflare یا API Gateway اتصالات طولانی‌مدت را قطعی قطع می‌کنند.

3. عدم تحمل خطا (Lack of Fault Tolerance): اگر گره پردازش‌کننده (Worker Node) در حین اجرای مدل به دلیل خطای اتمام حافظه کارت گرافیک (CUDA Out of Memory) منهدم شود، درخواست کلاینت کاملاً از دست رفته و هیچ سازوکار بازتولید یا Re-queue داخلی وجود ندارد.

4. عدم امکان کنترل جریان (Lack of Backpressure): پیک‌های ناگهانی ترافیک (Traffic Spikes) مستقیماً به گره‌های GPU فشار آورده و باعث خرابی کل سرویس می‌شوند.

[درخواست همزمان - نامناسب]
Client  ──(HTTP POST)──>  API Gateway  ──(Sync Call)──>  GPU Worker (OOM Crash!)
  ▲                                                           │
  └───────────────────(504 Gateway Timeout)───────────────────┘

[درخواست نامتقارن با RabbitMQ - استاندارد]
Client  ──(HTTP POST)──>  API Gateway  ──(Publish Job)──>  [RabbitMQ Queue]
  ▲                           │                                │
  │ (202 Accepted + JobID)    │                                ▼
  └───────────────────────────┘                        GPU Workers (Pull/Ack)

جایگاه RabbitMQ در زیست‌بوم بارهای کاری AI

کارگزار پیام RabbitMQ به دلیل پیاده‌سازی پروتکل AMQP (Advanced Message Queuing Protocol)، قابلیت‌های بسیار پیشرفته‌ای در اختیار معماران سیستم قرار می‌دهد که آن را برای صف‌بندی وظایف AI نسبت به ابزارهایی مانند Redis یا Kafka متمایز می‌کند:

• هدایت هوشمند پیام (Smart Routing): با استفاده از انواع Exchangeها (Direct, Topic, Fanout, Consistent-Hash) می‌توان وظایف را بر اساس نوع مدل (مثلاً مدل‌های متنی، تصویری، صوتی) به صف‌های اختصاصی هدایت کرد.

• کنترل دقیق جریان (Granular Backpressure & QoS): قابلیت تنظیم Prefetch Count مانع از احتکار پیام‌ها توسط یک Worker پردازش تصویر شده و توزیع متوازن بار روی GPUها را تضمین می‌کند.

• پایداری و تضمین تحویل پیام (Delivery Guarantees): قابلیت Persistent Queue و Message Acknowledgement عدم از دست رفتن وظایف سنگین پردازشی را حتی در صورت سقوط کامل گره‌ها تضمین می‌کند.

مفاهیم کلیدی RabbitMQ در معماری پردازش AI

برای طراحی درست سیستم، باید مفاهیم RabbitMQ را دقیقاً با نیازمندی‌های پردازش هوش مصنوعی تطبیق دهیم:

الف) انواع Exchange برای هدایت وظایف AI

1. Direct Exchange: برای سرویس‌های مشخص. مثلاً هدایت مستقیم وظیفه ai.task.yolo به صف پردازش کارت‌های گرافیک اختصاصی Computer Vision.

2. Topic Exchange: برای سناریوهای پیچیده و چندوجهی. به عنوان مثال الگوهای مسیردهی نظیر ai.image.upscale.v2 یا ai.text.summarize.llama3.

3. Consistent-Hash Exchange: جهت توزیع متوازن بار بر اساس کلید هش (مثلاً user_id یا session_id) برای جلوگیری از تداخل حافظه کش مدل‌ها روی Workerها.

ب) مفهوم نرخ پیش‌دریافت (Prefetch Count) و اهمیت حیاتی آن

در سرویس‌های معمول وب، مقدار Prefetch ممکن است ۱۰۰ یا ۱۰۰۰ تنظیم شود. اما در پردازش‌های AI، مقدار Prefetch Count باید دقیقاً برابر با تعداد وظایف همزمان قابل پردازش روی کارت گرافیک (معمولاً ۱ یا ۲) باشد.

اگر Prefetch Count بزرگتر از capacity واقعی GPU باشد، RabbitMQ پیام‌ها را به Worker متصل منتقل کرده و در حافظه RAM آن صف‌بندی می‌کند. در این حالت، اگر این Worker به دلیل خطای CUDA متوقف شود، تمامی آن پیام‌های پیش‌دریافت‌شده معلق می‌مانند؛ در حالی که سایر Workerهای سالم بی‌کار (Idle) هستند.

\text{Optimal Prefetch Count} = \text{Available Concurrent GPU Slots per Worker}

معماری مرجع (Reference Architecture) سیستم صف‌بندی AI

در این معماری، لایه API صرفاً وظیفه اعتبارسنجی اولیه، ذخیره فایل‌های ورودی در Object Storage (مانند MinIO یا AWS S3) و ارسال یک «فرمان اجرای وظیفه» (Task Command) به RabbitMQ را بر عهده دارد.

                                    ┌────────────────────────┐
                                    │      MinIO / S3        │
                                    │    (Object Store)      │
                                    └───────────▲────────────┘
                                                │
                                   1. Upload    │  3. Read Input /
                                      Input     │     Write Output
                                                │
┌──────────┐     2. Publish Task    ┌───────────┴────────────┐     4. Consume Job     ┌────────────────────────┐
│  API     │───────────────────────>│     RabbitMQ Broker    │───────────────────────>│   GPU Worker Node 1    │
│ Gateway  │                        │ (Direct/Topic Exchange)│                        │  (PyTorch/YOLO/vLLM)   │
└──────────┘                        └────────────────────────┘                        └────────────────────────┘
     │                                           │                                                 │
     │ 202 Accepted                              │ Dead Letter Exchange                            │ 5. Publish Result /
     │ (Return JobID)                            ▼ (On Failure)                                    │    Acknowledge
     ▼                              ┌────────────────────────┐                                     ▼
┌──────────┐                        │   Dead Letter Queue    │                        ┌────────────────────────┐
│ Client   │                        │  (Poison Pill / Retry) │                        │  Redis / Notification  │
└──────────┘                        └────────────────────────┘                        │  (WebSockets/Webhooks) │
                                                                                      └────────────────────────┘

ثبت درخواست: کلاینت تصویر یا متن را ارسال می‌کند. API تصویر را در S3 ذخیره کرده و یک JobId تولید می‌کند.

1. ارسال پیام (Produce): پیام شامل JobId و آدرس فایل در S3 به RabbitMQ ارسال می‌شود. API بلافاصله پاسخ 202 Accepted شامل JobId را به کلاینت برمی‌گرداند.

2. دریافت پیام (Consume): یکی از Workerهای آزاد پایتون/سی‌شارپ پیام را برداشته (Prefetch=1)، تصویر را از S3 دانلود و استنتاج مدل را روی GPU انجام می‌دهد.

3. ثبت نتیجه: نتیجه در دیتابیس/Redis ذخیره شده و خروجی فیزیکی در S3 آپلود می‌شود.

4. تاییدیه (ACK): Worker پیام را در RabbitMQ تایید (BasicAck) می‌کند.

5. اطلاع‌رسانی: سیستم از طریق WebSocket یا Webhook پایان پردازش را به کلاینت اعلام می‌کند.

الگوهای پیشرفته صف‌بندی در سناریوهای واقعی AI

الف) صف‌های اولویت‌دار (Priority Queues)

در اکثر سیستم‌های تجاری، کاربران طرح‌های متفاوت (Free vs Pro/VIP) دارند. نباید اجازه داد پردازش‌های سنگین کاربران رایگان باعث تاخیر در پردازش کاربران VIP شود.

در RabbitMQ با تنظیم آرگومان x-max-priority روی صف، می‌توان پیام‌ها را با اولویت‌های متفاوت (مثلاً بین ۰ تا ۱۰) ارسال کرد:

// تنظیم صف با قابلیت پشتیبانی از اولویت در .NET
var arguments = new Dictionary<string, object>
{
    { "x-max-priority", 10 }
};
channel.QueueDeclare(queue: "ai_inference_tasks", durable: true, exclusive: false, autoDelete: false, arguments: arguments);

ب) الگوی Dead Letter Exchange (DLX) و مدیریت خطاهای GPU

در پردازش AI، خطاها به دو دسته تقسیم می‌شوند:

1. خطاهای موقت (Transient Errors): مانند عدم دسترسی لحظه‌ای به S3 یا تکمیل بودن موقت VRAM. این خطاها باید با الگوی Exponential Backoff دوباره سعی (Retry) شوند.

2. خطاهای قطعی (Poison Messages): مانند خراب بودن فایل تصویر ورودی، نادرست بودن فرمت یا خطای ساختاری کد. این پیام‌ها هرگز نباید بی‌نهایت به صف اصلی برگردند، زیرا باعث چرخه بی‌نهایت خرابی (Crash Loop) می‌شوند.

با استفاده از DLX، پیام‌هایی که بیش از حد مجاز (مثلاً ۳ بار) رد شوند (BasicNack با requeue=false)، خودکار به Dead Letter Queue منتقل شده تا توسط تیم پشتیبانی یا فرآیندهای ایزوله بررسی شوند.

پیاده‌سازی عملی (Code Implementation)

برای درک کامل، پیاده‌سازی لایه ارسال‌کننده (Publisher) در C# (.NET Core) و لایه پردازش‌کننده (Consumer) در Python را بررسی می‌کنیم.

بخش اول: پیاده‌سازی Publisher در .NET 8 / C#

using System.Text;
using System.Text.Json;
using RabbitMQ.Client;

public class AiTaskPublisher
{
    private readonly string _hostname = "localhost";
    private readonly string _queueName = "ai_vision_tasks";

    public async Task PublishInferenceJobAsync(string jobId, string fileS3Url, int priorityLevel)
    {
        var factory = new ConnectionFactory() { HostName = _hostname };
        using var connection = await factory.CreateConnectionAsync();
        using var channel = await connection.CreateChannelAsync();

        // 1. تعریف صف با قابلیت پایداری (Durable) و اولویت‌بندی
        var args = new Dictionary<string, object> { { "x-max-priority", 10 } };
        await channel.QueueDeclareAsync(
            queue: _queueName,
            durable: true,
            exclusive: false,
            autoDelete: false,
            arguments: args
        );

        // 2. ساخت بدنه پیام (Payload)
        var payload = new
        {
            JobId = jobId,
            InputUrl = fileS3Url,
            ModelName = "yolov8x-detection",
            CreatedAt = DateTime.UtcNow
        };

        var messageBody = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(payload));

        // 3. تنظیمات پایداری پیام (Persistent Message) و اولویت
        var properties = new BasicProperties
        {
            Persistent = true,
            Priority = (byte)priorityLevel
        };

        // 4. ارسال به Exchange پیش‌فرض
        await channel.BasicPublishAsync(
            exchange: string.Empty,
            routingKey: _queueName,
            mandatory: true,
            basicProperties: properties,
            body: messageBody
        );

        Console.WriteLine($"[C# Publisher] AI Task Published: {jobId} with Priority: {priorityLevel}");
    }
}

بخش دوم: پیاده‌سازی GPU Worker در Python (PyTorch/YOLO)

این کد یک پردازنده حرفه‌ای هوش مصنوعی را نشان می‌دهد که پیام‌ها را دریافت کرده، کنترل جریان (prefetch_count=1) را اعمال می‌کند و مدیریت حافظه CUDA را به عهده می‌گیرد.

import pika
import json
import time
import torch
import logging

logging.basicConfig(level=logging.INFO)

RABBITMQ_HOST = 'localhost'
QUEUE_NAME = 'ai_vision_tasks'

def simulate_gpu_inference(input_url, model_name):
    """شبیه‌سازی بارگذاری مدل و استنتاج روی GPU"""
    logging.info(f"Loading data from {input_url} into GPU VRAM...")
    time.sleep(3) # شبیه‌سازی زمان استنتاج سنگین
    
    # در پروژه‌های واقعی:
    # results = model(input_data)
    # torch.cuda.empty_cache()  --> آزادسازی حافظه جهت جلوگیری از OOM
    
    if "corrupted" in input_url:
        raise ValueError("Poison Pill: Input file is corrupted!")
        
    return {"status": "SUCCESS", "objects_detected": ["car", "person"], "confidence": 0.94}

def process_task(ch, method, properties, body):
    job_data = json.loads(body.decode('utf-8'))
    job_id = job_data.get('JobId')
    input_url = job_data.get('InputUrl')
    model_name = job_data.get('ModelName')

    logging.info(f"[Python Worker] Processing Job ID: {job_id}")

    try:
        # اجرای استنتاج مدل AI
        result = simulate_gpu_inference(input_url, model_name)
        logging.info(f"[Python Worker] Job {job_id} Processed Successfully. Result: {result}")
        
        # تایید پیام پس از انجام موفقیت‌آمیز وظیفه
        ch.basic_ack(delivery_tag=method.delivery_tag)

    except ValueError as ve:
        # خطای قطعی (Poison Pill): رد کردن پیام بدون ورود مجدد به صف (هدایت به DLQ)
        logging.error(f"[Fatal Error] Task {job_id} failed permanently: {ve}")
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

    except Exception as e:
        # خطای موقت: بازگرداندن پیام به صف جهت تلاش مجدد
        logging.warning(f"[Transient Error] Task {job_id} failed: {e}. Re-queuing...")
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

def start_worker():
    connection = pika.BlockingConnection(pika.ConnectionParameters(host=RABBITMQ_HOST))
    channel = connection.channel()

    # تعریف صف همانند لایه ارسال‌کننده
    channel.queue_declare(queue=QUEUE_NAME, durable=True, arguments={'x-max-priority': 10})

    # کلید اصلی تضمین کارکرد GPU: تنظیم Prefetch Count روی ۱
    channel.basic_qos(prefetch_count=1)

    channel.basic_consume(queue=QUEUE_NAME, on_message_callback=process_task)

    logging.info("[Python Worker] Ready and waiting for AI tasks. To exit press CTRL+C")
    channel.start_consuming()

if __name__ == '__main__':
    start_worker()

مقیاس‌پذیری خودکار (Auto-scaling) با KEDA و Kubernetes

یکی از درخشان‌ترین کاربردهای ترکیب RabbitMQ و پردازش AI، مقیاس‌دهی خودکار گره‌های پردازشی بر اساس طول صف است.

با استفاده از KEDA (Kubernetes Event-driven Autoscaling)، می‌توان Podهای حاوی گره‌های پردازش GPU را به صورت متناسب با تعداد پیام‌های منتظر در صف RabbitMQ کم یا زیاد کرد.

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: rabbitmq-gpu-worker-scaler
  namespace: ai-processing
spec:
  scaleTargetRef:
    name: gpu-inference-worker-deployment
  minReplicaCount: 0  # مقیاس‌دهی به صفر در صورت خالی بودن صف (صرفه‌جویی هزینه GPU)
  maxReplicaCount: 10 # حداکثر گره‌های GPU قابل تخصیص
  cooldownPeriod: 300
  triggers:
  - type: rabbitmq
    metadata:
      protocol: amqp
      queueName: ai_vision_tasks
      mode: QueueLength
      value: "5" # به ازای هر ۵ پیام معلق در صف، یک Pod جدید GPU روشن کن
    authenticationRef:
      name: keda-rabbitmq-auth

مزیت اقتصادی: با تنظیم minReplicaCount: 0 در KEDA، هنگامی که هیچ درخواست پردازش هوش مصنوعی وجود ندارد، نمونه‌های گران‌قیمت GPU در ابر (مثل AWS EC2 g5 instances) خاموش شده و هزینه به شدت کاهش می‌یابد.

مقایسه جامع: RabbitMQ در برابر سایر راهکارهای صف‌بندی برای AI

چک‌لیست طلایی استقرار در محیط Production

برای جلوگیری از خرابی‌های رایج در محیط عملیاتی، رعایت نکات زیر الزامی است:

• [ ] فعال‌سازی گزینه‌های پایداری (Durability): هم صف‌ها (durable=true) و هم پیام‌ها (persistent=true) باید پایدار تعریف شوند تا با ریستارت شدن کارگزار، وظایف از بین نروند.

• [ ] تنظیم محدودیت‌های حافظه (Memory High Watermark): در فایل تنظیمات RabbitMQ، حد مجاز استفاده از RAM را تعیین کنید تا در صورت پر شدن حافظه، کارگزار به جای Crash کردن، روی لایه دیسک بفرستد یا ورود پیام جدید را بلوکه کند.

• [ ] پایش (Monitoring) با Prometheus و Grafana: متریک‌های حیاتی مانند rabbitmq_queue_messages_ready (تعداد پیام‌های معلق) و rabbitmq_queue_messages_unacknowledged (پیام‌های در حال پردازش توسط GPU) را در داشبورد監視 کنید.

• [ ] مدیریت قطع ارتباط (Heartbeat Timeout): به دلیل اینکه پردازش‌های AI ممکن است طویل‌المدت باشند، نباید رشته اصلی اتصال RabbitMQ را حین پردازش بلوکه کنید. فرایند شبکه AMQP و فرایند استنتاج GPU باید در نخ‌های مجزا اجرا شوند تا Heartbeat قطعی نشود.

استفاده از RabbitMQ به عنوان لایه واسط پردازش نامتقارن، معماری سرویس‌های مبتنی بر هوش مصنوعی را از یک سیستم شکننده و وابسته به زمان، به یک زیرساخت مقیاس‌پذیر، نفوذناپذیر در برابر پیک‌های ترافیکی و مقاوم در برابر خطا (Resilient) تبدیل می‌کند. با بهینه‌سازی پارامترهایی چون Prefetch Count روی اعداد پایین، جداسازی وظایف با انواع Exchange، پیاده‌سازی مکانیزم‌های بازتلاش خودکار (DLX) و تلفیق آن با ابزارهای مقیاس‌دهی خودکار پایه صف مانند KEDA، می‌توان خطوط پردازش هوش مصنوعی را با بالاترین بهره‌وری سخت‌افزاری و نازل‌ترین هزینه‌های عملیاتی مدیریت کرد.