Реализация очереди задач парсинга (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')