آموزش ماژول queue – صف امن بین تردها

مقدمه

در درس‌های قبلی با Lock، RLock، Semaphore و سایر ابزارهای همگام‌سازی در ماژول threading آشنا شدید. این ابزارها برای محافظت از یک منبع مشترک در برابر دسترسی همزمان عالی هستند، اما وقتی هدف شما «انتقال داده» بین تردها باشد – مثلاً یک ترد داده تولید می‌کند و ترد دیگری آن را مصرف می‌کند – استفاده مستقیم از Lock کار را پیچیده و مستعد خطا می‌کند. باید به صورت دستی صف بسازید، با Lock از آن محافظت کنید و با Condition تردهای منتظر را بیدار کنید.

ماژول queue در پایتون دقیقاً همین کار را از قبل برای شما انجام داده است. کلاس Queue یک صف FIFO (اول وارد، اول خارج) است که به صورت کامل thread-safe پیاده‌سازی شده و تمام جزئیات قفل‌گذاری و اطلاع‌رسانی بین تردها را داخلی مدیریت می‌کند.

کلاس Queue

برای استفاده از این کلاس باید آن را از ماژول queue ایمپورت کنید.

from queue import Queue

q = Queue(maxsize=10)

from queue import Queue

q = Queue(maxsize=10)

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

متدهای اصلی

• put(item, block=True, timeout=None): یک آیتم به صف اضافه می‌کند. اگر صف پر باشد و block=True باشد، ترد منتظر می‌ماند تا جا باز شود (یا تا پایان timeout اگر مشخص شده باشد و در غیر این صورت خطای queue.Full پرتاب می‌شود).
• get(block=True, timeout=None): یک آیتم از صف برمی‌گرداند و آن را حذف می‌کند. اگر صف خالی باشد، رفتار مشابه put دارد (انتظار، یا timeout، یا پرتاب queue.Empty).
• put_nowait(item) و get_nowait(): معادل غیرمسدودکننده‌ی دو متد بالا (یعنی block=False).
• qsize(): تعداد تقریبی آیتم‌های موجود در صف (در محیط چند تردی نباید کاملاً به آن اعتماد کرد، چون ممکن است بین خوانده شدن و استفاده از آن، مقدار صف تغییر کند).
• empty() و full(): بررسی خالی یا پر بودن صف؛ این متدها هم به همان دلیل بالا قابل اعتماد کامل برای تصمیم‌گیری نیستند و صرفاً جهت اطلاع هستند.

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

در مثال زیر یک ترد تولیدکننده (producer) چند عدد تولید می‌کند و در صف می‌گذارد، و دو ترد مصرف‌کننده (consumer) آن‌ها را از صف می‌خوانند و پردازش می‌کنند. توجه کنید که هیچ Lock یا Condition دستی در این کد وجود ندارد؛ کلاس Queue همه‌ی این جزئیات را خودش مدیریت می‌کند.

import threading
import time
from queue import Queue

def producer(q, count):
    for i in range(count):
        print(f'producer: putting {i}')
        q.put(i)
        time.sleep(0.1)
    # برای هر مصرف‌کننده یک پیام پایان (sentinel) می‌فرستیم
    for _ in range(2):
        q.put(None)

def consumer(q, name):
    while True:
        item = q.get()
        if item is None:
            break
        print(f'{name}: processing {item}')
        time.sleep(0.2)
    print(f'{name}: done')

if __name__ == '__main__':
    q = Queue(maxsize=5)
    threads = [
        threading.Thread(target=producer, args=(q, 10)),
        threading.Thread(target=consumer, args=(q, 'consumer-1')),
        threading.Thread(target=consumer, args=(q, 'consumer-2')),
    ]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

import threading
import time
from queue import Queue

def producer(q, count):
    for i in range(count):
        print(f'producer: putting {i}')
        q.put(i)
        time.sleep(0.1)
    # برای هر مصرف‌کننده یک پیام پایان (sentinel) می‌فرستیم
    for _ in range(2):
        q.put(None)

def consumer(q, name):
    while True:
        item = q.get()
        if item is None:
            break
        print(f'{name}: processing {item}')
        time.sleep(0.2)
    print(f'{name}: done')

if __name__ == '__main__':
    q = Queue(maxsize=5)
    threads = [
        threading.Thread(target=producer, args=(q, 10)),
        threading.Thread(target=consumer, args=(q, 'consumer-1')),
        threading.Thread(target=consumer, args=(q, 'consumer-2')),
    ]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

الگوی None در اینجا یک «مقدار نشانه» (sentinel) است که به مصرف‌کننده اعلام می‌کند دیگر داده‌ای در راه نیست و باید حلقه را ترک کند. چون دو مصرف‌کننده داریم، باید به همان تعداد پیام پایان بفرستیم تا هر دو متوجه شوند.

مدیریت پایان کار با task_done و join

روش دیگری برای اطلاع از اتمام پردازش – به‌جای ارسال مقدار نشانه – استفاده از زوج متد task_done() و join() است:

• هر بار که مصرف‌کننده با get() یک آیتم برمی‌دارد و پردازش آن را به طور کامل تمام می‌کند، باید q.task_done() را صدا بزند تا به صف اعلام کند «این آیتم را پردازش کردم».
• متد q.join() در ترد اصلی (یا هر تردی که آن را صدا بزند) مسدود می‌شود تا زمانی که برای هر آیتمی که با put() اضافه شده، یک task_done() متناظر هم صدا زده شده باشد.

import threading
from queue import Queue

def worker(q):
    while True:
        item = q.get()
        print(f'processing {item}')
        q.task_done()   # حتما باید صدا زده شود، حتی اگر پردازش با خطا مواجه شود

q = Queue()
t = threading.Thread(target=worker, args=(q,), daemon=True)
t.start()

for i in range(5):
    q.put(i)

q.join()   # تا زمانی که هر ۵ آیتم task_done نشوند، اینجا منتظر می‌ماند
print('all items processed')

import threading
from queue import Queue

def worker(q):
    while True:
        item = q.get()
        print(f'processing {item}')
        q.task_done()   # حتما باید صدا زده شود، حتی اگر پردازش با خطا مواجه شود

q = Queue()
t = threading.Thread(target=worker, args=(q,), daemon=True)
t.start()

for i in range(5):
    q.put(i)

q.join()   # تا زمانی که هر ۵ آیتم task_done نشوند، اینجا منتظر می‌ماند
print('all items processed')

نکته‌ی مهم: اگر در پردازش یک آیتم استثنایی (exception) رخ دهد و task_done() صدا زده نشود، join() تا ابد منتظر می‌ماند و برنامه عملاً قفل می‌شود (یک نوع deadlock). همیشه فراخوانی task_done() را درون بلوک try/finally قرار دهید تا حتی در صورت بروز خطا هم اجرا شود.

انواع دیگر صف: LifoQueue و PriorityQueue

ماژول queue دو کلاس دیگر هم دارد که همان رابط (interface) کلاس Queue را دارند اما ترتیب خروج آیتم‌ها متفاوت است:

• LifoQueue: ساختار LIFO (آخرین وارد، اول خارج) – مانند یک پشته (stack).
• PriorityQueue: آیتم‌ها بر اساس اولویت مرتب می‌شوند؛ کوچک‌ترین مقدار، اول خارج می‌شود. آیتم‌ها معمولاً به شکل تاپل (priority, item) وارد می‌شوند.

from queue import PriorityQueue

pq = PriorityQueue()
pq.put((2, 'task-b'))
pq.put((1, 'task-a'))
pq.put((3, 'task-c'))

while not pq.empty():
    priority, task = pq.get()
    print(priority, task)
# خروجی به ترتیب اولویت: 1 task-a   2 task-b   3 task-c

from queue import PriorityQueue

pq = PriorityQueue()
pq.put((2, 'task-b'))
pq.put((1, 'task-a'))
pq.put((3, 'task-c'))

while not pq.empty():
    priority, task = pq.get()
    print(priority, task)
# خروجی به ترتیب اولویت: 1 task-a   2 task-b   3 task-c

SimpleQueue؛ نسخه‌ای سریع‌تر و ساده‌تر

اگر فقط به یک صف FIFO ساده برای انتقال داده نیاز دارید و قابلیت‌های maxsize یا task_done/join برایتان مهم نیست، کلاس SimpleQueue گزینه‌ی بهتری است. این کلاس سربار کمتری دارد و رابط بسیار کوچکی شامل put، get، empty و qsize ارائه می‌کند. توجه کنید که SimpleQueue پارامتر maxsize ندارد و همیشه نامحدود است.

اشتباهات رایج

• فراموش کردن task_done: اگر از الگوی join() استفاده می‌کنید و حتی یک بار task_done() را فراموش کنید، برنامه برای همیشه در join() منتظر می‌ماند.
• اعتماد به qsize/empty برای تصمیم‌گیری: چون چند ترد ممکن است همزمان به صف دسترسی داشته باشند، بین لحظه‌ی بررسی empty() و لحظه‌ی صدا زدن get() ممکن است وضعیت صف تغییر کند. به‌جای بررسی شرطی، مستقیماً از get() با مدیریت قطعه‌ی استثنای queue.Empty (در حالت غیرمسدود) استفاده کنید.
• maxsize نامحدود برای داده‌ی نامحدود: اگر تولیدکننده همیشه سریع‌تر از مصرف‌کننده باشد و maxsize تعیین نکرده باشید، صف بدون محدودیت بزرگ می‌شود و مصرف حافظه افزایش می‌یابد. تعیین یک maxsize معقول باعث می‌شود put() به طور طبیعی تولیدکننده را کنترل کند (backpressure).

جمع‌بندی

کلاس Queue از ماژول queue ساده‌ترین و امن‌ترین راه برای انتقال داده بین تردها در پایتون است. به‌جای پیاده‌سازی دستی صف با Lock و Condition، از همین کلاس آماده استفاده کنید. در درس‌های بعدی، نسخه‌ی مخصوص multiprocessing این کلاس و همچنین نسخه‌ی asyncio آن (asyncio.Queue) را بررسی خواهیم کرد که از نظر مفهومی بسیار شبیه به این درس هستند.