همزمانی ترکیبی پایتون: AsyncIO + Multiprocessing

توسعه‌دهندگان پایتون اغلب هنگام ساخت خطوط لوله داده با کارایی بالا با یک دوراهی کلاسیک روبرو می‌شوند: قفل تفسیرگر سراسری (GIL). برای سال‌ها، انتخاب به صورت دوگانه بوده است—استفاده از asyncio برای وظایف وابسته به ورودی/خروجی (I/O) یا multiprocessing برای عملیات وابسته به پردازنده (CPU). با این حال، وب‌اسکرپینگ مدرن اغلب به هر دو نیاز دارد. تصور کنید یک اسکرپر وجود دارد که باید همزمان منتظر پاسخ‌های شبکه (I/O) بماند و در عین حال ساختارهای سنگین JSON را تجزیه کند یا روی صفحات دریافت شده تشخیص تصویر را اجرا کند (CPU).

ترکیب AsyncIO و Multiprocessing یک الگوی ترکیبی قدرتمند ایجاد می‌کند. این رویکرد از کارایی ناهمگام برای تاخیر شبکه بهره می‌برد و همزمان از GIL برای محاسبات عبور می‌کند. در این راهنما، ما بررسی می‌کنیم که چگونه این سیستم را معماری کنیم، ارتباط بین فرآیندی (IPC) لازم را مدیریت کنیم و یک مجموعه کارآمد (Worker Pool) قوی پیاده‌سازی کنیم.

درک معماری

برای ساخت یک سیستم ترکیبی، ابتدا باید مرزهای هر ابزار را بشناسیم. asyncio بر پایه حلقه رویداد (Event-loop) است و برای هزاران اتصال شبکه همزمان عالی است. در مقابل، multiprocessing فرآیندهای جداگانه را ایجاد می‌کند که هر کدام دارای GIL خود هستند، که آن را برای تجزیه‌های سنگین وابسته به پردازنده ایده‌آل می‌کند.

معماری معمولاً شامل یک حلقه رویداد ناهمگام اصلی است که به عنوان «مدیر» عمل می‌کند. این مدیر درخواست‌ها را ایجاد کرده و آن‌ها را از طریق یک صف به فرآیندهای کارگر ارسال می‌کند. کارگرها که در فرآیندهای جداگانه اجرا می‌شوند، کارهای سنگین را انجام داده و نتایج را برمی‌گردانند. نکته کلیدی این است که ارتباط بین مدیر ناهمگام و مجموعه کارگر باید کارآمد باشد تا از ایجاد گلوگاه جلوگیری شود.

استراتژی پیاده‌سازی با Asyncio

ما از asyncio برای مدیریت چرخه حیات فرآیندهای کارگر و از multiprocessing.Queue برای انتقال داده‌ها استفاده خواهیم کرد. نکته کلیدی این است که منطق مسدودکننده کارگر را درون یک وظیفه پس‌زمینه (Background Task) قرار دهید تا از توقف حلقه رویداد جلوگیری شود.

import asyncio
import multiprocessing as mp
from aiohttp import ClientSession
import json

def cpu_intensive_parser(job):
    """کار سنگین CPU را روی یک فرآیند شبیه‌سازی می‌کند."""
    data = job['data']
    # شبیه‌سازی پردازش با تکرار و هش
    result = 0
    for char in data:
        result += hash(char)
    return result

async def fetch_task(session, url, data_queue, result_queue):
    """داده‌ها را به صورت ناهمگام دریافت کرده و پردازش را به کارگران واگذار می‌کند."""
    try:
        async with session.get(url) as response:
            html = await response.text()
            
            # ایجاد یک وظیفه برای کارگر
            job = {'url': url, 'data': html}
            data_queue.put(job)
            
            # انتظار برای نتیجه از کارگر خاص
            result = result_queue.get()
            return result
    except Exception as e:
        print(f"خطا در دریافت {url}: {e}")
        return None

async def main():
    # ایجاد صف‌ها برای ارتباط بین فرآیندی (IPC)
    data_queue = mp.Queue()
    result_queue = mp.Queue()
    
    # تعریف مجموعه کارگر
    num_workers = 4
    pool = []
    for i in range(num_workers):
        p = mp.Process(target=worker_loop, args=(data_queue, result_queue, i))
        p.start()
        pool.append(p)

    urls = [f"https://example.com/page/{i}" for i in range(10)]
    
    async with ClientSession() as session:
        tasks = []
        for url in urls:
            task = asyncio.create_task(fetch_task(session, url, data_queue, result_queue))
            tasks.append(task)
        
        results = await asyncio.gather(*tasks)
    
    # پاکسازی
    data_queue.close()
    data_queue.join_thread()
    for p in pool:
        p.terminate()

def worker_loop(data_queue, result_queue, worker_id):
    """در یک فرآیند جداگانه اجرا می‌شود تا وظایف وابسته به CPU را مدیریت کند."""
    while True:
        try:
            job = data_queue.get(timeout=5)
            if job is None:
                break
            
            result = cpu_intensive_parser(job)
            # در یک سناریوی واقعی، ممکن است نیاز باشد نتایج را با استفاده از شناسه‌ها به وظایف نگاشت کنید
            # اینجا ما یک انتظار مسدودکننده ساده برای اسلات نتیجه خاص را فرض می‌کنیم
            result_queue.put({'worker': worker_id, 'result': result})
        except:
            break

if __name__ == "__main__":
    # در ویندوز، باید نقطه ورود را محافظت کنیم
    mp.set_start_method('spawn')
    asyncio.run(main())

بهینه‌سازی ارتباط و مقیاس‌پذیری

در مثال بالا، result_queue یک صف مسدودکننده ساده است. برای اسکرپرهای سطح تولید که میلیون‌ها آیتم را مدیریت می‌کنند، ممکن است به یک مکانیزم همگام‌سازی پیشرفته‌تر، مانند asyncio.Queue که با مدیر حافظه مشترک تعامل دارد، یا استفاده از asyncio.to_thread برای وظایف سبک‌تر CPU (اگر به طور دقیق نیاز به عبور از GIL ندارید) نیاز داشته باشید.

علاوه بر این، اطمینان حاصل کنید که asyncio.TimeoutError و خطاهای شبکه را به صورت مناسب مدیریت می‌کنید. هنگام مقیاس‌پذیری به صدها درخواست همزمان، صف می‌تواند به یک گلوگاه تبدیل شود. در نظر بگیرید که از asyncio.Lock هنگام به‌روزرسانی آمار مشترک استفاده کنید یا از یک تابع جمع‌آوری‌کننده نتیجه اختصاصی استفاده کنید که صف کارگر را به صورت ناهمگام در پس‌زمینه مصرف می‌کند.

نتیجه‌گیری

ساخت یک اسکرپر با کارایی بالا در پایتون دیگر نیازی به انتخاب بین asyncio و multiprocessing ندارد. با پیاده‌سازی یک الگوی ترکیبی، شما توانایی مدیریت هزاران اتصال شبکه همزمان را همزمان با استفاده از تمام هسته‌های CPU موجود برای تجزیه و تبدیل داده‌ها کسب می‌کنید. این رویکرد حداکثر کارایی و حداقل تاخیر را تضمین می‌کند و آن را به استاندارد خطوط لوله استخراج داده مدرن و در مقیاس بزرگ تبدیل کرده است.

به یاد داشته باشید که همیشه فرآیندهای کارگر خود را برای قفل‌شدگی (Deadlocks) آزمایش کنید و اطمینان حاصل کنید که مکانیزم‌های IPC شما از نظر ایمنی رشته (Thread-safe) هستند. با معماری مناسب، پایتون همچنان یک نیروی غالب در دنیای وب‌اسکرپینگ و مهندسی داده باقی می‌ماند.