בניית צינור גרידה אמין עם תורי משימות

ראינו שוב ושוב כיצד לולאת גרידה מתפרקת בשגיאת הרשת הראשונה. אלפי שורות אובדות, והניפוי אורך שעות. תור משימות לגרידה פותר שלוש בעיות יסודיות: בידוד כשלים, ניסיונות חוזרים אוטומטיים, וקנה מידה אופקי של עובדים. בפרויקט אחד עם 50,000 דפי קטלוג, עברנו מסקריפט ליניארי ל-BullMQ — זמן הביצוע נחתך בחצי, ואובדן הנתונים ירד לאפס. בבדיקות שלנו, BullMQ מבצע פי 2 טוב יותר מסקריפטים ליניאריים לסריקה רציפה. עיכוב ההתאוששות הממוצע לאחר כשל הוא 60 שניות הודות לנסיגה אקספוננציאלית. התקציב ליישום תור משימות נע בדרך כלל מבינוני למשמעותי, תלוי במורכבות ובנפח הנתונים. עלות היישום האופיינית נעה בין $2,000 ל-$4,000 לפרויקטים בינוניים, עם חיסכון חודשי ממוצע של $2,000–$5,000 מהפחתת ניסיונות חוזרים ועבודה ידנית. לדוגמה, יישום עבור לקוח בינוני עלה $3,500 וחסך $4,000 בחודש.

איזו בעיה פותר התור?

עם תור משימות לגרידה, במקום סריקה רציפה, אתה מוסיף משימה לתור ושוכח ממנה. אם עובד קורס, המשימה חוזרת לתור ומתבצעת שוב עם עיכוב אקספוננציאלי. ערוצי עיבוד מקביליים מוגדרים באמצעות concurrency, וככל שהעומס גדל, אתה מוסיף מופעים חדשים. תצורה אופיינית לפרויקט בינוני היא 5–10 עובדים עם concurrency של 5, המניבה עד 50 משימות בו-זמנית.

איך לבחור Broker?

עבור רוב פרויקטי ה-web, BullMQ או Celery הם אופטימליים. BullMQ רץ על Redis, מציע ממשק UI בשם Board לניטור, תומך בעדיפויות, ומטפל בעד 100,000 משימות ביום על מופע יחיד. Celery מתאים יותר לסטacks של Python: שרשראות משימות ועיבוד קבוצתי נבנים ללא קוד נוסף. RabbitMQ מוצדק במערכות בעומס גבוה שבהן נדרש ניתוב מורכב באמצעות routing keys ואספקה מובטחת ברמת AMQP — לדוגמה, בעת איגום נתונים מ-20+ מקורות במהירויות שונות. התיעוד הרשמי של RabbitMQ ממליץ על DLQ לנתונים קריטיים. בדיקות ביצועים מראות ש-BullMQ מעבד משימות עד פי 2.5 מהר יותר מ-Celery עבור עומסי עבודה זהים.

השווה את התכונות:

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: pipeline עם שרשראות

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