使用自定义 Django 命令自动重新加载 Celery 工作线程

Celery 之前有一个 --autoreload 标志，现已被删除。然而，Django 在其manage.py runserver 命令中内置了自动重新加载功能。 Celery 工作线程中缺乏自动重新加载会造成令人困惑的开发体验：更新 Python 代码会导致 Django 服务器使用当前代码重新加载，但服务器触发的任何任务都将在 Celery 工作线程中运行过时的代码。

这篇文章将向您展示如何构建一个自定义的 manage.py runworker 命令，该命令在开发过程中自动重新加载 Celery 工作线程。该命令将模仿 runserver，我们将看看 Django 的自动重新加载是如何在幕后工作的。

在我们开始之前

这篇文章假设您有一个已经安装了 Celery 的 Django 应用程序（指南）。它还假设您了解 Django 中的项目和应用程序之间的差异。

所有源代码和文档链接均适用于发布时（2024 年 7 月）当前版本的 Django 和 Celery。如果您在遥远的将来阅读本文，事情可能会发生变化。

最后，主项目目录将在帖子的示例中命名为 my_project。

解决方案：自定义命令

我们将创建一个名为 runworker 的自定义管理.py 命令。由于 Django 通过其 runsever 命令提供自动重新加载，因此我们将使用 runserver 的源代码作为自定义命令的基础。

您可以通过在项目的任何应用程序中创建 management/commands/ 目录来在 Django 中创建命令。创建目录后，您可以在该目录 (docs) 中放置一个带有您要创建的命令名称的 Python 文件。

假设您的项目有一个名为 polls 的应用程序，我们将在 polls/management/commands/runworker.py 中创建一个文件并添加以下代码：

# polls/management/commands/runworker.py

import sys
from datetime import datetime

from celery.signals import worker_init

from django.conf import settings
from django.core.management.base import BaseCommand
from django.utils import autoreload

from my_project.celery import app as celery_app


class Command(BaseCommand):
    help = "Starts a Celery worker instance with auto-reloading for development."

    # Validation is called explicitly each time the worker instance is reloaded.
    requires_system_checks = []
    suppressed_base_arguments = {"--verbosity", "--traceback"}

    def add_arguments(self, parser):
        parser.add_argument(
            "--skip-checks",
            action="store_true",
            help="Skip system checks.",
        )
        parser.add_argument(
            "--loglevel",
            choices=("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL", "FATAL"),
            type=str.upper,  # Transforms user input to uppercase.
            default="INFO",
        )

    def handle(self, *args, **options):
        autoreload.run_with_reloader(self.run_worker, **options)

    def run_worker(self, **options):
        # If an exception was silenced in ManagementUtility.execute in order
        # to be raised in the child process, raise it now.
        autoreload.raise_last_exception()

        if not options["skip_checks"]:
            self.stdout.write("Performing system checks...\n\n")
            self.check(display_num_errors=True)

        # Need to check migrations here, so can't use the
        # requires_migrations_check attribute.
        self.check_migrations()

        # Print Django info to console when the worker initializes.
        worker_init.connect(self.on_worker_init)

        # Start the Celery worker.
        celery_app.worker_main(
            [
                "--app",
                "my_project",
                "--skip-checks",
                "worker",
                "--loglevel",
                options["loglevel"],
            ]
        )

    def on_worker_init(self, sender, **kwargs):
        quit_command = "CTRL-BREAK" if sys.platform == "win32" else "CONTROL-C"

        now = datetime.now().strftime("%B %d, %Y - %X")
        version = self.get_version()
        print(
            f"{now}\n"
            f"Django version {version}, using settings {settings.SETTINGS_MODULE!r}\n"
            f"Quit the worker instance with {quit_command}.",
            file=self.stdout,
        )

重要提示： 请务必将 my_project 的所有实例替换为您的 Django 项目的名称。

如果您想复制并粘贴此代码并继续编程，您可以安全地停在此处，而无需阅读本文的其余部分。这是一个优雅的解决方案，将在您开发 Django 和 Celery 项目时为您提供良好的服务。但是，如果您想了解更多有关其工作原理的信息，请继续阅读。

它是如何工作的（可选）

我不会逐行查看此代码，而是按主题讨论最有趣的部分。如果您还不熟悉 Django 自定义命令，您可能需要在继续之前查看文档。

自动装弹

这部分感觉最神奇。在命令的handle()方法体内，调用了Django的内部autoreload.run_with_reloader()。它接受一个回调函数，每次项目中的 Python 文件发生更改时都会执行该函数。 实际上是如何运​​作的？

让我们看一下 autoreload.run_with_reloader() 函数源代码的简化版本。简化的函数重写、内联和删除代码，以使其操作更加清晰。

# NOTE: This has been dramatically pared down for clarity.

def run_with_reloader(callback_func, *args, **kwargs):
    # NOTE: This will evaluate to False the first time it is run.
    is_inside_subprocess = os.getenv("RUN_MAIN") == "true"

    if is_inside_subprocess:
        # The reloader watches for Python file changes.
        reloader = get_reloader()

        django_main_thread = threading.Thread(
            target=callback_func, args=args, kwargs=kwargs
        )
        django_main_thread.daemon = True
        django_main_thread.start()

        # When the code changes, the reloader exits with return code 3.
        reloader.run(django_main_thread)

    else:
        # Returns Python path and the arguments passed to the command.
        # Example output: ['/path/to/python', './manage.py', 'runworker']
        args = get_child_arguments()

        subprocess_env = {**os.environ, "RUN_MAIN": "true"}
        while True:
            # Rerun the manage.py command in a subprocess.
            p = subprocess.run(args, env=subprocess_env, close_fds=False)
            if p.returncode != 3:
                sys.exit(p.returncode)

当manage.py runworker在命令行中运行时，它会首先调用handle()方法，该方法会调用run_with_reloader()。

在 run_with_reloader() 内部，它将检查名为 RUN_MAIN 的环境变量是否具有“true”值。当函数第一次被调用时，RUN_MAIN 应该没有值。

当 RUN_MAIN 未设置为“true”时，run_with_reloader() 将进入循环。在循环内，它将启动一个子进程，重新运行传入的manage.py [command_name]，然后等待该子进程退出。如果子进程退出并返回代码 3，则循环的下一次迭代将启动一个新的子进程并等待。该循环将一直运行，直到子进程返回不为 3 的退出代码（或直到用户使用 ctrl + c 退出）。一旦得到非3的返回码，就会完全退出程序。

生成的子进程再次运行manage.py命令（在我们的例子中是manage.py runworker），并且该命令将再次调用run_with_reloader()。这次，RUN_MAIN 将设置为“true”，因为该命令在子进程中运行。

现在 run_with_reloader() 知道它位于子进程中，它将获得一个监视文件更改的重新加载器，将提供的回调函数放入线程中，并将其传递给开始监视更改的重新加载器。

当重新加载器检测到文件更改时，它会运行 sys.exit(3)。这将退出子流程，从而触发生成子流程的代码的下一次循环迭代。反过来，会启动一个使用更新版本代码的新子流程。

系统检查和迁移

默认情况下，Django 命令在运行其handle() 方法之前执行系统检查。但是，对于 runserver 和我们的自定义 runworker 命令，我们希望推迟运行这些命令，直到进入我们提供给 run_with_reloader() 的回调中。在我们的例子中，这是我们的 run_worker() 方法。这使我们能够运行自动重新加载的命令，同时修复损坏的系统检查。

为了推迟运行系统检查，需要将requires_system_checks属性的值设置为空列表，并通过在run_worker()主体中调用self.check()来执行检查。与 runserver 一样，我们的自定义 runworker 命令也会检查是否所有迁移都已运行，如果有待处理的迁移，它会显示警告。

因为我们已经在 run_worker() 方法中执行 Django 的系统检查，所以我们通过向 Celery 传递 --skip-checks 标志来禁用系统检查，以防止重复工作。

所有与系统检查和迁移相关的代码都是直接从 runserver 命令源代码中提取的。

celery_app.worker_main()

我们的实现使用 celery_app.worker_main() 直接从 Python 启动 Celery 工作程序，而不是向 Celery 发起攻击。

on_worker_init()

此代码在工作进程初始化时执行，显示日期和时间、Django 版本以及退出命令。它是根据 runserver 启动时显示的信息建模的。

其他 runserver 样板

以下行也从 runserver 源代码中删除：

• suppressed_base_arguments = {"--verbosity", "--traceback"}
• autoreload.raise_last_exception()

日志级别

我们的自定义命令具有可配置的日志级别，以防开发人员希望在不修改代码的情况下从 CLI 调整设置。

更进一步

我研究了 Django 和 Celery 的源代码来构建这个实现，并且有很多扩展它的机会。您可以配置该命令以接受更多 Celery 的工作参数。或者，您可以创建一个自定义的 manage.py 命令，它会自动重新加载任何 shell 命令，就像 David Browne 在本要点中所做的那样。

如果您觉得本文有用，请随时留下点赞或评论。感谢您的阅读。

以上是使用自定义 Django 命令自动重新加载 Celery 工作线程的详细内容。更多信息请关注PHP中文网其他相关文章！