Celery add_periodic_task阻止Django在uwsgi环境中运行

Question

我编写了一个模块，该模块根据项目设置中的字典列表(通过django.conf.settings导入)动态添加周期性的芹菜任务。我使用函数add_tasks来实现这一点，该函数使用设置中给定的特定uuid来调度要调用的函数：

def add_tasks(celery):
for new_task in settings.NEW_TASKS:
celery.add_periodic_task(
new_task['interval'],
my_task.s(new_task['uuid']),
name='My Task %s' % new_task['uuid'],
)

就像这里建议的那样，我使用on_after_configure.connect信号来调用celery.py:中的函数

app = Celery('my_app')
@app.on_after_configure.connect
def setup_periodic_tasks(celery, **kwargs):
from add_tasks_module import add_tasks
add_tasks(celery)

此设置适用于celery beat和celery worker，但破坏了我使用uwsgi为django应用程序提供服务的设置。Uwsgi平稳运行，直到视图代码第一次使用celener的.delay()方法发送任务为止。在这一点上，芹菜似乎是在uwsgi中初始化的，但在上面的代码中永远阻塞。如果我从命令行手动运行它，然后在它阻塞时中断，我会得到以下(缩短的)堆栈跟踪：

Traceback (most recent call last):
File "/usr/local/lib/python3.6/site-packages/kombu/utils/objects.py", line 42, in __get__
return obj.__dict__[self.__name__]
KeyError: 'tasks'
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/usr/local/lib/python3.6/site-packages/kombu/utils/objects.py", line 42, in __get__
return obj.__dict__[self.__name__]
KeyError: 'data'
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/usr/local/lib/python3.6/site-packages/kombu/utils/objects.py", line 42, in __get__
return obj.__dict__[self.__name__]
KeyError: 'tasks'
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
(SHORTENED HERE. Just contained the trace from the console through my call to this function)
File "/opt/my_app/add_tasks_module/__init__.py", line 42, in add_tasks
my_task.s(new_task['uuid']),
File "/usr/local/lib/python3.6/site-packages/celery/local.py", line 146, in __getattr__
return getattr(self._get_current_object(), name)
File "/usr/local/lib/python3.6/site-packages/celery/local.py", line 109, in _get_current_object
return loc(*self.__args, **self.__kwargs)
File "/usr/local/lib/python3.6/site-packages/celery/app/__init__.py", line 72, in task_by_cons
return app.tasks[
File "/usr/local/lib/python3.6/site-packages/kombu/utils/objects.py", line 44, in __get__
value = obj.__dict__[self.__name__] = self.__get(obj)
File "/usr/local/lib/python3.6/site-packages/celery/app/base.py", line 1228, in tasks
self.finalize(auto=True)
File "/usr/local/lib/python3.6/site-packages/celery/app/base.py", line 507, in finalize
with self._finalize_mutex:

获取互斥对象似乎有问题。

目前，我正在使用一种变通方法来检测sys.argv[0]是否包含uwsgi，然后不添加定期任务，因为只有beat需要这些任务，但我想了解这里出了什么问题，以更永久地解决问题。

这个问题是否与使用uwsgi多线程或多处理有关，其中一个线程/进程持有另一个所需的互斥体？

如果有任何能帮我解决这个问题的提示，我将不胜感激。非常感谢。

我使用的是：Django 1.11.7和Celery 4.1.0

编辑1

我为这个问题创建了一个最小的设置：

celery.py:

import os
from celery import Celery
from django.conf import settings
from myapp.tasks import my_task
# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'my_app.settings')
app = Celery('my_app')
@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
sender.add_periodic_task(
60,
my_task.s(),
name='Testtask'
)
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

tasks.py:

from celery import shared_task
@shared_task()
def my_task():
print('ran')

请确保CELERY_TASK_ALWAYS_EAGER=False并且您有一个工作消息队列。

运行：

./manage.py shell -c 'from myapp.tasks import my_task; my_task.delay()'

请等待大约10秒钟，然后中断以查看上述错误。

Answer 1

所以，我发现@shared_task装饰器造成了这个问题。当我在信号调用的函数中声明任务权限时，我可以绕过这个问题，如下所示：

def add_tasks(celery):
@celery.task
def my_task(uuid):
print(uuid)
for new_task in settings.NEW_TASKS:
celery.add_periodic_task(
new_task['interval'],
my_task.s(new_task['uuid']),
name='My Task %s' % new_task['uuid'],
)

这个解决方案实际上对我有效，但我还有一个问题：我在一个可插入的应用程序中使用了这个代码，所以我不能在信号处理程序之外直接访问芹菜应用程序，但也希望能够从其他代码中调用my_task函数。通过在函数中定义它，它在函数之外是不可用的，所以我不能将它导入其他任何地方。

我可以通过在信号函数之外定义任务函数来解决这个问题，并在这里和tasks.py中与不同的装饰器一起使用它。不过，我想知道除了@shared_task装饰器之外，是否还有一个装饰器可以在tasks.py中使用，而不会产生问题。

目前最好的解决方案可能是：

task_app__init__.py:

def my_task(uuid):
# do stuff
print(uuid)
def add_tasks(celery):
celery_my_task = celery.task(my_task)
for new_task in settings.NEW_TASKS:
celery.add_periodic_task(
new_task['interval'],
celery_my_task(new_task['uuid']),
name='My Task %s' % new_task['uuid'],
)

task_app.tasks.py:

from celery import shared_task
from task_app import my_task
shared_my_task = shared_task(my_task)

myapp.celery.py:

import os
from celery import Celery
from django.conf import settings

# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'my_app.settings')
app = Celery('my_app')
@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
from task_app import add_tasks
add_tasks(sender)

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

Answer 2

你能试试那个信号@app.on_after_finalize.connect:吗

工作项目celery==4.1.0、Django==2.0、django-celery-beat==1.1.0和django-celery-results==1.0.1中的一些快速片段

@app.on_after_finalize.connect
def setup_periodic_tasks(sender, **kwargs):
""" setup of periodic task :py:func:shopify_data_fetcher.celery.fetch_shopify
based on the schedule defined in: settings.CELERY_BEAT_SCHEDULE
"""
for task_name, task_config in settings.CELERY_BEAT_SCHEDULE.items():
sender.add_periodic_task(
task_config['schedule'],
fetch_shopify.s(**task_config['kwargs']['resource_name']),
name=task_name
)

CELERY_BEAT_SCHEDULE:

CELERY_BEAT_SCHEDULE = {
'fetch_shopify_orders': {
'task': 'shopify.tasks.fetch_shopify',
'schedule': crontab(hour="*/3", minute=0),
'kwargs': {
'resource_name': shopify_constants.SHOPIFY_API_RESOURCES_ORDERS
}
}
}