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

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

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

Вместо последовательного обхода вы ставите задачу в очередь и забываете. Если воркер упал, задача возвращается в очередь и повторяется по экспоненциальной задержке. Потоки параллельной обработки настраиваются через 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' } } 

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')