python django与celery的集成

2023-01-25 16:46:43

一、celery与django

关于celery介绍和使用可以查看上篇Python中任务队列-芹菜celery的使用

关于django的介绍和使用可查看python django框架+vue.js前后端分离

我来看一下celery集成到django后的整个工作链：django将任务转发给消息队列，celery读取到任务后执行并将结果通过django ORM 存储。

在本文中两个插件：

使用django-celery-results将celery处理结果进行ORM存储
使用django-celery-beat 对任务管理和周期调度

二、基本使用

1.创建一个django项目,不再赘述

建好后目录看起来是这样的

django_celery
├── celery_app
│   ├── __init__.py
│   ├── apps.py
│   ├── migrations/
│   ├── models.py
│   ├── admin.py
│   └── views.py
├── manage.py
├── django_celery
│   ├── __init__.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
└── env/
│── db.sqlite3

安装插件

pip install django-celery-results
pip install django-celery-beat

在项目settings.py我们会安装appcelery_app、django_celery_results、django_celery_beat并开启我们的admin后台用于查看和管理我们的任务调度。

2.添加任务tasks.py

我们在celery_app中添加任务文件tasks.py其中包含上篇文章中单args_add1任务，并通过shared_task进行装饰。

from __future__ import absolute_import
from celery import shared_task
import time

@shared_task
def args_add1(x,y):
    print("start task no.1 now!")
    time.sleep(10)
    print("task no.1 end!")
    return x+y

3.配置celery应用

在django_celery目录下添加celery.py用于创建我们的celery应用

import os
from celery import Celery
#加载配置
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'django_celery.settings')
#创建celery app
app = Celery('django_celery')
app.config_from_object('django.conf:settings', namespace='CELERY')
#自动发现项目中的tasks
app.autodiscover_tasks()

在setting中添加celery相关配置，指定Broker和Backend。这里我们延用上篇的rabbitmq作为Broker，后端配置为django-db

CELERY_BROKER_URL = 'amqp://rabbit_user:rabbit_pass@i-k9pwet2d/rabbit_vhost'

CELERY_RESULT_BACKEND = 'django-db'

我们的celery应用配置好了，然后我们要启动它，在__init__.py中加入我们的app

from .celery import app as celery_app

__all__ = ['celery_app']

4.配置django

通过上面的配置我们有了celery应用，也有了任务，接下来我们配置一个URL请求来发送我们的任务给celery。

对我们app下views.py编写如下

from django.http import JsonResponse
from celery_app import tasks
from celery.result import AsyncResult

# Create your views here.

def celery(request,*args,**kwargs):
    res=tasks.args_add1.delay(123,456) #发送任务给celery  
    result = AsyncResult(res.task_id)
    return JsonResponse({'status':result.status,'task_id':result.task_id})

配置路由

from django.contrib import admin
from django.urls import path
import celery_app.views

urlpatterns = [
    path('admin/', admin.site.urls),
    path('celery/',celery_app.views.celery)
]

5.查看任务执行

通过上面配置我们基本款已经成了。我们启动django和celery worker看看

python manage.py migrate
python manage.py runserver

启动celery worker

celery -A django_celery worker --loglevel=info --concurrency=10

访问地址：http://127.0.0.1:8000/celery/ 发送我们的任务，此时任务已处于等待状态。

在worker终端上可以查看到任务正在被处理。

我们在django后台查看task执行记录和细节

三、周期性任务调度

在前面我们已经安装了django-celery-beat

我们在后台Periodic tasks中配置我们的周期性任务，也可以通过Crontab Schedule来配置计划任务。

我们配置一个每5小时执行一次的任务。

启动调度器,这样我们的任务就会在后台默默无闻的周期性工作啦。

celery -A django_celery beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler

以上是celery集成到django中的基础使用。

文章有不足的地方欢迎指出。

欢迎收藏、点赞、提问。关注*饮水机管理员，除了管烧热水，有时还做点别的。

码农公寓