Основы параллельной обработки задач в системах

Параллельная обработка задач — это как наличие нескольких кассиров для обработки очереди в супермаркете или банке. При таком подходе нескольких клиентов обслуживают одновременно, что сокращает время ожидания в очереди. В программировании всё точно так же.

В мире приложений мы сталкиваемся с такими задачами, как обработка запросов пользователей, выполнение сложных вычислений и обработка данных. Если выполнять всё это последовательно, то производительность системы резко пострадает. В таких случаях параллельная обработка становится ключевым элементом для повышения эффективности.

---

Параллельная vs последовательная обработка

В последовательной обработке каждая задача выполняется одна за другой. Например:

def process_tasks_sequentially(tasks):
    for task in tasks:
        perform_task(task)

Все задачи здесь выполняются последовательно, и ни одна не начинается, пока не завершится предыдущая. Подходит для простых и не критичных по времени систем, но явно не для высоконагруженных приложений.

А вот так выглядит параллельная обработка:

from concurrent.futures import ThreadPoolExecutor

def process_tasks_in_parallel(tasks):
    with ThreadPoolExecutor() as executor:
        results = executor.map(perform_task, tasks)

С помощью ThreadPoolExecutor мы можем одновременно запускать обработку задач. Это позволяет распределить нагрузку и ускорить выполнение.

---

Где применяется параллельная обработка?

1. Обработка изображений: скажем, у вас есть сервис, который позволяет пользователям загружать фотографии, а вы применяете фильтры или ваши магические алгоритмы обработки. Если обрабатывать каждую фотографию последовательно, пользователям придётся ждать, пока каждый снимок обработается.
2. Обработка сообщений: в системах реального времени (например, чатах или игровых серверах) сообщения должны обрабатываться максимально быстро, иначе пользователи будут недовольны.
3. Аналитика больших данных: когда ваш босс хочет отчёт по всем продажам за последние пять лет прямо сейчас (и это миллиард записей), понятно, что последовательный подход — путь к увольнению. Параллельная обработка — спасение.

---

Рыбы, птицы и RabbitMQ: роль брокеров в параллельности

Чтобы организовать параллельную обработку задач, вам нужен инструмент, который будет распределять задачи между рабочими процессами (воркерами). Здесь на сцену выходит RabbitMQ.

RabbitMQ — это система очередей сообщений, которая позволяет распределить задачи по нескольким воркерам. Параллельная обработка в связке RabbitMQ + Celery выглядит примерно так:

1. Задача помещается в очередь RabbitMQ.
2. Один из воркеров забирает задачу из очереди.
3. Воркеры работают параллельно, обрабатывая разные задачи.

А почему бы не просто всё положить в одну очередь? Ну, представьте, что у вас есть только один кассир (воркер), который пытается обслужить очередь из тысячи человек. Это слишком медленно.

---

Разница между параллельной и асинхронной обработкой

Да-да, я слышу ваши мысли: "А разве асинхронная обработка — это не параллельная?". Близко, но немного не так.

• Асинхронная обработка позволяет выполнять другие задачи, пока текущая задача "ждёт" (например, сеть или диск).
• Параллельная обработка означает выполнение нескольких задач одновременно, используя несколько потоков или процессов.

Пример асинхронного кода в Python:

import asyncio

async def perform_task(task):
    print(f"Starting {task}")
    await asyncio.sleep(2)  # Имитируем долгую операцию
    print(f"Completed {task}")

async def main():
    await asyncio.gather(perform_task("Task 1"), perform_task("Task 2"))

asyncio.run(main())

Этот код умеет "ждать" завершения одной задачи, пока выполняется другая. Однако если вы хотите реальной параллельности, вам нужны либо процессы (multiprocessing), либо потоки (threading).

---

Как начать параллельную обработку задач?

Параллельную обработку лучше всего доверить инструментам, уже зарекомендовавшим себя. Например:

• Celery: это библиотека для управления задачами, которая поддерживает параллельную обработку прямо "из коробки".
• RabbitMQ: используется как брокер сообщений для распределения задач.

Когда мы задействуем связку RabbitMQ + Celery, рабочий процесс выглядит так:

1. Задачи отправляются в Celery.
2. Celery передаёт их RabbitMQ для распределения.
3. Несколько воркеров (workers) обрабатывают задачи параллельно.

---

Практический пример

Давайте рассмотрим, как настроить параллельную обработку на простом проекте с использованием Celery.

Сначала установим Celery и RabbitMQ:

pip install celery

Теперь настройте RabbitMQ. Если у вас Docker, это можно сделать с помощью команды:

docker run -d --hostname my-rabbit --name some-rabbit -p 5672:5672 -p 15672:15672 rabbitmq:management

Конфигурация Celery

Создайте файл celery_app.py:

from celery import Celery

# Конфигурация Celery
celery_app = Celery('tasks', backend='rpc://', broker='pyamqp://guest@localhost//')

@celery_app.task
def add(x, y):
    return x + y

Теперь запустим Celery воркеров:

celery -A celery_app worker --loglevel=info

Отправим задачу в очередь. Выполним задачу, которая рассчитает сумму:

from celery_app import add

result = add.delay(4, 6)  # Отправляем задачу в очередь
print(result.get())       # Ожидаем результат

Воркеры выполнят задачу параллельно, если вы запустите несколько экземпляров воркеров.

---

Типичные сложности

Иногда кажется, что всё идет хорошо, но потом... бэм! Проблемы:

1. "Очередь заполнилась": если воркеры не успевают обрабатывать задачи, очередь начнёт раздуваться. Это сигнал о том, что нужно либо добавить воркеры, либо оптимизировать задачи.
2. Пересечение задач: иногда одна задача может зависеть от другой. Отслеживайте зависимости, чтобы избежать конфликтов.
3. Перерасход ресурсов: слишком много воркеров могут завалить сервер. Найдите баланс между количеством воркеров и доступными ресурсами.

---

Мы только начали раскручивать маховик параллельной обработки. В следующих лекциях вы узнаете, как настроить Celery для эффективной обработки задач и оптимизировать работу RabbitMQ для больших нагрузок. Вперёд к обработке задач со скоростью света!