آموزش ماژول asyncio – کلاس Queue

مقدمه

وقتی چند Task به‌صورت هم‌زمان در حال اجرا هستند، گاهی لازم است داده‌ای بین آن‌ها به‌صورت امن منتقل شود؛ مثلاً چند Task «تولیدکننده» (Producer) داده تولید می‌کنند و چند Task «مصرف‌کننده» (Consumer) آن‌ها را پردازش می‌کنند. کلاس Queue در ماژول asyncio دقیقاً برای همین منظور طراحی شده است و امکان تبادل امن داده بین Task‌ها را، بدون نیاز به قفل (Lock) دستی، فراهم می‌کند.

این کلاس از نظر رفتار شبیه به queue.Queue در ماژول threading است، با این تفاوت که متدهای آن به‌جای مسدود کردن Thread، به‌صورت await کار می‌کنند و با Event Loop سازگار هستند.

ساخت یک Queue

برای ساخت یک Queue کافیست از asyncio.Queue() استفاده کنید. این کلاس یک پارامتر اختیاری به نام maxsize می‌پذیرد که حداکثر تعداد آیتم‌های مجاز در صف را تعیین می‌کند. اگر مقدار آن صفر باشد یا تعیین نشود، صف نامحدود خواهد بود.

import asyncio

async def main():
    queue = asyncio.Queue(maxsize=5)
    print(f'ظرفیت صف: {queue.maxsize}')

asyncio.run(main())

متدهای اصلی کلاس Queue

• await queue.put(item): «یک آیتم به صف اضافه می‌کند؛ اگر صف پر باشد، تا خالی شدن جا منتظر می‌ماند.»
• queue.put_nowait(item): «بدون انتظار آیتم را اضافه می‌کند؛ اگر صف پر باشد خطای QueueFull رخ می‌دهد.»
• await queue.get(): «یک آیتم از صف برمی‌گرداند؛ اگر صف خالی باشد تا رسیدن آیتم منتظر می‌ماند.»
• queue.get_nowait(): «بدون انتظار آیتم را برمی‌گرداند؛ اگر صف خالی باشد خطای QueueEmpty رخ می‌دهد.»
• queue.empty(): «اگر صف خالی باشد True برمی‌گرداند.»
• queue.full(): «اگر صف به حداکثر ظرفیت رسیده باشد True برمی‌گرداند.»
• queue.qsize(): «تعداد آیتم‌های فعلی موجود در صف را برمی‌گرداند.»
• queue.task_done(): «اعلام می‌کند که پردازش یک آیتم گرفته‌شده از صف به پایان رسیده است.»
• await queue.join(): «تا زمانی که برای همهٔ آیتم‌های put شده، متد task_done صدا زده شود، منتظر می‌ماند.»

مثال: الگوی Producer-Consumer

یکی از پرکاربردترین الگوهای استفاده از Queue، الگوی Producer-Consumer است. در این الگو، یک یا چند Task نقش تولیدکننده را دارند و آیتم‌هایی را در صف قرار می‌دهند، و یک یا چند Task دیگر نقش مصرف‌کننده را دارند و آن آیتم‌ها را از صف برداشت و پردازش می‌کنند.

نکتهٔ مهم در پایان کار، استفاده از queue.join() در کنار queue.task_done() است؛ این دو با هم تضمین می‌کنند که برنامه تا پردازش کامل همهٔ آیتم‌ها صبر کند، حتی اگر مصرف‌کننده‌ها در یک حلقهٔ بی‌پایان (while True) منتظر آیتم بعدی باشند.

import asyncio
import random

async def producer(queue, name):
    for i in range(3):
        item = f'{name}-item-{i}'
        delay = random.uniform(0.1, 0.5)
        await asyncio.sleep(delay)
        await queue.put(item)
        print(f'{name}: تولید شد -> {item}')

async def consumer(queue, name):
    while True:
        item = await queue.get()
        process_delay = random.uniform(0.1, 0.3)
        await asyncio.sleep(process_delay)
        print(f'{name}: پردازش شد -> {item}')
        queue.task_done()

async def main():
    queue = asyncio.Queue()

    producers = [asyncio.create_task(producer(queue, f'producer-{i}')) for i in range(2)]
    consumers = [asyncio.create_task(consumer(queue, f'consumer-{i}')) for i in range(3)]

    await asyncio.gather(*producers)
    await queue.join()

    for c in consumers:
        c.cancel()

asyncio.run(main())

در این مثال، چون مصرف‌کننده‌ها در حلقهٔ بی‌پایانی منتظر آیتم جدید می‌مانند، باید پس از اتمام کار با queue.join()، به‌صورت دستی با c.cancel() آن‌ها را متوقف کرد؛ در غیر این صورت برنامه هرگز پایان نمی‌یابد.

جمع‌بندی

کلاس Queue ابزاری امن و کاربردی برای هماهنگی بین Task‌های هم‌زمان در asyncio است و الگوی Producer-Consumer یکی از رایج‌ترین کاربردهای آن به شمار می‌رود. در درس بعدی، با توابع wait و wait_for آشنا می‌شویم که کنترل دقیق‌تری روی اجرای هم‌زمان چند Task به ما می‌دهند.