اسکریپینگ موازی با Requests و ThreadPoolExecutor

مقدمه

در این مقاله مطرح می‌کنیم چطور با استفاده از کتابخانهٔ requests و اجرای موازی با ThreadPoolExecutor سرعت اسکریپینگ را افزایش دهید. فرض می‌کنیم خواننده توسعه‌دهندهٔ پایتون در سطح متوسط است و می‌خواهیم در پایان: نمونهٔ کد عملی برای اجرای موازی، روش‌های ایمن و پایدار برای مدیریت نشست‌ها، راهنمای استفاده از پروکسی (مثلاً ScrapeOps) و بهترین روش‌های خطایابی و افزایش پایداری را توضیح دهیم.

چرا اسکریپینگ موازی؟ مزایا و محدودیت‌ها

اجرای همزمان درخواست‌ها می‌تواند زمان کلی اسکِراپ را به‌طرز چشمگیری کاهش دهد، ولی همراه خود ریسک‌ها و محدودیت‌هایی دارد:

• مزایا:
  • افزایش throughput—بیشتر کردن تعداد درخواست‌های فعال در هر ثانیه.
  • بهتر استفاده شدن از زمان‌های انتظار شبکه (I/O bound).
• معایب / هشدارها:
  • افزایش احتمال بلاک شدن توسط سرور هدف یا شبکه (rate limit / IP ban).
  • مسائل thread-safety در استفادهٔ مشترک از منابع مثل requests.Session و ساختارهای دادهٔ اشتراکی.
  • چنانچه تعداد تردها خیلی بالا باشد، ممکن است منابع محلی (CPU، حافظه، فایل descriptor) یا سرویس پروکسی محدودیت ایجاد کنند.

روش پایه: ThreadPoolExecutor با Requests

ایدهٔ کلی: یک تابع استخراج می‌نویسیم که یک URL را می‌گیرد و دادهٔ مورد نیاز را برمی‌گرداند؛ سپس با استفاده از ThreadPoolExecutor چند کارگر (worker) می‌سازیم که از فهرست URLها بخوانند و تابع استخراج را اجرا کنند.

import requests
from bs4 import BeautifulSoup
import concurrent.futures

NUM_THREADS = 5
list_of_urls = [
    'http://quotes.toscrape.com/page/1/',
    'http://quotes.toscrape.com/page/2/',
    'http://quotes.toscrape.com/page/3/',
]

def scrape_page(url):
    """ورودی: url (رشته)
    خروجی: دیکشنری با داده‌های استخراج‌شده یا None در صورت خطا
    شرح: یک درخواست ساده GET ارسال می‌کند، وضعیت را بررسی و عنوان صفحه را استخراج می‌کند."""
    try:
        resp = requests.get(url, timeout=10)
        if resp.status_code == 200:
            soup = BeautifulSoup(resp.text, 'html.parser')
            title = soup.find('h1').get_text(strip=True) if soup.find('h1') else ''
            return {'url': url, 'title': title}
        else:
            return {'url': url, 'error': f'status_{resp.status_code}'}
    except Exception as e:
        return {'url': url, 'error': str(e)}

if __name__ == '__main__':
    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=NUM_THREADS) as executor:
        for res in executor.map(scrape_page, list_of_urls):
            results.append(res)
    print(results)

توضیح گام‌به‌گام:

1. تعریف list_of_urls که می‌خواهیم از آن‌ها اسکریپ کنیم.
2. تعریف scrape_page(url) که یک URL می‌گیرد، درخواست ارسال می‌کند، HTML را پارس می‌کند و دادهٔ موردنظر را بازمی‌گرداند یا خطا را گزارش می‌دهد.
3. ساخت ThreadPoolExecutor با پارامتر max_workers=NUM_THREADS و استفاده از executor.map برای تخصیص کارها به تردها.
4. جمع‌آوری نتایج به شکل فهرستی از دیکشنری‌ها تا مدیریت خطا در هر آیتم آسان شود.

بهبودها: استفادهٔ بهینه از نشست‌ها (Session)، Retry و timeout

برای افزایش کارایی و پایداری باید از قابلیت‌های زیر استفاده کنید:

• اتصال مجدد خودکار و retry کنترل‌شده.
• تنظیم timeout برای جلوگیری از تردهای قفل‌شده روی درخواست‌های آویزان.
• استفاده از requests.Session به‌صورت یک جلسهٔ اختصاصی برای هر ترد تا از connection pooling بهره‌مند شوید بدون برخورد با مشکلات هم‌زمانی.

import threading
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from bs4 import BeautifulSoup
import concurrent.futures

thread_local = threading.local()

NUM_THREADS = 8
list_of_urls = [
    'http://quotes.toscrape.com/page/1/',
    # ...
]

def create_session():
    session = requests.Session()
    retries = Retry(total=3, backoff_factor=0.5, status_forcelist=[429, 500, 502, 503, 504])
    adapter = HTTPAdapter(max_retries=retries, pool_maxsize=100)
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    session.headers.update({'User-Agent': 'my-scraper/1.0'})
    return session

def get_session():
    if not hasattr(thread_local, 'session'):
        thread_local.session = create_session()
    return thread_local.session

def scrape_page(url):
    s = get_session()
    try:
        resp = s.get(url, timeout=10)
        resp.raise_for_status()
        soup = BeautifulSoup(resp.text, 'html.parser')
        title = soup.find('h1').get_text(strip=True) if soup.find('h1') else ''
        return {'url': url, 'title': title}
    except Exception as e:
        return {'url': url, 'error': str(e)}

if __name__ == '__main__':
    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=NUM_THREADS) as executor:
        for r in executor.map(scrape_page, list_of_urls):
            results.append(r)
    print(results)

شرح نکات مهم کد بالا:

• thread_local: برای داشتن یک Session اختصاصی در هر ترد به‌کار می‌رود؛ این الگو از مزایای pooling بهره‌مند می‌کند بدون به‌وجود آوردن رقابت هم‌زمان روی یک شیء واحد.
• Retry: با urllib3.util.retry.Retry تنظیم می‌کنیم که در خطاهای موقت (مثلاً 502 یا 429) دوباره تلاش شود و با backoff_factor فُرکانس تلاش‌ها تدریجی شود.
• timeout و resp.raise_for_status(): برای جلوگیری از درخواست‌های معلق و مدیریت وضعیت‌های غیر 200.

مدیریت نتایج و ایمنی دادهٔ مشترک

به‌جای نوشتن مستقیم در یک لیست مشترک از داخل تردها، بهتر است نتایج را بازگردانده و از خروجی executor.map یا concurrent.futures.as_completed استفاده کنید تا ترتیب و خطاگیری ساده‌تر شود. اگر به‌هرحال باید ساختار مشترک را نوشت، از قفل (threading.Lock) یا ساختارهای thread-safe مثل queue.Queue استفاده کنید.

استفاده از پروکسی و ScrapeOps

وقتی از پروکسی‌ها استفاده می‌کنید، معمولاً باید تعداد تردها را متناسب با پلان پروکسی خود تنظیم کنید تا از concurrency مجاز خارج نشوید. نمونهٔ سادهٔ تبدیل URL به آدرس پروکسی (روش معمولی برای استفاده از APIهایی مانند ScrapeOps Proxy Aggregator):

from urllib.parse import urlencode

SCRAPEOPS_API_KEY = 'YOUR_API_KEY'

def get_scrapeops_url(url):
    payload = {'api_key': SCRAPEOPS_API_KEY, 'url': url}
    proxy_url = 'https://proxy.scrapeops.io/v1/?' + urlencode(payload)
    return proxy_url

# در scrape_page کافی است به‌جای url مستقیماً از get_scrapeops_url(url) استفاده کنید
# و درخواست‌ها را به آدرس پروکسی بزنید.

نکات عملی دربارهٔ پروکسی:

• هر پلان پروکسی محدودیت concurrency دارد؛ NUM_THREADS را مطابق پلان تنظیم کنید.
• استفاده از پروکسی می‌تواند خطاهای اضافی ایجاد کند—بنابراین retry و لاگینگ دقیق مهم است.
• اگر پروکسی API کلید می‌خواهد، کلیدها را در متغیرهای محیطی یا vault نگهداری کنید، نه در کد ثابت.

معیارها و انتخاب تعداد ترد

برای تعیین مقدار مناسب NUM_THREADS این موارد را اندازه‌گیری کنید:

1. تست با مقدارهای مختلف و اندازه‌گیری زمان کل و خطاها.
2. پایش نرخ خطاها و نرخ 429 (Too Many Requests).
3. اندازه‌گیری استفادهٔ منابع محلی (فایل‌دسکریپتورها، حافظه، CPU) و محدود کردن تردها در صورت سربار زیاد.

مقایسهٔ روش‌ها و جایگزین‌ها

چند گزینه برای اجرای موازی وجود دارد:

• ThreadPoolExecutor: مناسب برای I/O bound و ساده برای استفاده با requests.
• Multiprocessing: زمانی که پردازش CPU-intensive داریم؛ overhead بیشتر و مناسب برای موارد متفاوت.
• Async (مثل aiohttp): در مقیاس بزرگ و زمانی که می‌خواهید بیشترین کارایی I/O را بدست آورید، سریع‌تر و کارآمدتر است اما نیاز به بازنویسی کدها به سبک async دارد.

نکات امنیتی و اخلاقی

• همیشه robots.txt را بررسی کنید و در برابر سایت‌هایی که اجازهٔ scraping صریح نمی‌دهند محتاط باشید.
• هدر User-Agent و رفتار polite را رعایت کنید؛ نرخ درخواست‌ها را محدود کنید تا به سرور هدف آسیب نزنید.
• کلیدهای API و اعتبارنامه‌ها را امن نگهداری کنید و در لاگ‌ها افشا نکنید.

جمع‌بندی و توصیه‌های عملی

برای اسکریپینگ موازی با Requests این روند را پیشنهاد می‌کنم:

1. ابتدا نسخهٔ همزمان ساده را پیاده کنید تا فرایند استخراج درست کار کند.
2. جلسات (Session) را با الگوی thread-local یا مشابه به‌کار بگیرید تا از connection pooling بهره ببرید.
3. Retry، backoff، و timeout را تنظیم کنید و لاگینگ مناسبی برای خطاها داشته باشید.
4. اگر از پروکسی استفاده می‌کنید (مثلاً ScrapeOps)، NUM_THREADS را مطابق محدودیت‌های پلان تعیین کنید و تست‌های بار را انجام دهید.
5. برای مقیاس‌های بسیار بزرگ، بررسی کنید که آیا مهاجرت به مدل async (مثلاً aiohttp) یا ترکیب با صف‌بندی و توزیع کار (task queue) مناسب نیست.

این راهنما به شما امکان می‌دهد با دانش فنی مناسب اسکریپینگ موازی را ایمن، پایدار و مؤثر پیاده‌سازی کنید.