环境

  • celery==4.4.6
  • python==3.8.3
  • redis==3.3.8

Celery异步任务配置

  • 在setting.py的同级目录下(例如:项目为名site,则目录为site/site/)创建celery.py文件,这个文件用来生成celery的实例
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
from django.conf import settings


os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'site.settings')
app = Celery('site',broker = 'redis://localhost:6379/0',backend='redis://localhost')
app.config_from_object('django.conf:settings',namespace='CELERY')
#下面这个设置就是对所有django的app进行监听
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
  • 在你自己创建的Django项目的app中,新建tasks.py,celery会自动监听app下面的tasks中的任务
  • 在tasks.py中定义一个任务
from __future__ import absolute_import, unicode_literals
from site.celery import app
from celery import shared_task
import time


@shared_task
def waste_time():
    time.sleep(3)
    return "Run function 'waste_time' finished."

@task和@shared_task的区别:当我们使用task装饰器定义我们的异步任务时,那么这个任务依赖于根据项目名生成的celery实例,然而我们在进行Django开发时为了保证每个app的可重用性,我们经常会在每个app文件夹下编写异步任务,这些任务并不依赖于具体的Django项目名。使用@shared_task 装饰器能让我们避免对某个项目名对应Celery实例的依赖,使app的可移植性更强

  • 在项目的__init__.py文件中加入以下内容,保证在启动Django时把celery也启动
from __future__ import absolute_import, unicode_literals
from .celery import app as celery_app


__all__ = ("celery_app", )
  • 先启动Django项目python manage.py runserver,再激活celery的workercelery -A site worker -l info就可以了

Celery定时任务配置

在异步任务中我们只用到了worker,而在定时任务中,还需用到celery中的beat调度器

  • 在celery.py里面添加任务
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
from django.conf import settings
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'site.settings')


app = Celery('site',broker = 'redis://localhost:6379/0',backend='redis://localhost')
app.config_from_object('django.conf:settings',namespace='CELERY')

#下面的设置就是关于调度器beat的设置
app.conf.beat_schedule ={
        'test':{                  # 取个名字
            'task':'stats.tasks.test',    # 设置是要将哪个任务进行定时
            'schedule':crontab(hour=20, minute=47),   # 调用crontab进行具体时间的定义
        },
}

# 下面这个设置就是对所有django的app进行监听
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
  • 在settings.py中配置时区
CELERY_TIME_ZONE = 'Asia/Shanghai'
CELERY_ENABLE_UTC = True
  • 具体的任务
@shared_task
def test():
    print ('test?')
    return 'hello'
  • 启动项目后,使用以下命令同时激活worker和beat
celery -A site worker -B -l info

版权声明:如无特殊说明,文章均为本站原创,转载请注明出处

本文链接:https://www.ltfred.com/article/django-celery/