Настройка RQ worker для redis очередь без импорта на сервере

Да, в redis очередь с помощью RQ (Redis Queue) можно полностью разделить backend-сервер и rq worker, чтобы сервер не импортировал код worker’а и не тянул его зависимости. Просто используйте enqueue по строковому пути функции — q.enqueue('my_module.my_func', arg1, arg2), — сервер запишет только имя в Redis, а worker сам разберётся с импортом в своём окружении через PYTHONPATH. Это стандартный подход для очередей задач python, где backend остаётся лёгким, а тяжёлые задачи (ML-модели, обработка видео) висят на отдельном сервере.

Содержание

• Введение в redis очередь и RQ
• Проблема импорта на сервере в python rq
• Enqueue по строковому пути без зависимостей
• Настройка rq worker с PYTHONPATH
• Обработка ошибок импорта в очередях задач python
• Примеры кода для backend и worker
• Расширенные возможности RQ
• Источники
• Заключение

Введение в redis очередь и RQ

Представьте: ваш backend генерирует тонну задач — отправка email, обработка изображений, расчёты. Redis очередь как раз для этого. RQ — это простая библиотека на Python, которая использует Redis как брокер сообщений. Почему она крутая? Легко масштабируется, не требует сложного Celery, и главное — позволяет очереди задач python работать асинхронно.

RQ берёт задачу (функцию + аргументы), сериализует её в JSON и кидает в Redis. Worker на другом конце подхватывает, выполняет и кладёт результат обратно. Но вот засада из вашего примера: from my_module import count_words_at_url. Сервер импортирует модуль worker’а? Зависимости полетят! А если worker жрёт TensorFlow или FFmpeg? Backend взорвётся.

Хорошая новость: RQ поддерживает строковые пути. Сервер не трогает код — только строка вроде 'workers.heavy_task.process_image'. Worker импортирует сам. Это разделение — основа микросервисов.

Проблема импорта на сервере в python rq

В классическом примере из документации RQ сервер делает from my_module import func перед enqueue. Почему так? RQ сериализует объект функции, а не код. Чтобы получить этот объект, нужен импорт.

Но представьте ваш сценарий: backend на Flask/FastAPI, чистый, без лишнего. Worker — монстр с libs вроде opencv-python (сотни МБ). Импорт на сервере? Docker-образ раздуется, деплой сломается. Плюс security: зачем серверу знать внутренности worker’а?

В issues на GitHub RQ народ жалуется: “ImportError: No module named api.scripts”. Относительные импорты, разные PYTHONPATH — классика. Решение? Строковый enqueue. Сервер пишет 'module.func', Redis хранит строку. Worker парсит и делает getattr(importlib, ...).

Это не хак — фича RQ с версии 0.4+. Работает в контейнерах, Kubernetes. Главное — унифицировать имена модулей.

Enqueue по строковому пути без зависимостей

Вот как это работает. На backend’е:

import redis
from rq import Queue

redis_conn = redis.Redis(host='localhost', port=6379)
q = Queue(connection=redis_conn)

# Вместо импорта: строка!
job = q.enqueue('workers.tasks.process_image', '/path/to/image.jpg', timeout=300)
print(job.id) # Задача в **redis очередь**

Сервер не знает о workers.tasks. Он сериализует строку 'workers.tasks.process_image' + args в JSON. Redis — dumb storage, просто ключи-значения. Никаких pickle-зависимостей.

RQ использует cloudpickle или dill для сложных объектов, но для строк — чистый JSON. Доки подтверждают: “You can also enqueue functions by their string path”. Идеально для redis queue python.

Что если args сложные? Сериализуйте в dict или JSON. Worker распакует. Масштаб? Несколько очередей: high, low — q = Queue('high').

Плюс: сервер может enqueue из любого языка, если пишет правильный JSON в Redis. Но stick to Python для простоты.

Настройка rq worker с PYTHONPATH

Worker живёт отдельно. Установите RQ: pip install rq. Но ключ — PYTHONPATH. Worker должен видеть модули по тому же пути, что в строке.

Запуск:

# В директории worker'а
export PYTHONPATH=/path/to/workers:$PYTHONPATH
rq worker --url redis://localhost:6379 --path /path/to/workers

Или в Docker:

ENV PYTHONPATH=/app/workers
CMD rq worker --url redis://redis:6379

Worker берёт задачу из redis очередь, парсит 'module.func', делает:

import importlib
module = importlib.import_module('workers.tasks')
func = getattr(module, 'process_image')
func(*args)

Если PYTHONPATH не совпадает — boom, AttributeError. На Stack Overflow советуют: добавьте --path или env. Тестируйте локально.

Несколько worker’ов? rq worker high low default. Автоскейлинг через supervisor или Kubernetes.

Обработка ошибок импорта в очередях задач python

Ошибки — норма. Worker не нашёл модуль? No module named 'workers'. Решение: логируйте, ретрайте.

В RQ:

@job
def process_image(path):
 # heavy code
 pass

Но с строкой ошибок больше. Ловите:

# В worker скрипте
import traceback
try:
 # RQ magic
except Exception as e:
 print(traceback.format_exc())
 raise # RQ отметит failed

Ретраи: q.enqueue(..., retry=Retry(max=3, delay=60)). Failed jobs в FailedQueue.

Мониторинг: rq-dashboard — веб-UI для очередей задач python. Установите pip install rq-dashboard, запустите rq-dashboard.

В проде: Sentry или ELK для логов. Redis >=3.0 обязателен — старые версии глючат с очередями.

А если функция меняется? Worker перезапустите — RQ подхватит новую версию при следующем job.

Примеры кода для backend и worker

Backend (Flask endpoint):

from flask import Flask
from rq import Queue
from redis import Redis

app = Flask(__name__)
redis_conn = Redis()
q = Queue(connection=redis_conn)

@app.route('/process/<path>')
def enqueue(path):
 job = q.enqueue('image_processor.resize', path, width=800)
 return f'Job ID: {job.id}'

Worker (отдельный скрипт или сервис):

# workers/image_processor.py
def resize(path, width=800):
 # Используйте Pillow или что-то тяжёлое
 from PIL import Image
 img = Image.open(path)
 img.resize((width, width)).save(path)
 return 'Done'

Запуск: PYTHONPATH=. rq worker. Boom — сервер не знает о PIL.

Для стек и очередь python задачи: RQ — LIFO по умолчанию? Нет, FIFO. Стек через Queue(as_fifo=False).

Тестируйте: rq info покажет jobs.

Расширенные возможности RQ

Не всё просто. Планировщик: q.enqueue_in(timedelta(hours=1), 'module.func'). Или rq-scheduler.

Асинхронные результаты: job.result (блокирующий) или fetch_job(job_id).result.

Множественные Redis: sentinel, cluster.

Безопасность: ACL в Redis, аутентификация workers.

Масштаб: 1000+ workers? RQ держит, но мониторьте память Redis (jobs ~1KB каждая).

Альтернативы? Celery — тяжелее, но с broker’ами. RQ — для Python-only.

В 2026 Redis 8+ добавит Streams — RQ мигрирует?

Источники

1. RQ Documentation — Официальная документация по настройке очередей и jobs: https://python-rq.org/docs/
2. RQ Jobs Guide — Детали сериализации задач и строковых путей: https://python-rq.org/docs/jobs/
3. RQ GitHub Repository — Основной репозиторий с примерами и обсуждениями: https://github.com/rq/rq
4. RQ Issue #759 — Анализ ошибок импорта в worker: https://github.com/rq/rq/issues/759
5. Stack Overflow RQ String Ref — Решения проблем с PYTHONPATH и атрибутами: https://stackoverflow.com/questions/77885291/rq-is-unable-to-find-the-task-using-string-like-function-ref

Заключение

Настроить rq worker для redis очередь без импорта на backend — просто и эффективно: строковый enqueue + правильный PYTHONPATH. Сервер остаётся чистым, worker берёт всю нагрузку. Начните с простого примера, добавьте мониторинг — и ваши очереди задач python полетят. Если ошибки — чекните пути и логи. Это масштабируемо, безопасно и без головной боли.