Celery 之前有一個 --autoreload 標誌，現已刪除。然而，Django 在其manage.py runserver 指令中內建了自動重新載入功能。 Celery Workers 中缺乏自動重新加載會造成令人困惑的開發體驗：更新 Python 程式碼會導致 Django 伺服器使用當前程式碼重新加載，但伺服器觸發的任何任務都將在 Celery Workers 中運行過時的程式碼。
這篇文章將向您展示如何建立自訂的 manage.py runworker 命令，該命令會在開發過程中自動重新載入 Celery 工作執行緒。這個指令將模仿 runserver，我們將看看 Django 的自動重新載入是如何在幕後工作的。
這篇文章假設您有一個已經安裝了 Celery 的 Django 應用程式（指南）。它還假設您了解 Django 中的項目和應用程式之間的差異。
所有原始程式碼和文件連結都適用於發佈時（2024 年 7 月）目前版本的 Django 和 Celery。如果你在遙遠的將來讀到這篇文章，事情可能已經改變了。
最後，主項目目錄將在帖子的範例中命名為 my_project。
我們將建立一個名為 runworker 的自訂管理.py 指令。因為 Django 透過其 runsever 命令提供自動重新加載，所以我們將使用 runserver 的原始程式碼作為我們自訂命令的基礎。
您可以透過在專案的任何應用程式中建立 management/commands/ 目錄來在 Django 中建立命令。建立目錄後，您可以在該目錄中放置一個帶有您要建立的命令名稱的 Python 檔案（文件）。
假設您的專案有一個名為 polls 的應用程序，我們將在 polls/management/commands/runworker.py 中建立一個檔案並添加以下程式碼：
重要提示： 請務必將 my_project 的所有實例替換為您的 Django 專案的名稱。
如果您想複製並貼上此程式碼並繼續編程，您可以安全地停在這裡，而無需閱讀本文的其餘部分。這是一個優雅的解決方案，將在您開發 Django 和 Celery 專案時為您提供良好的服務。但是，如果您想了解更多有關其工作原理的信息，請繼續閱讀。
我不會逐行查看此程式碼，而是按主題討論最有趣的部分。如果您還不熟悉 Django 自訂命令，您可能需要在繼續之前查看文件。
這部分感覺最神奇。在指令的handle()方法體內，呼叫了Django的內部autoreload.run_with_reloader()。它接受一個回調函數，每次專案中的 Python 檔案發生變更時都會執行該函數。 實際上是如何運作的？
讓我們來看看 autoreload.run_with_reloader() 函數原始碼的簡化版本。簡化的函數會重寫、內聯和刪除程式碼，使其操作更加清晰。
當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)。這將退出子流程，從而觸發生成子流程的程式碼的下一次循環迭代。反過來，會啟動一個使用更新版本程式碼的新子流程。
By default, Django commands perform system checks before they run their handle() method. However, in the case of runserver & our custom runworker command, we'll want to postpone running these until we're inside the callback that we provide to run_with_reloader(). In our case, this is our run_worker() method. This allows us to run the command with automatic reloading while fixing broken system checks.
To postpone running the system checks, the value of the requires_system_checks attribute is set to an empty list, and the checks are performed by calling self.check() in the body of run_worker(). Like runserver, our custom runworker command also checks to see if all migrations have been run, and it displays a warning if there are pending migrations.
Because we're already performing Django's system checks within the run_worker() method, we disable the system checks in Celery by passing it the --skip-checks flag to prevent duplicate work.
All of the code that pertains to system checks and migrations was lifted directly from the runserver command source code.
Our implementation launches the Celery worker directly from Python using celery_app.worker_main() rather than shelling out to Celery.
This code executes when the worker is initialized, displaying the date & time, Django version, and the command to quit. It is modeled after the information that displays when runserver boots.
The following lines were also lifted from the runserver source:
Our custom command has a configurable log level incase the developer wants to adjust the setting from the CLI without modifying the code.
I poked-and-prodded at the source code for Django & Celery to build this implementation, and there are many opportunities to extend it. You could configure the command to accept more of Celery's worker arguments. Alternatively, you could create a custom manage.py command that automatically reloads any shell command like David Browne did in this Gist.
If you found this useful, feel free to leave a like or a comment. Thanks for reading.
以上是使用自訂 Django 命令自動重新載入 Celery 工作線程的詳細內容。更多資訊請關注PHP中文網其他相關文章！
