如何阻止Celery运行Django应用内的自定义Python线程

现有基于Django开发的应用，使用Celery搭配Redis实现后台任务处理，同时在应用内编写了独立Python线程执行周期性检查逻辑。实际运行时发现启动Celery后，该自定义线程会作为Celery进程的一部分自动启动，需要规避该行为，保证线程仅作为Django应用组件运行，全局仅存在一个运行实例。

以下为复现问题的最小代码与运行输出：

• settings.py 配置片段

...
CELERY_BROKER_URL = 'redis://localhost:6379/1'

• 项目包下的__init__.py

from threading import Thread
from time import sleep
from .celery import app as celery_app

__all__ = ('celery_app',)

def mythread():
    while True:
      print("thread is running")
      sleep(10)

new_thread = Thread(target=mythread, daemon=True)
new_thread.start()

• 同目录下的celery.py

import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myapp.settings')

app = Celery('myapp')

app.config_from_object('django.conf:settings', namespace='CELERY')

app.autodiscover_tasks()

• Celery启动输出

...
[2022-06-15 16:21:49,923: INFO/MainProcess] Events of group {task} enabled by remote.
[2022-06-15 16:21:49,958: WARNING/MainProcess] thread is running
[2022-06-15 16:21:59,968: WARNING/MainProcess] thread is running

问题根源 #

Celery启动时会完整加载Django应用的运行上下文，因此会直接执行项目包__init__.py中写死的线程启动逻辑，导致自定义线程在Celery进程中被拉起。

可通过以下两种方式解决，推荐使用第二种规范性更强的方案：

方案1：通过进程启动参数判断进程类型 #

Celery启动的进程命令中会包含celery关键字，可以在启动线程前增加判断，仅在非Celery进程中拉起线程。
修改__init__.py代码如下：

import sys
from threading import Thread
from time import sleep
from .celery import app as celery_app

__all__ = ('celery_app',)

def mythread():
    while True:
      print("thread is running")
      sleep(10)

# 非celery进程才启动线程
if 'celery' not in sys.argv[0]:
    new_thread = Thread(target=mythread, daemon=True)
    new_thread.start()

该判断同时覆盖Celery worker、Celery beat进程，避免线程在两类Celery进程中启动。

方案2：通过AppConfig钩子+环境变量控制启动（推荐） #

不要将线程启动逻辑写在包的__init__.py中（包导入时直接执行逻辑容易产生不可预期的副作用），将逻辑迁移到Django AppConfig的ready()钩子中，通过环境变量精确区分进程类型，同时增加重复启动校验，保证线程全局单实例：

1. 修改celery.py，在初始化Celery前设置进程标记环境变量：

import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myapp.settings')
# 标记当前进程为Celery进程
os.environ['IS_CELERY_PROCESS'] = '1'

app = Celery('myapp')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

2. 在对应应用的apps.py中编写线程启动逻辑，放入ready()方法：

import os
from django.apps import AppConfig
from threading import Thread
from time import sleep

def mythread():
    while True:
        print("thread is running")
        sleep(10)

class MyAppConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'myapp'

    def ready(self):
        # 跳过Celery进程
        if os.getenv('IS_CELERY_PROCESS') == '1':
            return
        # 增加启动标记，避免Django开发模式自动重载导致线程重复启动
        if not os.environ.get('MY_THREAD_ALREADY_STARTED'):
            os.environ['MY_THREAD_ALREADY_STARTED'] = '1'
            new_thread = Thread(target=mythread, daemon=True)
            new_thread.start()

3. 清理__init__.py，移除所有线程相关代码，仅保留Celery导入逻辑：

from .celery import app as celery_app

__all__ = ('celery_app',)

该方案可以同时解决两个问题：

• 彻底阻止自定义线程在Celery进程中启动
• 避免Django开发模式下自动重载导致的线程重复启动，保证全局仅存在一个线程实例

内容的提问来源于stack exchange，提问作者michael