django apscheduler 定時任務，輪詢執行某個任務（上篇）

這是一個 Django 應用程序，它為 APScheduler 添加了一個輕量級的包裝器。它允許使用 Django 的 ORM 在數據庫中存儲持久作業。

django-apscheduler 是一個很好的選擇，可以快速輕松地將基本調度功能添加到您的 Django 應用程序中，並且依賴最少，附加配置很少。理想的用例可能涉及按固定執行計划運行少量任務。

這種簡單性的代價是您需要小心確保只有一個調度程序在特定時間點處於活動狀態。

這種限制源於這樣一個事實，即 APScheduler 當前沒有任何進程間同步和信令方案 ，可以在作業已添加、修改或從作業存儲中刪除時通知調度程序。

似乎計划在即將發布的 APScheduler 4.0 版本中支持多個調度程序之間的持久作業存儲共享。在此之前，一個典型的Django部署在生產中會啟動多個工作進程，如果每個工作進程結束了運行其自己的調度那么這可能會導致工作中被遺漏或多次執行，以及重復的條目DjangoJobExecution表正在創建。

所以現在你的選擇是：

1. 使用自定義 Django 管理命令在其自己的專用進程中啟動單個調度程序（推薦- 請參見runapscheduler.py下面的示例）；或者

2. 實現您自己的遠程處理 邏輯，以確保DjangoJobStore所有 Web 服務器的工作進程都可以以協調和同步的方式使用單個進程（對於大多數用例而言，額外的努力和增加的復雜性可能不值得）；或者

3. 選擇的替代任務處理庫確實使用某種像的Redis，RabbitMQ的，亞馬遜SQS或類似共享消息代理的支持進程間通信（參見： https://djangopackages.org/grids/g/workers-queues-tasks /對於流行的選項）。

該軟件包的特點包括：

• 自定義DjangoJobStore：一個APScheduler 作業存儲 ，將預定作業保存到 Django 數據庫。您可以直接通過 Django 管理界面查看計划作業並監控作業執行情況：

  

• 作業存儲還維護當前計划作業的所有作業執行歷史，以及狀態代碼和異常（如果有）：

  

• 注意： APScheduler 會 在作業的最后一次計划執行被觸發后自動從作業存儲中刪除作業。這也會從數據庫中刪除相應的作業執行條目（即，作業執行日志僅針對“活動”作業進行維護。）

• 也可以通過DjangoJob管理頁面手動觸發作業執行：

  

• 注意：為了防止長時間運行的作業導致 Django HTTP 請求超時，通過 Django 管理站點啟動的所有 APScheduler 作業的組合最大運行時間為 25 秒。這個超時值可以通過APSCHEDULER_RUN_NOW_TIMEOUT設置來配置。

• 安裝

  pip  install django - apscheduler

   

  快速開始

  • 像這樣添加django_apscheduler到您的INSTALLED_APPS設置中：

  INSTALLED_APPS  = ( # ... "django_apscheduler" , )

  • django-apscheduler 帶有開箱即用的合理配置默認值。可以通過將以下設置添加到 Djangosettings.py文件來覆蓋默認值：

  • # Format string for displaying run time timestamps in the Django admin site. The default
    # just adds seconds to the standard Django format, which is useful for displaying the timestamps
    # for jobs that are scheduled to run on intervals of less than one minute.
    # 
    # See https://docs.djangoproject.com/en/dev/ref/settings/#datetime-format for format string
    # syntax details.
    APSCHEDULER_DATETIME_FORMAT = "N j, Y, f:s a"
    
    # Maximum run time allowed for jobs that are triggered manually via the Django admin site, which
    # prevents admin site HTTP requests from timing out.
    # 
    # Longer running jobs should probably be handed over to a background task processing library
    # that supports multiple background worker processes instead (e.g. Dramatiq, Celery, Django-RQ,
    # etc. See: https://djangopackages.org/grids/g/workers-queues-tasks/ for popular options).
    APSCHEDULER_RUN_NOW_TIMEOUT = 25  # Seconds
    

    　　

    # runapscheduler.py
    import logging
    
    from django.conf import settings
    
    from apscheduler.schedulers.blocking import BlockingScheduler
    from apscheduler.triggers.cron import CronTrigger
    from django.core.management.base import BaseCommand
    from django_apscheduler.jobstores import DjangoJobStore
    from django_apscheduler.models import DjangoJobExecution
    
    
    logger = logging.getLogger(__name__)
    
    
    def my_job():
        #  Your job processing logic here... 
        pass
    
    def delete_old_job_executions(max_age=604_800):
        """This job deletes all apscheduler job executions older than `max_age` from the database."""
        DjangoJobExecution.objects.delete_old_job_executions(max_age)
    
    
    class Command(BaseCommand):
        help = "Runs apscheduler."
    
        def handle(self, *args, **options):
            scheduler = BlockingScheduler(timezone=settings.TIME_ZONE)
            scheduler.add_jobstore(DjangoJobStore(), "default")
            
            scheduler.add_job(
                my_job,
                trigger=CronTrigger(second="*/10"),  # Every 10 seconds
                id="my_job",  # The `id` assigned to each job MUST be unique
                max_instances=1,
                replace_existing=True,
            )
            logger.info("Added job 'my_job'.")
    
            scheduler.add_job(
                delete_old_job_executions,
                trigger=CronTrigger(
                    day_of_week="mon", hour="00", minute="00"
                ),  # Midnight on Monday, before start of the next work week.
                id="delete_old_job_executions",
                max_instances=1,
                replace_existing=True,
            )
            logger.info(
                "Added weekly job: 'delete_old_job_executions'."
            )
    
            try:
                logger.info("Starting scheduler...")
                scheduler.start()
            except KeyboardInterrupt:
                logger.info("Stopping scheduler...")
                scheduler.shutdown()
                logger.info("Scheduler shut down successfully!")
    

    　　

    • ./manage.py runapscheduler每當啟動為 Django 應用程序提供服務的 Web 服務器時，都應調用此管理命令。如何以及在何處完成此操作的詳細信息取決於具體實現，並取決於您使用的 Web 服務器以及您如何將應用程序部署到生產環境。對於大多數人來說，這應該涉及配置各種主管進程。

    • 像往常一樣注冊任何 APScheduler 作業。請注意，如果您尚未設置DjangoJobStore為'default' 作業存儲，則需要將其包含jobstore='djangojobstore'在您的scheduler.add_job()調用中。

    高級用法

    django-apscheduler 假設您已經熟悉 APScheduler 及其正確使用。如果沒有，請轉到項目頁面並查看APScheduler 文檔。

    根據您的環境和用例，可以使用不同類型的調度程序。如果您更喜歡運行 aBackgroundScheduler而不是使用 a BlockingScheduler，那么您應該知道將 APScheduler 與 uWSGI 一起使用需要一些額外的 配置步驟才能重新啟用線程支持。

    支持的數據庫

    請注意Django 官方支持的數據庫列表。django-apscheduler 可能不適用於不受支持的數據庫，如 Microsoft SQL Server、MongoDB 等。

    數據庫連接和超時

    django-apscheduler 依賴於標准的 Django 數據庫配置設置。這些設置與您的數據庫的配置方式相結合，決定了您的特定部署將如何進行連接管理。

    如果您遇到任何類型的“丟失連接”錯誤，則可能意味着：

    • 您的數據庫連接已超時。可能是時候開始考慮部署連接池程序（如 pgbouncer）來為您管理數據庫連接。
    • 您的數據庫服務器已崩潰/已重新啟動。Django不會自動重新連接。