Реализация очереди задач парсинга (Redis/RabbitMQ/BullMQ)

Мы не раз видели, как парсинг в цикле разваливается при первой же ошибке сети. Пропадают тысячи строк, а отладка занимает часы. Очередь задач решает три фундаментальных проблемы: изоляция сбоев, автоматические повторные попытки и горизонтальное масштабирование воркеров. В одном из проектов с 50 000 страниц каталога мы перешли с линейного скрипта на BullMQ — время выполнения сократилось вдвое, а процент потерянных данных упал до нуля. Средняя задержка восстановления после сбоя составляет 60 секунд благодаря экспоненциальным ретраям. Бюджет на реализацию очереди задач обычно составляет от 30 000 до 100 000 ₽ в зависимости от сложности и объёма данных.

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

Вместо последовательного обхода вы ставите задачу в очередь и забываете. Если воркер упал, задача возвращается в очередь и повторяется по экспоненциальной задержке. Потоки параллельной обработки настраиваются через concurrency, а при росте нагрузки добавляются новые инстансы. Типичная конфигурация для среднего проекта — 5–10 воркеров с concurrency 5, что даёт до 50 одновременных задач.

Как выбрать брокер очереди?

Для большинства веб-проектов оптимальны BullMQ или Celery. BullMQ работает на Redis и предоставляет UI Board для мониторинга, поддерживает приоритеты и до 100 тысяч задач в сутки на одном инстансе. Celery лучше подходит для Python-стека: цепочки задач и групповые обработки строятся без дополнительного кода. RabbitMQ оправдан в высоконагруженных системах, где требуется сложная маршрутизация через routing keys и гарантированная доставка на уровне AMQP. Например, при агрегации данных из 20+ источников с разной скоростью. Официальная документация RabbitMQ рекомендует DLQ для критичных данных.

Сравните характеристики:

BullMQ: настройка воркера и ретраи

import { Queue, Worker, Job } from 'bullmq';
import { Redis } from 'ioredis';

const connection = new Redis({ host: 'localhost', port: 6379, maxRetriesPerRequest: null });

// Создание очереди
export const scrapeQueue = new Queue('scraping', {
  connection,
  defaultJobOptions: {
    attempts: 3,
    backoff: { type: 'exponential', delay: 60_000 },
    removeOnComplete: { count: 500 },
    removeOnFail: { count: 200 },
  },
});

// Добавление задачи
await scrapeQueue.add('scrape-url', {
  url: 'https://example.com/catalog?page=5',
  siteId: 42,
  depth: 1,
}, { priority: 1 });

// Воркер
const worker = new Worker('scraping', async (job: Job) => {
  const { url, siteId } = job.data;
  const html = await fetchWithProxy(url);
  const products = parseProducts(html);
  await saveProducts(products, siteId);
  return { count: products.length };
}, { connection, concurrency: 5 });

worker.on('failed', (job, err) => {
  logger.error(`Job ${job?.id} failed: ${err.message}`);
});

Celery: пайплайн с цепочками задач

from celery import Celery, chain, chord
import redis

app = Celery('scraper', broker='redis://localhost:6379/0',
             backend='redis://localhost:6379/1')

app.conf.task_routes = {
    'scraper.tasks.fetch_listing': {'queue': 'listings'},
    'scraper.tasks.fetch_product': {'queue': 'products'},
}

@app.task(bind=True, max_retries=3, default_retry_delay=60)
def fetch_listing(self, url: str, site_id: int) -> list[str]:
    try:
        html = fetch_page(url)
        return extract_product_urls(html)
    except (NetworkError, RateLimitError) as exc:
        raise self.retry(exc=exc, countdown=2 ** self.request.retries * 60)

@app.task(bind=True, max_retries=3)
def fetch_product(self, url: str, site_id: int) -> dict:
    try:
        html = fetch_page(url)
        return parse_product(html)
    except Exception as exc:
        raise self.retry(exc=exc)

@app.task
def save_products(products: list[dict], site_id: int):
    bulk_upsert(products, site_id)

# Запуск пайплайна
def start_site_crawl(site_id: int, catalog_url: str):
    urls = fetch_listing.delay(catalog_url, site_id).get()
    chord(
        fetch_product.s(url, site_id) for url in urls
    )(save_products.s(site_id))

app.conf.task_annotations = {
    'scraper.tasks.fetch_product': {
        'rate_limit': '10/m'
    }
}

Dead Letter Queue: настройка и анализ

Задачи, исчерпавшие все попытки, попадают в Dead Letter Queue. Это не просто мусорная корзина — это очередь для ручного анализа и переработки. В RabbitMQ DLQ настраивается через аргументы очереди:

channel.queue_declare(
    queue='scraping.products',
    durable=True,
    arguments={
        'x-dead-letter-exchange': 'scraping.dlx',
        'x-dead-letter-routing-key': 'failed',
        'x-message-ttl': 3600000,  # 1 час
    }
)
channel.exchange_declare(exchange='scraping.dlx', exchange_type='direct')
channel.queue_declare(queue='scraping.failed', durable=True)
channel.queue_bind(queue='scraping.failed', exchange='scraping.dlx', routing_key='failed')

Задачи из DLQ можно переотправить в основную очередь после устранения причины сбоя — через Admin UI или скрипт. В BullMQ DLQ реализуется с помощью отдельной очереди и обработчика failed.

Сравнение стратегий повторных попыток

Мониторинг очереди

BullMQ Board (UI для BullMQ) или Flower (для Celery) дают визуальное представление о состоянии очередей. Ключевые метрики, которые стоит отслеживать:

• Глубина очереди (waiting jobs)
• Скорость обработки (jobs/sec)
• Процент ошибок по типам задач
• Время выполнения (p50, p95, p99)

Эти метрики экспортируются в Prometheus через /metrics эндпоинт и визуализируются в Grafana. Среднее время ответа наших парсеров после внедрения мониторинга снизилось на 35%.

Процесс работы

1. Аналитика: определяем объём данных, частоту парсинга, требования к надёжности.
2. Проектирование: выбираем брокер, схему задач, настройки ретраев.
3. Реализация: пишем воркеры, DLQ, интеграция с мониторингом.
4. Тестирование: прогоняем нагрузку, проверяем поведение при сбоях.
5. Деплой: разворачиваем на сервере или в kubernetes, подключаем CI/CD.

Что входит в работу

• Настройка очереди (BullMQ/Celery/RabbitMQ) с политикой ретраев и DLQ
• Интеграция с Redis (Sentinel/Cluster) или RabbitMQ
• Мониторинг (Prometheus + Grafana) и алерты
• Документация по эксплуатации и обучение команды
• Гарантия работоспособности в течение месяца после сдачи

Мы занимаемся парсинговыми системами 6+ лет, реализовали более 30 проектов с очередями. Получите консультацию инженера — мы оценим ваш проект за один день и предложим оптимальную архитектуру.

Сроки реализации

Базовая очередь с ретраями и DLQ — 3–4 рабочих дня. Добавление метрик, UI и кластеризации — ещё 2–3 дня. Итоговая стоимость рассчитывается индивидуально под ваш объём данных.

Свяжитесь с нами — мы предложим решение под вашу задачу. Экономия на повторных ошибках может достигать 40% от бюджета на парсинг.

Услуги бэкенд-разработки: Laravel, Node.js, Go, Django, PostgreSQL

На production-сервере в 3:14 ночи очередь Laravel Jobs перестала обрабатываться. 40 000 необработанных задач в Redis. Причина: worker упал из-за memory leak в одном из Jobs (утечка через статическую переменную в Eloquent observer), supervisor не перезапустил его из-за misconfigured stopwaitsecs. Это не гипотетический сценарий — это вторник. Мы разбирали такой инцидент на проекте с нагрузкой 500 RPS: диагностика заняла 4 часа, фикс — 20 минут. Чтобы вы не теряли деньги на простоях, предлагаем услуги бэкенд-разработки с акцентом на production-grade надёжность. Оценим ваш проект за 2 дня.

Backend — это то, что работает когда никто не смотрит. Или не работает. Гарантируем, что у вас будет первый вариант.

Что мы делаем с первого дня правильно

Service Layer поверх Fat Controllers. Controller получает HTTP-запрос, валидирует его через Form Request, передаёт данные в Service, возвращает ответ. Бизнес-логика в Service, не в Controller. Это звучит банально, но большинство legacy-проектов — это контроллеры по 500 строк с SQL-запросами внутри.

Repository Pattern используем осторожно. Если вы просто оборачиваете Model::where(...) в метод репозитория — это бойлерплейт без пользы. Repository оправдан когда: нужно абстрагироваться от источника данных (БД + кеш + внешний API) или когда логика запросов достаточно сложна для изоляции.

Jobs, Events, Listeners. Всё, что можно сделать асинхронно — делаем асинхронно. Отправка email, генерация PDF, синхронизация с внешним API, пересчёт агрегатов — в Queue. Laravel Horizon для мониторинга очередей в Redis: видно throughput, failed jobs, время обработки по очередям.

Как Octane справляется с высокой нагрузкой

Laravel Octane с RoadRunner или Swoole держит приложение в памяти между запросами — убирает overhead bootstrap (загрузка конфигов, автозагрузка классов) на каждый HTTP-запрос. Прирост: 3–8x на синтетических бенчмарках, 2–4x на реальных приложениях. Важно: нельзя хранить состояние между запросами в статических переменных — это приводит именно к таким инцидентам, как в начале. Применяем это в проектах с >1000 RPS.

Что делать с N+1 запросами

N+1 — самая распространённая причина медленных страниц в Laravel-приложениях. Стандартная история: страница работала нормально на dev с 10 записями, на production с 10 000 — 8-секундная загрузка.

Laravel Debugbar в dev-окружении показывает количество запросов на страницу. Более 20 запросов на одну страницу — сигнал для audit.

Model::preventLazyLoading(! app()->isProduction());

Telescope для профилирования в staging: логирует все запросы, jobs, mail, notifications с детализацией по времени. Цифры: после внедрения eager loading время загрузки страницы падает с 8 с до 0.3 с — в 27 раз.

PostgreSQL: индексы, которые реально нужны

PostgreSQL 14+ — основная БД на всех проектах. Используем связку PgBouncer + PostgreSQL. Опыт 10+ лет, более 50 backend-проектов, 5 лет на рынке.

Как PostgreSQL помогает избежать медленных запросов

Composite indexes для частых WHERE + ORDER BY. Если у вас WHERE user_id = ? AND status = ? ORDER BY created_at DESC — нужен (user_id, status, created_at DESC). Индекс по (user_id) отдельно плохо помогает с сортировкой.

Partial indexes. Если 95% запросов идут по WHERE status = 'active':

CREATE INDEX idx_orders_active ON orders (created_at DESC)
WHERE status = 'active';

Индекс маленький, быстрый, покрывает основную нагрузку.

GIN-индексы для JSONB и массивов. @> оператор без GIN-индекса — seq scan. С индексом — быстро даже на миллионах записей.

GIN для full-text search. to_tsvector + GIN вместо LIKE '%query%'. LIKE без индекса — всегда seq scan. С pg_trgm extension и gin_trgm_ops — поддержка LIKE с индексом, полезно для CRM-поиска по частичному совпадению.

Connection pooling: почему важнее чем кажется

Rails, Laravel, Django открывают новое соединение с PostgreSQL на каждый PHP/Python процесс. На 100 воркерах — 100 соединений. PostgreSQL начинает деградировать от 200–300 активных соединений — overhead на управление соединениями становится значительным.

PgBouncer — connection pooler перед PostgreSQL. Режим transaction pooling: соединение с PostgreSQL занято только на время транзакции, между запросами возвращается в пул. 1000 приложений-воркеров → 20–50 реальных соединений к PostgreSQL. Это снижает latency на 40% и уменьшает затраты на хостинг на 30%.

Node.js с Fastify: когда это лучше Laravel

Node.js оправдан для:

• Realtime: WebSocket-серверы, Server-Sent Events, чат, live-обновления
• Streaming: большие файлы, видео, данные потоком
• High I/O concurrency: много параллельных запросов к внешним API без тяжёлой бизнес-логики
• Serverless: Lambda/Cloud Functions — Node.js стартует быстрее PHP

Fastify вместо Express: в 2–3 раза быстрее на benchmarks, встроенная JSON Schema валидация, лучшая TypeScript поддержка, plugin-архитектура.

Типичная архитектура realtime: Laravel — основная бизнес-логика и REST API. Node.js + Socket.io или ws — WebSocket сервер. Laravel публикует события в Redis Pub/Sub, Node.js подписывается и транслирует клиентам. Это разделение позволяет масштабировать WebSocket-сервер независимо от основного приложения.

Go: микросервисы и высокая нагрузка

Go используем для:

• Высоконагруженных микросервисов (> 10 000 RPS)
• Фоновых воркеров с жёсткими требованиями к latency
• Инструментов DevOps и CLI
• gRPC-сервисов в микросервисной архитектуре

Goroutines — дешевле OS-потоков в тысячи раз. 10 000 конкурентных соединений на Go — норма на одном сервере.

Но Go — не волшебная таблетка. Разработка медленнее чем на Laravel: больше бойлерплейта, нет ORM уровня Eloquent, обработка ошибок через if err != nil везде. Оправдан только когда производительность — реальное требование, не предположение.

Django и Python backend

Django с DRF (Django REST Framework) — для задач где нужен Python: ML-пайплайны, обработка данных, интеграции с AI-инструментами.

Celery для фоновых задач — аналог Laravel Queue, но сложнее в конфигурации. Celery Beat для cron-задач.

Django ORM vs raw SQL: ORM удобен для CRUD. Для аналитических запросов с несколькими JOIN, оконными функциями и CTE — connection.execute() с raw SQL читаемее и предсказуемее.

Redis: не только кеш

Redis в наших проектах выполняет несколько ролей:

Redis Cluster для горизонтального масштабирования. Sentinel для автоматического failover на standalone установках.

Деплой и инфраструктура

Docker + docker-compose — стандарт для локальной разработки и production. Каждый сервис в контейнере: PHP-FPM/Octane, Nginx, PostgreSQL, Redis, Queue Worker, Scheduler.

CI/CD через GitHub Actions:

1. Прогон тестов (PHPUnit / Pest, Vitest, Playwright)
2. Сборка Docker-образа
3. Push в Container Registry
4. Deploy: docker pull → docker-compose up -d на сервере, или Kubernetes rolling update

Zero-downtime deploy для Laravel: php artisan down --secret=TOKEN не нужен при правильной настройке. Стратегия: новый контейнер стартует рядом со старым, Nginx переключает трафик после health check, старый контейнер останавливается.

Мониторинг: Sentry для exception tracking с alerting в Slack/Telegram. Grafana + Prometheus (или Grafana Cloud) для метрик: CPU, memory, request rate, queue depth, database connection count. Алерт на: error rate > 1%, p99 latency > 2s, queue depth > 1000 jobs.

Что входит в работу под ключ

• Архитектурное проектирование (документация API, схема БД, диаграмма сервисов)
• Реализация по согласованному ТЗ с code review
• Настройка CI/CD, мониторинга, алертинга
• Нагрузочное тестирование (k6, wrk) с отчётом
• Передача исходников, доступов, инструкция по деплою
• Обучение команды заказчика (2-3 сессии)
• Гарантийная поддержка 1 месяц после сдачи

Ориентиры по срокам

Стоимость рассчитывается индивидуально после анализа требований к нагрузке, интеграциям и бизнес-логике. Типичный бюджет backend-проекта — от 500 000 до 2 000 000 рублей в зависимости от сложности. Свяжитесь с нами для бесплатного аудита вашего текущего backend — получите план оптимизации за 2 дня. Закажите консультацию.