Scraping Task Queue (Redis/RabbitMQ/BullMQ)

Implementing Scraping Task Queue (Redis/RabbitMQ/BullMQ)

Scraping in a loop — naive solution that breaks on first network error. Task queue solves three problems at once: failure isolation, retry logic, and horizontal scaling.

Broker Selection

For most web projects BullMQ or Celery — optimal choice. RabbitMQ justified when need delivery guarantees at AMQP level and complex routing between services.

BullMQ: Basic Setup

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

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

// Create queue
export const scrapeQueue = new Queue('scraping', {
  connection,
  defaultJobOptions: {
    attempts: 3,
    backoff: { type: 'exponential', delay: 60_000 },
    removeOnComplete: { count: 500 },
    removeOnFail: { count: 200 },
  },
});

// Add task
await scrapeQueue.add('scrape-url', {
  url: 'https://example.com/catalog?page=5',
  siteId: 42,
  depth: 1,
}, { priority: 1 });

// Worker
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: Pipeline with Task Chains

Celery allows building chains: first scrape list, then for each element launch detailed scrape.

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)

# Start pipeline
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))

Dead Letter Queue

Tasks that exhausted all retries go to DLQ. Not just trash — it's queue for manual analysis and reprocessing.

# RabbitMQ: DLQ setup via queue arguments
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 hour
    }
)
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')

Tasks from DLQ can be re-sent to main queue after fixing the cause — via Admin UI or script.

Queue Monitoring

BullMQ Board (UI for BullMQ) or Flower (for Celery) provide visual representation of queue state. Key metrics to track:

• Queue depth (waiting jobs)
• Processing speed (jobs/sec)
• Error percentage by task type
• Execution time (p50, p95, p99)

These metrics exported to Prometheus via /metrics endpoint and visualized in Grafana.

Implementation Timeline

Queue on BullMQ or Celery with retries and basic monitoring — 3–4 business days. RabbitMQ integration, DLQ, Prometheus metrics, task management UI — additional 2–3 days.