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

Ми не раз бачили, як парсинг у циклі розвалюється при першій же помилці мережі. Пропадають тисячі рядків, а налагодження займає години. Черга завдань вирішує три фундаментальні проблеми: ізоляція збоїв, автоматичні повторні спроби та горизонтальне масштабування воркерів. В одному з проектів з 50 000 сторінок каталогу ми перейшли з лінійного скрипта на BullMQ — час виконання скоротився вдвічі, а відсоток втрачених даних впав до нуля. Середня затримка відновлення після збою становить 60 секунд завдяки експоненціальним ретраям.

Яку проблему вирішує черга?

Замість послідовного обходу ви ставите завдання в чергу і забуваєте. Якщо воркер впав, завдання повертається в чергу і повторюється з експоненціальною затримкою. Потоки паралельної обробки налаштовуються через 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')