Інтеграція Celery з RabbitMQ для обробки завдань

Давайте розберемося, як зв'язати Celery і RabbitMQ, щоб побудувати справжню асинхронну магію для обробки завдань. RabbitMQ буде нашим брокером повідомлень, який роздає завдання воркерам Celery. В результаті отримаємо систему, яка може масштабуватися, обробляти запити у фоні і витримувати великі навантаження.

Коли ваш сервер Web-застосунку стикається з важкими завданнями, наприклад обробкою великих даних або відправкою тонни листів, він може почати "задихатися". Ми хочемо розгрузити сервер, відправляючи ці завдання у фоновий режим. Ось тут і вступає RabbitMQ.

RabbitMQ працює як посередник:

• Він приймає завдання від вашого застосунку (продюсера).
• Зберігає їх у чергах.
• Воркери Celery забирають завдання і виконують їх.

Це як доставка піци:

• FastAPI/Django — кухар, який віддає піцу кур'єру.
• RabbitMQ — кур'єр, який везе піцу клієнту.
• Celery — клієнт, який із задоволенням з'їдає піцу (виконує завдання).

---

Встановлення RabbitMQ

Почнемо з встановлення RabbitMQ, щоб він став частиною нашого асинхронного стеку.

Якщо у вас вже встановлений Docker, встановлення RabbitMQ займе всього кілька хвилин. Відкрийте термінал і виконайте:

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

• Порт 5672 використовується для спілкування RabbitMQ з Celery.
• Порт 15672 дозволяє вам отримати доступ до зручного веб-інтерфейсу RabbitMQ.

Після успішного запуску ви можете відкрити браузер і перейти за адресою http://localhost:15672. Логін: guest, пароль: guest.

Якщо Docker не ваш вибір, перевірте офіційну документацію RabbitMQ для встановлення на вашому комп'ютері.

---

Налаштування Celery для використання RabbitMQ

Тепер пов'яжемо Celery з RabbitMQ у нашому проєкті.

Крок 1: встановлення залежностей

Спочатку встановимо Celery і бібліотеку kombu, яка забезпечує зв'язок з RabbitMQ:

pip install celery kombu

Якщо ви працюєте з Django, переконайтеся, що Celery вже інтегрований у ваш проєкт, як це обговорювалося в попередніх лекціях.

Крок 2: налаштування конфігурації Celery

Тепер створимо або оновимо файл конфігурації Celery (наприклад, celery.py у вашому проєкті). Цього разу вкажемо RabbitMQ як брокер повідомлень.

Приклад конфігурації:

from celery import Celery

# Створюємо застосунок Celery
app = Celery("example_project")

# Вказуємо RabbitMQ як брокер повідомлень
app.conf.broker_url = "amqp://guest:guest@localhost:5672//"

# Додаткові налаштування (опційно)
app.conf.result_backend = "rpc://"
app.conf.task_serializer = "json"  # Для серіалізації даних завдань
app.conf.accept_content = ["json"]  # Прийняті формати даних
app.conf.timezone = "UTC"  # Часовий пояс

• amqp://guest:guest@localhost:5672// — рядок підключення до RabbitMQ. Тут використовується стандартний логін і пароль guest. Якщо ви налаштували свої дані для доступу, замініть їх.

Крок 3: оголошення простої задачі

Додамо просту задачу, щоб протестувати роботу Celery через RabbitMQ. Створіть файл tasks.py:

from celery import shared_task

@shared_task
def add(x, y):
    return x + y

Крок 4: запуск RabbitMQ і Celery

1. Переконайтеся, що RabbitMQ працює. У випадку з Docker це легко перевірити командою:

   
   docker ps
           

2. Запустіть Celery воркери:

   
   celery -A example_project worker --loglevel=info
           

   Замініть example_project на ім'я вашого проєкту.

---

Перевірка роботи Celery і RabbitMQ

Тепер, коли все налаштовано, викличемо нашу задачу add з Python Shell (або через Django):

from tasks import add

result = add.delay(3, 5)
print(f"Task ID: {result.id}")
print(f"Task Result: {result.get()}")

Подивіться на логи в терміналі, де запущений воркер Celery. Ви маєте побачити обробку задачі. Це означає, що RabbitMQ успішно отримав задачу і передав її Celery.

---

Переваги використання RabbitMQ з Celery

RabbitMQ надає кілька переваг, які роблять його ідеальним брокером повідомлень для Celery:

• Збереженість завдань. RabbitMQ надійно зберігає завдання навіть при збоях у мережі.
• Масштабованість. RabbitMQ підтримує десятки черг, які можуть обробляти Celery-воркери.
• Довговічність даних. Ви можете налаштувати "durable" черги, щоб завдання зберігалися навіть між перезапусками RabbitMQ.
• Підтримка високих навантажень. RabbitMQ добре справляється з передачею великої кількості завдань одночасно.

---

Приклад складнішої задачі

Давайте створимо складнішу задачу, наприклад, відправку електронної пошти:

import time
from celery import shared_task

@shared_task
def send_email(recipient, subject, body):
    print(f"Sending email to {recipient} with subject '{subject}'...")
    time.sleep(5)  # Імітуємо довгу відправку
    print("Email sent!")

Таку задачу можна використовувати для реальних великих проєктів, де потрібно відправляти сповіщення, звіти або масові розсилки листів.

---

Налаштування черг і маршрутів задач

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

app.conf.task_routes = {
    "tasks.send_email": {"queue": "email_queue"},
    "tasks.add": {"queue": "math_queue"},
}

Тепер RabbitMQ буде використовувати черги email_queue і math_queue для маршрутизації задач. Це спрощує масштабування, бо можна запускати воркерів для конкретних черг.

---

Примітки щодо реальної розробки

1. Якщо ви використовуєте RabbitMQ у продакшені, обов'язково налаштуйте свої логіни й паролі. За замовчуванням використовуються "guest/guest", що небезпечно.
2. Слідкуйте за продуктивністю RabbitMQ, щоб уникнути перевантаження черг. Використовуйте веб-інтерфейс RabbitMQ для перегляду завантаження.
3. Залежно від задач, може знадобитися тонке налаштування конфігурації воркерів Celery (наприклад, кількість процесів, таймаути).

Тепер у вас є повноцінна інтеграція Celery з RabbitMQ для обробки асинхронних завдань. У наступній лекції ми продовжимо створювати більш складні та реальні сценарії, використовуючи можливості асинхронних систем.