آموزش ماژول asyncio – بهبود عملکر با gather

اجرای هم‌زمان چند تسک با gather

برای اینکه چند تسک را با هم شروع کنید، همه را هم‌زمان پیش ببرید و در پایان نتایجشان را یک‌جا جمع کنید، از asyncio.gather استفاده می‌کنید. این تابع مجموعه‌ای از coroutineها یا تسک‌ها را می‌گیرد، همه را هم‌روند اجرا می‌کند و نتیجه‌ها را به ترتیب ورودی برمی‌گرداند (فارغ از اینکه کدام زودتر تمام شده).

import asyncio

async def download_file(file_name):
    assert '%' not in file_name, "Bad file name!"
    print(f'download {file_name} started')
    await asyncio.sleep(2)
    print(f'download {file_name} finished')
    return file_name

async def manage_downloads():
    t1 = asyncio.create_task(download_file('test1.txt'))
    t2 = asyncio.create_task(download_file('test2.txt'))

    res = await asyncio.gather(t1, t2)     # run both concurrently, collect results
    print(res)

asyncio.run(manage_downloads())

import asyncio

async def download_file(file_name):
    assert '%' not in file_name, "Bad file name!"
    print(f'download {file_name} started')
    await asyncio.sleep(2)
    print(f'download {file_name} finished')
    return file_name

async def manage_downloads():
    t1 = asyncio.create_task(download_file('test1.txt'))
    t2 = asyncio.create_task(download_file('test2.txt'))

    res = await asyncio.gather(t1, t2)     # run both concurrently, collect results
    print(res)

asyncio.run(manage_downloads())

هر دو دانلود با هم شروع می‌شوند و کل کار حدود ۲ ثانیه (نه ۴ ثانیه) طول می‌کشد؛ در پایان res برابر ['test1.txt', 'test2.txt'] است — دقیقاً به ترتیبی که به gather داده‌ایم.

تعامل با خطاها

در این مثال اگر نام فایل حاوی % باشد، coroutine خطا می‌دهد. ببینیم رفتار پیش‌فرض gather در برابر خطا چیست:

import asyncio

async def download_file(file_name):
    assert '%' not in file_name, "Bad file name!"
    print(f'download {file_name} started')
    await asyncio.sleep(2)
    print(f'download {file_name} finished')
    return file_name

async def manage_downloads():
    t1 = asyncio.create_task(download_file('test.txt'))
    t2 = asyncio.create_task(download_file('text%.txt'))   # this one raises

    res = await asyncio.gather(t1, t2)
    print(res)

asyncio.run(manage_downloads())

import asyncio

async def download_file(file_name):
    assert '%' not in file_name, "Bad file name!"
    print(f'download {file_name} started')
    await asyncio.sleep(2)
    print(f'download {file_name} finished')
    return file_name

async def manage_downloads():
    t1 = asyncio.create_task(download_file('test.txt'))
    t2 = asyncio.create_task(download_file('text%.txt'))   # this one raises

    res = await asyncio.gather(t1, t2)
    print(res)

asyncio.run(manage_downloads())

به‌صورت پیش‌فرض، به‌محض اینکه یکی از تسک‌ها خطا بدهد، همان خطا از gather بیرون می‌زند (propagate می‌شود) و برنامه با AssertionError: Bad file name! متوقف می‌شود. یعنی حتی نتیجه‌ی تسک سالم را هم از دست می‌دهیم.

جمع‌آوری خطاها با return_exceptions=True

اگر نمی‌خواهید یک خطا کل کار را متوقف کند، پارامتر return_exceptions=True را به gather بدهید. در این حالت به‌جای پرتاب‌شدن، خطاها هم مثل یک نتیجه‌ی عادی در فهرست خروجی برمی‌گردند:

import asyncio

async def download_file(file_name):
    assert '%' not in file_name, "Bad file name!"
    print(f'download {file_name} started')
    await asyncio.sleep(2)
    print(f'download {file_name} finished')
    return file_name

async def manage_downloads():
    t1 = asyncio.create_task(download_file('test.txt'))
    t2 = asyncio.create_task(download_file('text%.txt'))

    res = await asyncio.gather(t1, t2, return_exceptions=True)
    print(res)

asyncio.run(manage_downloads())

import asyncio

async def download_file(file_name):
    assert '%' not in file_name, "Bad file name!"
    print(f'download {file_name} started')
    await asyncio.sleep(2)
    print(f'download {file_name} finished')
    return file_name

async def manage_downloads():
    t1 = asyncio.create_task(download_file('test.txt'))
    t2 = asyncio.create_task(download_file('text%.txt'))

    res = await asyncio.gather(t1, t2, return_exceptions=True)
    print(res)

asyncio.run(manage_downloads())

حالا برنامه متوقف نمی‌شود و خروجی چنین چیزی است:

['test.txt', AssertionError('Bad file name!')]

تسک سالم نتیجه‌اش را می‌دهد و به‌جای تسک خطادار، خودِ شیء استثنا در فهرست قرار می‌گیرد. بعداً می‌توانید فهرست را بررسی کنید و با isinstance(item, Exception) موارد خطادار را جدا کنید.

مثال واقعی: فراخوانی چند API و جمع‌آوری همه‌ی نتایج

در دنیای واقعی اغلب می‌خواهید چند API را هم‌زمان صدا بزنید و نتایجشان را کنار هم بگذارید — اما نمی‌خواهید خرابیِ یکی، بقیه را از بین ببرد. gather با return_exceptions=True دقیقاً برای همین است:

import asyncio

async def call_api(name, delay, fail=False):
    await asyncio.sleep(delay)
    if fail:
        raise RuntimeError(f"{name} failed")
    return f"{name} ok"

async def main():
    results = await asyncio.gather(
        call_api("users", 1),
        call_api("orders", 2, fail=True),
        call_api("payments", 1),
        return_exceptions=True,
    )
    for r in results:
        if isinstance(r, Exception):
            print("error:", r)
        else:
            print("success:", r)

asyncio.run(main())

import asyncio

async def call_api(name, delay, fail=False):
    await asyncio.sleep(delay)
    if fail:
        raise RuntimeError(f"{name} failed")
    return f"{name} ok"

async def main():
    results = await asyncio.gather(
        call_api("users", 1),
        call_api("orders", 2, fail=True),
        call_api("payments", 1),
        return_exceptions=True,
    )
    for r in results:
        if isinstance(r, Exception):
            print("error:", r)
        else:
            print("success:", r)

asyncio.run(main())

این‌طور همه‌ی سرویس‌ها هم‌زمان فراخوانی می‌شوند، نتایج به ترتیب ورودی برمی‌گردند و حتی اگر یکی از سرویس‌ها خطا دهد، بقیه‌ی داده‌ها را در دست داریم.

gather در برابر as_completed

هر دو کارها را هم‌زمان اجرا می‌کنند، اما: gather منتظر می‌ماند تا همه تمام شوند و نتایج را یک‌جا و به ترتیب ورودی می‌دهد؛ در حالی که as_completed نتایج را یکی‌یکی و به ترتیب اتمام می‌دهد. اگر همه‌ی نتایج را مرتب و باهم می‌خواهید از gather، و اگر می‌خواهید هر نتیجه را به‌محض آماده‌شدن مصرف کنید از as_completed استفاده کنید.

جمع‌بندی

• asyncio.gather چند تسک را هم‌زمان اجرا می‌کند و نتایج را به ترتیب ورودی برمی‌گرداند.
• به‌صورت پیش‌فرض، اولین خطا از gather بیرون می‌زند و کار را متوقف می‌کند.
• با return_exceptions=True، خطاها به‌جای توقف برنامه، مثل نتیجه در فهرست خروجی برمی‌گردند.
• کاربرد واقعی: فراخوانی هم‌زمان چند سرویس و جمع‌آوری همه‌ی نتایج، حتی وقتی بعضی خطا می‌دهند.