Приклад налаштування паралельної системи з високим навантаженням

Постановка задачі

Уявімо, що ми розробляємо застосунок, де користувачі завантажують зображення, а наша система їх обробляє (наприклад, додає водяні знаки, зменшує розмір, конвертує формат). При цьому кількість користувачів зростає, і система має масштабуватися пропорційно навантаженню.

Основні вимоги:

1. Обробка зображень має бути асинхронною.
2. Використання кількох черг, щоб різні типи задач (наприклад, малі та великі зображення) не блокували одна одну.
3. Підтримка високого навантаження зі сотнями задач за секунду.
4. Моніторинг і керування системою.

---

Підготовка середовища

Для роботи нам знадобляться:

• RabbitMQ для роботи з чергами.
• Celery для обробки задач.

Якщо ви пропустили попередні лекції, ось швидкий спосіб розгорнути RabbitMQ за допомогою Docker:

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:management

І не забудьте встановити необхідні бібліотеки:

pip install celery[redis] pillow

---

Налаштування проєкту

Ми будемо використовувати наступний шаблон проєкту:

/parallel_image_processor
    ├── app.py
    ├── celery_app.py
    ├── tasks.py
    ├── settings.py
    ├── utils.py
    └── requirements.txt

Налаштування Celery і RabbitMQ.

Створимо файл settings.py для зберігання конфігурацій:

# settings.py

# Налаштування RabbitMQ
BROKER_URL = 'pyamqp://guest:guest@localhost//'

# Налаштування Celery
CELERY_RESULT_BACKEND = 'rpc://'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'

Тепер підключимо Celery. Створимо файл celery_app.py:

# celery_app.py

from celery import Celery
from settings import BROKER_URL, CELERY_RESULT_BACKEND

# Створюємо екземпляр Celery додатка
celery_app = Celery('parallel_image_processor', broker=BROKER_URL, backend=CELERY_RESULT_BACKEND)

# Завантажуємо налаштування
celery_app.conf.update(
    task_routes={
        'tasks.process_small_image': {'queue': 'small_images'},
        'tasks.process_large_image': {'queue': 'large_images'},
    }
)

---

Завдання

Додамо файл tasks.py і визначимо завдання:

# tasks.py

import time
from celery_app import celery_app
from utils import resize_image, add_watermark

@celery_app.task
def process_small_image(image_path):
    """Обробляє маленьке зображення."""
    print(f"Processing small image: {image_path}")
    resize_image(image_path, size=(200, 200))  # Зменшуємо зображення
    time.sleep(2)  # Емуляція тривалої обробки
    print(f"Small image processed: {image_path}")

@celery_app.task
def process_large_image(image_path):
    """Обробляє велике зображення."""
    print(f"Processing large image: {image_path}")
    add_watermark(image_path, watermark="© Your Brand")  # Додаємо водяний знак
    time.sleep(5)  # Емуляція тривалої обробки
    print(f"Large image processed: {image_path}")

Функції resize_image і add_watermark знаходяться в utils.py.

---

Налаштування моніторингу

Flower — це веб-інтерфейс для моніторингу Celery. Встановимо його:

pip install flower

Запустимо Flower для відстеження завдань:

celery -A celery_app flower

Тепер Flower доступний за адресою: http://localhost:5555. Тут ми можемо побачити активні завдання, статистику воркерів та інші метрики.

---

Запуск системи

Воркери повинні обробляти завдання з різних черг:

celery -A celery_app worker --loglevel=info --queues=small_images
celery -A celery_app worker --loglevel=info --queues=large_images

Публікація завдань.

Додамо в app.py простий приклад:

# app.py

from tasks import process_small_image, process_large_image

# Публікуємо завдання в черги
process_small_image.delay("path/to/small_image.jpg")
process_large_image.delay("path/to/large_image.jpg")

print("Tasks published!")

Запустіть цей файл, щоб завдання почали оброблятися.

---

Аналіз продуктивності

Після запуску системи можна провести навантажувальне тестування, наприклад, за допомогою інструментів на кшталт Locust або Apache JMeter. Відстежте, як зростає час виконання завдань зі збільшенням кількості, і додайте більше воркерів за потреби.

---

Оптимізація

Якщо помітили вузькі місця, ось кілька можливих рішень:

• Збільшіть prefetch_limit для воркерів, щоб задачі розподілялися рівномірно.
• Додайте більше producers/consumers.
• Налаштуйте QoS для пріоритетної обробки завдань.
• Використовуйте моніторинг і аналізуйте метрики RabbitMQ, щоб виявити перевантажені черги.

---

Приклад навантаження

Перевірте, як система справляється з високим трафіком. Запустіть багато завдань у циклі:

for i in range(100):
    process_small_image.delay(f"small_image_{i}.jpg")
    process_large_image.delay(f"large_image_{i}.jpg")

Відстежуйте, як черги заповнюються і як воркери їх обробляють.

---

Підсумковий результат

Наша система справляється з сотнями задач за секунду завдяки використанню кількох черг, розподілу навантаження і паралельної обробки. А за допомогою інструментів моніторингу ми отримали можливість керувати та оптимізувати продуктивність у реальному часі.

🎉 Вітаю, ви щойно налаштували паралельну систему з високим навантаженням! Тепер ви знаєте, як застосовувати Celery і RabbitMQ у реальних проєктах.

Якщо щось піде не так — не біда! Як кажуть досвідчені розробники,
«помилки — це спосіб системи сказати вам, що ви ще не все про неї знаєте». Удачі у ваших проєктах!