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

问题描述 投票：5回答：2

我编写了一个模块，根据项目设置中的字典列表（通过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'],
        )

像建议的here我使用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运行顺利，直到第一次使用celery的.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.朋友:

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.朋友:

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秒钟以查看上述错误。

python django celery celery-task celerybeat
2个回答
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__.朋友:

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.朋友:

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

没有app.celery.朋友:

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)
0
投票

你能尝试一下信号@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
        }
    }
}
最新问题
© www.soinside.com 2019 - 2024. All rights reserved.