کتابخانه sched در پایتون — راهنمای کامل

ماژول sched در استاندارد پایتون یک زمان‌بند ساده و سبک است که برای برنامه‌ریزی اجرای توابع در زمان‌های مشخص یا بعد از تأخیری مشخص طراحی شده است. این ماژول برای کارهای سبک و تک‌نخی مناسب است و در بسیاری از سناریوهای ساده‌ی زمان‌بندی کاربردی و مفید است. در این مقاله مفاهیم اصلی، نمونه‌های عملی، نکات پیشرفته و محدودیت‌ها را بررسی می‌کنیم.

مفاهیم پایه

• scheduler: شیئی از کلاس sched.scheduler که رویدادها را نگه می‌دارد و اجرا می‌کند.
• event: یک رویداد زمان‌بندی‌شده که شامل زمان اجرا، اولویت و تابع قابل اجراست.
• timefunc و delayfunc: توابعی که برای تعیین زمان کنونی و تأخیر استفاده می‌شوند (به‌طور پیش‌فرض time.time و time.sleep).

ساختار و متدهای اصلی

متدهای پرکاربرد:

نمونه ساده: اجرای تابع بعد از تأخیر

import sched, time

s = sched.scheduler(time.time, time.sleep)

def hello(name):
    print(time.time(), "Hello", name)

# زمان‌بندی اجرای hello بعد از 5 ثانیه
s.enter(5, 1, hello, ("World",))
s.run()

در این مثال یک شی scheduler ساخته‌ایم که از time.time و time.sleep استفاده می‌کند. تابع hello بعد از 5 ثانیه اجرا می‌شود. اگر run فراخوانی شود، پردازش اصلی تا اجرای رویداد بلوک می‌شود.

توضیح بیشتر و نکات عملی

• اولویت (priority): وقتی دو رویداد زمان یکسان دارند، مقدار priority برای ترتیب اجرا استفاده می‌شود (مقدار کمتر => اول اجرا).
• run به‌صورت پیش‌فرض بلوک‌کننده است؛ برای کار در پس‌زمینه می‌توان scheduler را در یک نخ جدا اجرا کرد.
• sched از کُپه (heapq) برای نگهداری رویدادها استفاده می‌کند، بنابراین پیچیدگی عملیاتی مناسب و کارا است.

سیناریوی چند تکه: اجرای زمان‌بندی در یک Thread

import sched, time, threading

s = sched.scheduler(time.time, time.sleep)

def task(name):
    print(time.time(), "Task:", name)

# زمان‌بندی چند رویداد
s.enter(2, 1, task, ("A",))
s.enter(4, 1, task, ("B",))

# اجرای scheduler در یک نخ جدا
t = threading.Thread(target=s.run)
t.start()

print("Main thread continues")

در این نمونه scheduler در یک نخ جدا اجرا می‌شود تا نخ اصلی آزاد بماند. توجه کنید که sched به‌تنهایی thread-safe نیست؛ اگر از نخ‌های مختلف قرار است به scheduler رویداد اضافه یا حذف کنید، لازم است از قفل (Lock) استفاده کنید تا شرایط رقابتی رخ ندهد.

زمان‌بندی دوره‌ای (Periodic) — دو روش و نکته مهم درباره drift

برای اجرای دوره‌ای یک وظیفه معمولاً دو روش وجود دارد: re-enter در انتهای اجرا یا استفاده از زمان‌های مطلق برای جلوگیری از انباشته شدن خطا (drift).

# روش ساده (ممکن است drift ایجاد کند)
def periodic_simple(scheduler, interval, action, args=()):
    def wrapper():
        try:
            action(*args)
        finally:
            scheduler.enter(interval, 1, wrapper)
    scheduler.enter(interval, 1, wrapper)

این روش هر بار که اجرا تمام می‌شود، اجرا بعدی را به فاصله interval برنامه‌ریزی می‌کند. اگر اجرای action طولانی شود، اجرای بعدی به تأخیر می‌افتد و drift رخ می‌دهد.

# روش دقیق‌تر با زمان‌های مطلق (کمترین drift)
def periodic_precise(scheduler, start, interval, action, args=()):
    def wrapper(next_time):
        now = time.time()
        try:
            action(*args)
        finally:
            next_time += interval
            scheduler.enterabs(next_time, 1, wrapper, (next_time,))
    scheduler.enterabs(start, 1, wrapper, (start,))

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

لغو رویداد و مدیریت خطا

e = s.enter(10, 1, print, ("Hi",))
# ... در صورتی که لازم باشد
s.cancel(e)

لغو رویداد با ارجاع به شی event انجام می‌شود. همچنین اگر یک رویداد هنگام اجرا استثنا پرتاب کند، آن استثنا می‌تواند اجرای بعدی scheduler را متوقف یا مسئله‌ساز کند؛ بنابراین بهتر است در داخل action از try/except استفاده کنید تا scheduler پایدار بماند.

محدودیت‌ها و جایگزین‌ها

• sched مناسب وظایف سبک و ساده است؛ برای برنامه‌ریزی پیچیده، تکرارهای پیچیده، persistence یا job stores بهتر است از کتابخانه‌هایی مانند APScheduler استفاده کنید.
• sched به‌صورت بومی thread-safe نیست؛ برای استفاده concurrent باید سینک‌سازی دستی انجام دهید یا scheduler را تنها از یک نخ مدیریت کنید.
• اگر نیاز به کار با IO غیرهمزمان یا event loop دارید، از asyncio استفاده کنید که قابلیت زمان‌بندی و سازگاری با عملیات غیربلاک‌کننده را دارد.

مثال پیشرفته: باز تلاش (retry) با backoff

import sched, time
s = sched.scheduler(time.time, time.sleep)

def do_task():
    print(time.time(), "Trying")
    raise RuntimeError("fail")

def retry_wrapper(attempt=1, delay=1):
    try:
        do_task()
    except Exception as e:
        print("Attempt", attempt, "failed:", e)
        # افزایش backoff نمایی
        next_delay = delay * 2
        s.enter(next_delay, 1, retry_wrapper, (attempt+1, next_delay))

s.enter(0, 1, retry_wrapper, ())
s.run()

در این کد نمونه یک مکانیزم بازتلاش با backoff نمایی را پیاده‌سازی کرده‌ایم. توجه کنید که exception درون wrapper مدیریت می‌شود تا scheduler از کار نیفتد.

جمع‌بندی و توصیه‌های عملی

ماژول sched ابزار ساده و مفیدی برای زمان‌بندی وظایف سبک در برنامه‌های پایتون است. برای استفاده صحیح توجه داشته باشید:

• اگر ریزه‌کاری زمان‌بندی دقیق لازم است از enterabs استفاده کنید تا drift کاهش یابد.
• برای وظایف طولانی یا blocking آن‌ها را در نخ یا پردازش جدا اجرا کنید تا صف scheduler مسدود نشود.
• برای برنامه‌های پیچیده، چندمنظوره یا نیازمند persistence به سراغ APScheduler یا سیستم‌های cron/queue بروید.

با رعایت این نکات می‌توانید از sched برای طیف وسیعی از نیازهای زمان‌بندی ساده بهره ببرید و در صورت رشد نیازها به ابزارهای پیشرفته‌تر مهاجرت کنید.

لطفاً از کمبود ها و مشکلات این محتوا برای ما بنویسید

ارسال