Какие существуют способы организации фоновых задач в Python без использования Celery

«Какие существуют способы организации фоновых задач в Python без использования Celery» — вопрос из категории Архитектура, который задают на 10% собеседований Python Разработчик. Ниже — развёрнутый ответ с разбором ключевых моментов.

Ответ

Для выполнения фоновых задач в Python без Celery можно использовать как встроенные модули, так и легковесные библиотеки. Выбор зависит от типа задачи и сложности инфраструктуры.

1. Встроенные модули

  • threading: Идеально для I/O-bound задач (например, сетевые запросы, работа с файлами), где основной процесс ожидает завершения внешних операций. Не подходит для CPU-bound задач из-за Global Interpreter Lock (GIL).

    from threading import Thread
    import time
    
    def slow_network_request(data):
        print(f"Starting request with {data}...")
        time.sleep(5)
        print("Request finished.")
    
    # Задача выполняется в фоновом потоке, не блокируя основной
    thread = Thread(target=slow_network_request, args=("some_data",))
    thread.start()
    print("Основной поток продолжает работу.")
  • multiprocessing: Подходит для CPU-bound задач (сложные вычисления, обработка данных), так как позволяет обойти ограничения GIL, запуская задачи в отдельных процессах. Более ресурсоемкий, чем threading.

    from multiprocessing import Process
    
    def cpu_intensive_task():
        # Имитация тяжелых вычислений
        sum(i * i for i in range(10**7))
        print("CPU-bound task finished.")
    
    process = Process(target=cpu_intensive_task)
    process.start()
    print("Основной процесс не ждет завершения вычислений.")

2. Легковесные библиотеки-очереди

Если требуется персистентность, очереди задач и повторные попытки, но Celery кажется избыточным, стоит рассмотреть:

  • RQ (Redis Queue): Простая в настройке библиотека, использующая Redis в качестве брокера сообщений. Отлично подходит для большинства веб-приложений.

    # app.py
    from redis import Redis
    from rq import Queue
    from tasks import count_words
    
    q = Queue(connection=Redis())
    # Асинхронный вызов задачи
    job = q.enqueue(count_words, 'http://example.com')
  • Huey: Легковесная альтернатива с поддержкой периодических задач, отложенного выполнения и хранения результатов. Может использовать Redis, SQLite или работать в памяти.