هم‌زمانی در پایتون با TaskGroup؛ زمان انتظار و لغو کارها

import asyncio

async def fetch(name, gate):
    async with gate:
        await asyncio.sleep(0.05)
        return name.upper()

async def main():
    gate = asyncio.Semaphore(2)
    async with asyncio.timeout(1):
        async with asyncio.TaskGroup() as group:
            jobs = [group.create_task(fetch(name, gate))
                    for name in ("stock", "payment", "delivery")]
    print([job.result() for job in jobs])

asyncio.run(main())

فایل را با python اجرا کنید؛ خروجی باید STOCK، PAYMENT و DELIVERY باشد. برای مشاهده TimeoutError، زمان sleep را به 2 تغییر دهید. مثال خطای timeout را عمداً به فراخواننده منتقل می‌کند؛ در سرویس واقعی آن را در مرز درخواست به پاسخ مناسب تبدیل کنید.

این فایل مستقل را به نام group_failure.py اجرا کنید. خطای وابستگی باید به فراخواننده برسد و کار کندتر نیز فرصت پاک‌سازی داشته باشد؛ بنابراین finally بخشی از رفتار درست است، نه صرفاً یک پیام اضافی.

import asyncio

async def slow():
    try:
        await asyncio.sleep(30)
    finally:
        print("slow task cleaned up")

async def broken():
    await asyncio.sleep(0.01)
    raise ValueError("dependency failed")

async def main():
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(slow())
            group.create_task(broken())
    except* ValueError as errors:
        print("handled errors:", len(errors.exceptions))

asyncio.run(main())

انتظار دو پیام پاک‌سازی و handled errors: 1 را داریم. except* فقط ValueError را رسیدگی می‌کند و خطاهای دیگر پنهان نمی‌شوند. در برنامه واقعی، بستن اتصال یا آزادکردن فایل را جایگزین پیام پاک‌سازی کنید.

در این فایل مستقل، لایه بارگیری از بودجه درخواست خبر ندارد و مرز درخواست timeout را تعیین می‌کند. به جای صفر یا لیست خالی، نتیجه unavailable برمی‌گردد تا رابط کاربر نتواند نبود داده را با داده واقعی اشتباه بگیرد.

import asyncio

async def load_summary():
    await asyncio.sleep(0.2)
    return {"status": "ready"}

async def handle_request():
    try:
        async with asyncio.timeout(0.05):
            return await load_summary()
    except TimeoutError:
        return {"status": "unavailable", "reason": "deadline"}

result = asyncio.run(handle_request())
assert result == {"status": "unavailable", "reason": "deadline"}
print(result)

خروجی باید reason برابر deadline داشته باشد. زمان بارگیری را به 0.01 تغییر دهید تا مسیر موفق را هم ببینید و assertion را با انتظار همان سناریو هماهنگ کنید. تکرار خودکار نامحدود داخل این محدوده اضافه نکنید.

Semaphore فقط کار فعال را محدود می‌کند؛ اگر برای یک میلیون ورودی یک میلیون task بسازید، حافظه همچنان مصرف می‌شود. این نسخه مستقل با دو worker و صف چهارعضوی، backpressure را به تولیدکننده منتقل می‌کند.

import asyncio

async def worker(queue, results):
    while True:
        item = await queue.get()
        try:
            if item is None:
                return
            await asyncio.sleep(0.001)
            results.append(item * 2)
        finally:
            queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=4)
    results = []
    async with asyncio.TaskGroup() as group:
        for _ in range(2):
            group.create_task(worker(queue, results))
        for item in range(10):
            await queue.put(item)
        for _ in range(2):
            await queue.put(None)
        await queue.join()
    assert sorted(results) == list(range(0, 20, 2))
    print("processed:", len(results))

asyncio.run(main())

خروجی processed: 10 است و ترتیب پایان کارها را نباید قرارداد نتیجه بدانید. برای ورودی نامحدود، خود results هم باید به خروجی جریانی تبدیل شود. خطای worker در TaskGroup به کل دامنه منتقل می‌شود؛ کار دارای اثر جانبی به سیاست تلاش مجدد و ثبت نتیجه مستقل نیاز دارد.