如何在django中设置定时任务？

BigYoung小站

发布于 2020-05-04 21:31:31

2.5K0

发布于 2020-05-04 21:31:31

文章被收录于专栏：BigYoung小站BigYoung小站

Django 作为后端Web开发框架，有时候我们需要用到定时任务来或者固定频次的任务来执行某段代码，这时我们就要用到Celery了。Django中有一个中间件：Django-celery

环境：

Python 3.6

Django为小于1.8版本

Celery为3.1版本

第一步安装：django-celery

pip install django-celery

第二步：配置celery和任务

创建测试django环境：

django-admin.py createproject test
django-admin.py startapp demo

创建好的项目布局如下：

- proj/
- manage.py
- proj/
- __init__.py
- celery.py
- settings.py
- urls.py
- demo/
- migrations
- __init__.py
- admin.py
- apps.py
- models.py
- tasks.py
- tests.py
- views.py

2.1 配置celery.py文件

需要替换的内容，我都在对应的行后提示了，剩下的内容默认就好创建test/test/celery.py文件，内容如下：

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery


# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')    # “proj.settings”替换为你的项目信息：test.settings


app = Celery('proj') # 这里的proj替换为你的项目名称:test


# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
#   should have a `CELERY_` prefix.
app.config_from_object('django.conf:settings', namespace='CELERY')


# Load task modules from all registered Django app configs.
app.autodiscover_tasks()




@app.task(bind=True)
def debug_task(self):
print('Request: {0!r}'.format(self.request))

2.2 配置项目的init.py中配置celery内容

打开test/test/_init.py文件，添加内容：

from __future__ import absolute_import, unicode_literals


# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from .celery import app as celery_app


__all__ = ('celery_app',)

2.3 在task.py中添加计划任务

编辑test/demo/task.py文件，添加计划任务，内容如下：

# Create your tasks here
from __future__ import absolute_import, unicode_literals
from celery import shared_task




@shared_task
def add(x, y):
return x + y




@shared_task
def mul(x, y):
return x * y




@shared_task
def xsum(numbers):
return sum(numbers)

第三步：任务执行

运行django项目：python manage.py runserver

3.1 后台添加计划任务

访问“http://localhost:8000/admin/”，在celery的管理页面里，选择Periodic tasks，进行任务添加。选择对应的任务，设置定时或者周期时间

3.2 启动定时的celery服务

注意：celery依赖redis服务，需要提前运行redis服务：`redis-server`

# 以下两个命令在不同的shell窗口里执行，需要在django的目录下
python manager.py celery beat  -l info   #接收定时任务的命令
python manager.py celery worker -l info  #执行定时任务的命令，此shell窗口会看到任务的输入信息

3.3 启动单次的celery服务

注意：celery依赖redis服务，需要提前运行redis服务：`redis-server`

python manager.py shell    # 进到django的shell里
from demo.task import mul, xsum    # 导入task任务
a = mul()
b = xsum()
# 执行a， b会输出信息
a(1,2)
b(1)

本文参与腾讯云自媒体分享计划，分享自微信公众号。

原始发表：2019-01-16，如有侵权请联系 cloudcommunity@tencent.com 删除

python

django

分布式

本文分享自 BigYoung小站微信公众号，前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体分享计划，欢迎热爱写作的你一起参与！

python

django

分布式

登录后参与评论

0 条评论

热度