Django APScheduler

APScheduler for Django.

这是一个Django应用程序，它通过APScheduler添加了一个轻量级的包装器。它使您可以使用Django的ORM将持久作业存储在数据库中。

django-apscheduler是一个很好的选择，可以快速轻松地将基本调度功能添加到您的Django应用程序中，具有最少的依赖项和非常少的额外配置。理想的使用案例可能涉及在固定的执行计划上运行少量任务。

请注意：这种简单性的代价是需要你确保在某个特定时间点只有一个调度器正在运行。

这个限制是由于 APScheduler 目前没有任何 进程间同步和信号机制，使得调度器能够在作业被添加、修改或从作业存储中删除时得到通知（换句话说，不同的调度器无法判断作业是否已经被其他调度器运行过，并且直接在数据库中更改作业的预定运行时间是没有效果的，除非你同时重启调度器）。

根据你当前的 Django 部署方式，解决这个限制可能需要一些思考。在生产环境中，启动许多 webserver 工作进程以实现扩展和处理大量用户流量是很常见的。如果这些工作进程中的每一个都运行自己的调度器，则可能导致作业被遗漏或多次执行，以及 DjangoJobExecution 表中创建重复条目。

似乎计划在即将到来的 APScheduler 4.0 版本 中支持在多个调度器之间共享持久作业存储。

因此，目前你的选择是：

1. 使用自定义的 Django 管理命令以自己的专用进程启动单个调度器（推荐 - 请参阅下面的 runapscheduler.py 示例）；或者

2. 实现自己的 远程处理逻辑，以确保单个 DjangoJobStore 可以由所有 webserver 的工作进程以协调和同步的方式使用（这可能对于大多数用例来说不值得额外的努力和复杂性增加）；或者

3. 选择一个支持使用某些类型的共享消息代理（如 Redis、RabbitMQ、Amazon SQS 等）进行进程间通信的替代任务处理库（有关常用选项，请参阅：https://djangopackages.org/grids/g/workers-queues-tasks/）。

功能

• 自定义 APScheduler 作业存储（DjangoJobStore），将预定作业持久化到 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 默认提供合理的配置。默认值可以通过向你的 Django settings.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.django.ac.cn/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

• 运行 python manage.py migrate 以创建 django_apscheduler 模型。

• 将一个用于安排 APScheduler 任务并启动调度器的自定义 Django 管理命令添加到您的项目中。自定义 Django 管理命令。

# 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
from django_apscheduler import util

logger = logging.getLogger(__name__)


def my_job():
  # Your job processing logic here...
  pass


# The `close_old_connections` decorator ensures that database connections, that have become
# unusable or are obsolete, are closed before and after your job has run. You should use it
# to wrap any jobs that you schedule that access the Django database in any way. 
@util.close_old_connections
def delete_old_job_executions(max_age=604_800):
  """
  This job deletes APScheduler job execution entries older than `max_age` from the database.
  It helps to prevent the database from filling up with old historical records that are no
  longer useful.
  
  :param max_age: The maximum length of time to retain historical job execution records.
                  Defaults to 7 days.
  """
  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 应用的网络服务器时。具体的操作方法和位置取决于您使用的网络服务器以及如何部署您的应用程序到生产环境。对于大多数人来说，这通常涉及到配置某种类型的 supervisor 进程。

• 像平常一样注册任何 APScheduler 任务。请注意，如果您尚未将 DjangoJobStore 设置为默认的 'default' 作业存储，那么您需要在 scheduler.add_job() 调用中包含 jobstore='djangojobstore'。

高级用法

django-apscheduler 假定您已经熟悉 APScheduler 及其正确使用。如果您不熟悉，请访问项目页面，并查阅 APScheduler 文档。

根据您的环境和用例，可以使用不同类型的调度器。如果您希望运行 BackgroundScheduler 而不是使用 BlockingScheduler，那么您应该知道，使用 APScheduler 与 uWSGI 需要一些额外的 配置步骤 以重新启用线程支持。

支持的数据库

请注意 Django 官方支持的数据库列表。Django 支持的数据库。django-apscheduler 可能不会与 Microsoft SQL Server、MongoDB 等不受支持的数据库一起工作。

数据库连接和超时

django-apscheduler 依赖于标准的 Django 数据库 配置设置。这些设置，结合您的数据库服务器配置，决定了您的特定部署中如何执行连接管理。

对于需要数据库访问的 APScheduler 作业，应应用 close_old_connections 装饰器。这样做可以确保在您的作业运行前后强制执行 Django 的 CONN_MAX_AGE 配置设置。这反映了在处理每个 HTTP 请求前后执行相同操作的标准的 Django 功能。

如果您仍然遇到各种类型的“丢失数据库连接”错误，那么这可能意味着：

• 数据库连接在执行作业中途超时。您可能需要考虑将连接池器作为您部署的一部分，以实现更稳健的数据库连接管理（例如，PostgreSQL 的 pgbouncer 或其他 DB 平台的等效产品）。
• 您的数据库服务器已崩溃/已重启。Django 不会自动重新连接，您还需要重新启动 django-apscheduler。

常见陷阱

除非您有一组非常具体的严格要求，并且对 APScheduler 的内部工作原理有深入了解，否则您真的不应该使用 BackgroundScheduler。这样做可能会导致各种诱惑，如

• 在 Django 视图中启动调度器：这很可能会导致同时运行多个调度器，并导致作业多次运行（有关此主题的更详细说明，请参阅此 README 的上述介绍）。
• 在Django应用程序内部其他位置启动调度器：这似乎应该解决上述提到的问题，并保证只有一个调度器在运行。缺点是您已经将所有后台任务处理线程的管理委托给了您正在使用的任何web服务器（Gunicorn、uWSGI等）。web服务器可能会极端地终止任何长时间运行的线程（您的作业），认为它们是由不良的HTTP请求引起的。

依赖于 BlockingScheduler 会强制您在独立的进程中运行 APScheduler，该进程不受或不受web服务器的处理或监控。上面提供的 runapscheduler.py 示例代码是一个很好的起点。

项目资源

• 变更日志
• 发布程序