Django配置celery(非djcelery)试行异步任务和定期职

作者:编程技术

二、配置使用

  celery很容易集成到Django框架中,当然如果想要实现定时任务的话还需要安装django-celery-beta插件,后面会说明。需要注意的是Celery4.0只支持Django版本>=1.8的,如果是小于1.8版本需要使用Celery3.1。

网上有很多celery django实现定时任务的教程,不过它们大多数是基于djcelery celery3的;
或者是使用django_celery_beat配置较为繁琐的。

3.celery不能用root用户启动的话需要在主配置文件中添加 platforms.C_FORCE_ROOT = True

扩展

  除了redis、rabbitmq能做结果存储外,还可以使用Django的orm作为结果存储,当然需要安装依赖插件,这样的好处在于我们可以直接通过django的数据查看到任务状态,同时为可以制定更多的操作,下面介绍如何使用orm作为结果存储。

1.安装

pip install django-celery-results

2.配置settings.py,注册app

INSTALLED_APPS = (
    ...,
    'django_celery_results',
)

4.修改backend配置,将redis改为django-db

#CELERY_RESULT_BACKEND = 'redis://10.1.210.69:6379/0' # BACKEND配置,这里使用redis

CELERY_RESULT_BACKEND = 'django-db'  #使用django orm 作为结果存储

5.修改数据库

python3 manage.py migrate django_celery_results

此时会看到数据库会多创建:

分分快三计划 1 当然你有时候需要对task表进行操作,以下源码的表结构定义:

class TaskResult(models.Model):
    """Task result/status."""

    task_id = models.CharField(_('task id'), max_length=255, unique=True)
    task_name = models.CharField(_('task name'), null=True, max_length=255)
    task_args = models.TextField(_('task arguments'), null=True)
    task_kwargs = models.TextField(_('task kwargs'), null=True)
    status = models.CharField(_('state'), max_length=50,
                              default=states.PENDING,
                              choices=TASK_STATE_CHOICES
                              )
    content_type = models.CharField(_('content type'), max_length=128)
    content_encoding = models.CharField(_('content encoding'), max_length=64)
    result = models.TextField(null=True, default=None, editable=False)
    date_done = models.DateTimeField(_('done at'), auto_now=True)
    traceback = models.TextField(_('traceback'), blank=True, null=True)
    hidden = models.BooleanField(editable=False, default=False, db_index=True)
    meta = models.TextField(null=True, default=None, editable=False)

    objects = managers.TaskResultManager()

    class Meta:
        """Table information."""

        ordering = ['-date_done']

        verbose_name = _('task result')
        verbose_name_plural = _('task results')

    def as_dict(self):
        return {
            'task_id': self.task_id,
            'task_name': self.task_name,
            'task_args': self.task_args,
            'task_kwargs': self.task_kwargs,
            'status': self.status,
            'result': self.result,
            'date_done': self.date_done,
            'traceback': self.traceback,
            'meta': self.meta,
        }

    def __str__(self):
        return '<Task: {0.task_id} ({0.status})>'.format(self)

 

  • 插件提供的定时任务管理将不在可用,当我们只需要任务定期执行而不需要人为调度的时候这点忽略不计。
  • 无法高效的管理或追踪定时任务,定时任务的跟踪其实交给日志更合理,但是对任务的修改就没有那么方便了,不过如果不需要经常变更/增减任务的话这点也在可接受范围内。

2.如果使用的不是rabbitmq做队列那么需要在主配置文件中 website/celery.py 配置broker和backend,如下:

定义与触发任务

  任务定义在每个tasks文件中,app01/tasks.py:

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

视图中触发任务

from django.http import JsonResponse
from app01 import tasks

# Create your views here.

def index(request,*args,**kwargs):
    res=tasks.add.delay(1,3)
    #任务逻辑
    return JsonResponse({'status':'successful','task_id':res.task_id})

访问

分分快三计划 2

 若想获取任务结果,可以通过task_id使用AsyncResult获取结果,还可以直接通过backend获取:

分分快三计划 3

 

当然使用原生方案的同时有几点插件所带来的好处被我们放弃了:

2.安装celery

一、前言

  Celery是一个基于python开发的分布式任务队列,如果不了解请阅读笔者上一篇博文Celery入门与进阶,而做python WEB开发最为流行的框架莫属Django,但是Django的请求处理过程都是同步的无法实现异步任务,若要实现异步任务处理需要通过其他方式(前端的一般解决方案是ajax操作),而后台Celery就是不错的选择。倘若一个用户在执行某些操作需要等待很久才返回,这大大降低了网站的吞吐量。下面将描述Django的请求处理大致流程(图片来源于网络):

分分快三计划 4

请求过程简单说明:浏览器发起请求-->请求处理-->请求经过中间件-->路由映射-->视图处理业务逻辑-->响应请求(template或response)

Celery定时任务配置

在进行配置前先来看看项目结构:

.├── linux_news│   ├── celery.py│   ├── __init__.py│   ├── settings.py│   ├── urls.py│   └── wsgi.py├── manage.py├── news│   ├── admin.py│   ├── apps.py│   ├── __init__.py│   ├── migrations│   ├── models│   ├── tasks.py│   ├── tests.py│   └── views└── start-celery.sh

其中news是我们的app,用于从一些rss订阅源获取新闻信息,linux_news则是我们的project。我们需要关心的主要是celery.py,settings.py,tasks.py和start-celery.sh。

首先是celery.py,想让celery执行任务就必须实例化一个celery app,并把settings.py里的配置传入app:

import osfrom celery import Celery# set the default Django settings module for the 'celery' program.os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'linux_news.settings')app = Celery('linux_news')# 'django.conf:settings'表示django,conf.settings也就是django项目的配置,celery会根据前面设置的环境变量自动查找并导入# - namespace表示在settings.py中celery配置项的名字的统一前缀,这里是'CELERY_',配置项的名字也需要大写app.config_from_object('django.conf:settings', namespace='CELERY')# Load task modules from all registered Django app configs.app.autodiscover_tasks()

配置就是这么简单,为了能在django里使用这个app,我们需要在__init__.py中导入它:

from .celery import app as celery_app

然后我们来看tasks.py,它应该位于你的app目录中,前面我们配置了自动发现,所以celery会自动找到这些tasks,我们的tasks将写在这一模块中,代码涉及了一些orm的使用,为了契合主题我做了些精简:

from linux_news.celery import celery_app as appfrom .models import *import timeimport feedparserimport pytzimport html@app.task(ignore_result=True)def fetch_news(origin_name):    """    fetch all news from origin_name    """    origin = get_feeds_origin(origin_name)    feeds = feedparser.parse(origin.feed_link)    for item in feeds['entries']:        entry = NewsEntry()        entry.title = item.title        entry.origin = origin        entry.author = item.author        entry.link = item.link        # add timezone        entry.publish_time = item.time.replace(tzinfo=pytz.utc)        entry.summary = html.escape(item.summary)        entry.save()@app.task(ignore_result=True)def fetch_all_news():    """    这是我们的定时任务    fetch all origins' news to db    """    origins = NewsOrigin.objects.all()    for origin in origins:        fetch_news.delay(origin.origin_name)

tasks里是一些耗时操作,比如网络IO或者数据库读写,因为我们不关心任务的返回值,所以使用@app.task(ignore_result=True)将其屏蔽了。

任务配置完成后我们就要配置celery了,我们选择redis作为任务队列,我强烈建议在生产环境中使用rabbitmq或者redis作为任务队列或结果缓存后端,而不应该使用关系型数据库:

# redisREDIS_PORT = 6379REDIS_DB = 0# 从环境变量中取得redis服务器地址REDIS_HOST = os.environ.get('REDIS_ADDR', 'redis')# celery settings# 这两项必须设置,否则不能正常启动celery beatCELERY_ENABLE_UTC = TrueCELERY_TIMEZONE = TIME_ZONE# 任务队列配置CELERY_BROKER_URL = f'redis://{REDIS_HOST}:{REDIS_PORT}/{REDIS_DB}'CELERY_ACCEPT_CONTENT = ['application/json', ]CELERY_RESULT_BACKEND = f'redis://{REDIS_HOST}:{REDIS_PORT}/{REDIS_DB}'CELERY_TASK_SERIALIZER = 'json'

然后是我们的定时任务设置:

from celery.schedules import crontabCELERY_BEAT_SCHEDULE={        'fetch_news_every-1-hour': {            'task': 'news.tasks.fetch_all_news',            'schedule': crontab(minute=0, hour='*/1'),        }}

定时任务配置对象是一个dict,由任务名和配置项组成,主要配置想如下:

  • task:任务函数所在的模块,模块路径得写全,否则找不到将无法运行该任务
  • schedule:定时策略,一般使用celery.schedules.crontab,上面例子为每小时的0分执行一次任务,具体写法与linux的crontab类似可以参考文档说明
  • args:是个元组,给出任务需要的参数,如果不需要参数也可以不写进配置,就像例子中的一样
  • 其余配置项较少用,可以参考文档
    至此,配置celery beat的部分就结束了。

5.在 website/__init__.py 文件中增加如下内容,确保django启动的时候这个app能够被加载到

三、Django中使用定时任务

  如果想要在django中使用定时任务功能同样是靠beat完成任务发送功能,当在Django中使用定时任务时,需要安装django-celery-beat插件。以下将介绍使用过程。

启动celery beat

配置完成后只需要启动celery了。

启动之前配置一下环境。不要用root运行celery!不要用root运行celery!不要用root运行celery!重要的事情说三遍。

start-celery.sh:

export REDIS_ADDR=127.0.0.1celery -A linux_news worker -l info -B -f /path/to/log

-A 表示app所在的目录,-B表示启动celery beat运行定时任务。
celery正常启动后就可以通过日志来查看任务是否正常运行了:

[2018-12-21 13:00:00,022: INFO/MainProcess] Received task: news.tasks.fetch_all_news[e4566ede-2cfa-4c19-b2f3-0c7d6c38690d]  [2018-12-21 13:00:00,046: INFO/MainProcess] Received task: news.tasks.fetch_news[583e96dc-f508-49fa-a24a-331e0c07a86b]  [2018-12-21 13:00:00,051: INFO/ForkPoolWorker-2] Task news.tasks.fetch_all_news[e4566ede-2cfa-4c19-b2f3-0c7d6c38690d] succeeded in 0.02503809699555859s: None[2018-12-21 13:00:00,052: INFO/MainProcess] Received task: news.tasks.fetch_news[c61a3e55-dd3c-4d49-8d6d-ca9b1757db25]  [2018-12-21 13:00:00,449: INFO/ForkPoolWorker-5] Task news.tasks.fetch_news[c61a3e55-dd3c-4d49-8d6d-ca9b1757db25] succeeded in 0.39487219898728654s: None[2018-12-21 13:00:00,606: INFO/ForkPoolWorker-3] Task news.tasks.fetch_news[583e96dc-f508-49fa-a24a-331e0c07a86b] succeeded in 0.5523456179944333s: None

以上就是celery4运行定时任务的内容,如有错误和疏漏,欢迎指正。

from __future__ import absolute_import

# 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']

配置

  新建立项目taskproj,目录结构(每个app下多了个tasks文件,用于定义任务):

taskproj
├── app01
│   ├── __init__.py
│   ├── apps.py
│   ├── migrations
│   │   └── __init__.py
│   ├── models.py
│   ├── tasks.py
│   └── views.py
├── manage.py
├── taskproj
│   ├── __init__.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
└── templates

在项目目录taskproj/taskproj/目录下新建celery.py:

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# Author:wd
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery


os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'taskproj.settings')  # 设置django环境

app = Celery('taskproj')

app.config_from_object('django.conf:settings', namespace='CELERY') #  使用CELERY_ 作为前缀,在settings中写配置

app.autodiscover_tasks()  # 发现任务文件每个app下的task.py

taskproj/taskproj/__init__.py:

from __future__ import absolute_import, unicode_literals

from .celery import app as celery_app

__all__ = ['celery_app']

taskproj/taskproj/settings.py

CELERY_BROKER_URL = 'redis://10.1.210.69:6379/0' # Broker配置,使用Redis作为消息中间件

CELERY_RESULT_BACKEND = 'redis://10.1.210.69:6379/0' # BACKEND配置,这里使用redis

CELERY_RESULT_SERIALIZER = 'json' # 结果序列化方案

进入项目的taskproj目录启动worker:

celery worker -A taskproj -l debug

显然简洁而高效才是我们最终的追求,而celery4已经不需要额外插件即可与django结合实现定时任务了,原生的celery beat就可以很好的实现定时任务功能。

8.启动celery

二次开发

  django-celery-beat插件本质上是对数据库表变化检查,一旦有数据库表改变,调度器重新读取任务进行调度,所以如果想自己定制的任务页面,只需要操作beat插件的四张表就可以了。当然你还可以自己定义调度器,django-celery-beat插件已经内置了model,只需要进行导入便可进行orm操作,以下我用django reset api进行示例:

settings.py

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'app01.apps.App01Config',
    'django_celery_results',
    'django_celery_beat',
    'rest_framework',
]

urls.py

urlpatterns = [
    url(r'^admin/', admin.site.urls),
    url(r'^index$', views.index),
    url(r'^res$', views.get_res),
    url(r'^tasks$', views.TaskView.as_view({'get':'list'})),
]

views.py

from django_celery_beat.models import PeriodicTask  #倒入插件model
from rest_framework import serializers
from rest_framework import pagination
from rest_framework.viewsets import ModelViewSet
class Userserializer(serializers.ModelSerializer):
    class Meta:
        model = PeriodicTask
        fields = '__all__'

class Mypagination(pagination.PageNumberPagination):
    """自定义分页"""
    page_size=2
    page_query_param = 'p'
    page_size_query_param='size'
    max_page_size=4

class TaskView(ModelViewSet):
    queryset = PeriodicTask.objects.all()
    serializer_class = Userserializer
    permission_classes = []
    pagination_class = Mypagination

访问

分分快三计划 5

 

from __future__ import absolute_import
from celery import shared_task

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

安装配置

1.beat插件安装

pip3 install django-celery-beat

2.注册APP

INSTALLED_APPS = [
    ....   
    'django_celery_beat',
]

3.数据库变更

python3 manage.py migrate django_celery_beat

4.分别启动woker和beta

celery -A proj beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler  #启动beta 调度器使用数据库

celery worker -A taskproj -l info #启动woker

5.配置admin

urls.py

# urls.py
from django.conf.urls import url
from django.contrib import admin

urlpatterns = [
    url(r'^admin/', admin.site.urls),
]

6.创建用户

python3 manage.py createsuperuser 

7.登录admin进行管理(地址

分分快三计划 6

 

 使用示例:

分分快三计划 7

 

 

 

 

分分快三计划 8

 

 

 查看结果:

分分快三计划 9

 

# celery -A website beat -l info

9.这样在调用post这个方法时,里边的add就可以异步处理了

  • 名字为'sum-task'的task,每20秒执行一次add函数,并传了两个参数5和6
  • 名字为'send-report'的task,每周一早上4:30执行report函数

month_of_year
day_of_month
day_of_week
hour
minute

timedelta是datetime中的一个对象,需要 from datetime import timedelta 引入,有如下几个参数

from deploy.tasks import add

def post(request):
  result = add.delay(2, 3)


result.ready()
result.get(timeout=1)
result.traceback

定时任务的使用场景就很普遍了,比如我需要定时发送报告给老板~

注意tasks.py必须建在各app的根目录下,且只能叫tasks.py,不能随意命名

1.安装rabbitmq,这里我们使用rabbitmq作为broker,安装完成后默认启动了,也不需要其他任何配置

定时任务

所有演示均基于Django2.0

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持脚本之家。

定义了两个task:

website/
|-- deploy
|  |-- admin.py
|  |-- apps.py
|  |-- __init__.py
|  |-- models.py
|  |-- tasks.py
|  |-- tests.py
|  |-- urls.py
|  `-- views.py
|-- manage.py
|-- README
`-- website
  |-- celery.py
  |-- __init__.py
  |-- settings.py
  |-- urls.py
  `-- wsgi.py

分分快三计划 10

您可能感兴趣的文章:

  • 异步任务队列Celery在Django中的使用方法
  • Django使用Celery异步任务队列的使用
  • Django中使用celery完成异步任务的示例代码

6.各应用创建tasks.py文件,这里为 deploy/tasks.py

crontab的参数有:

  1. website/celery.py 文件添加如下配置以支持定时任务crontab
# celery -A website worker -l info
  • days
  • seconds
  • microseconds
  • milliseconds
  • minutes
  • hours

异步任务

7.views.py中引用使用这个tasks异步处理

# pip3 install celery
# redis做MQ配置
app = Celery('website', backend='redis', broker='redis://localhost')
# rabbitmq做MQ配置
app = Celery('website', backend='amqp', broker='amqp://admin:admin@localhost')
  • 消息队列broker:broker实际上就是一个MQ队列服务,可以使用redis、rabbitmq等作为broker
  • 处理任务的消费者workers:broker通知worker队列中有任务,worker去队列中取出任务执行,每一个worker就是一个进程
  • 存储结果的backend:执行结果存储在backend,默认也会存储在broker使用的MQ队列服务中,也可以单独配置用何种服务做backend

3.启动celery beat,celery启动了一个beat进程一直在不断的判断是否有任务需要执行

4.celery在长时间运行后可能出现内存泄漏,需要添加配置 CELERYD_MAX_TASKS_PER_CHILD = 10 ,表示每个worker执行了多少个任务就死掉

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

# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'website.settings')

app = Celery('website')

# Using a string here means the worker don'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()

# 允许root 用户运行celery
platforms.C_FORCE_ROOT = True

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

4.创建 website/celery.py 主文件

@shared_task
def report():
  return 5

异步任务配置

我的异步使用场景为项目上线:前端web上有个上线按钮,点击按钮后发请求给后端,后端执行上线过程要5分钟,后端在接收到请求后把任务放入队列异步执行,同时马上返回给前端一个任务执行中的结果。若果没有异步执行会怎么样呢?同步的情况就是执行过程中前端一直在等后端返回结果,页面转呀转的就转超时了。

3.celery用在django项目中,django项目目录结构(简化)如下

定时任务配置

from celery.schedules import crontab

app.conf.update(
  CELERYBEAT_SCHEDULE = {
    'sum-task': {
      'task': 'deploy.tasks.add',
      'schedule': timedelta(seconds=20),
      'args': (5, 6)
    }
    'send-report': {
      'task': 'deploy.tasks.report',
      'schedule': crontab(hour=4, minute=30, day_of_week=1),
    }
  }
)
  1. deploy/tasks.py 文件添加report方法:
# apt-get install rabbitmq-server

celery是一个基于python开发的简单、灵活且可靠的分布式任务队列框架,支持使用任务队列的方式在分布式的机器/进程/线程上执行任务调度。采用典型的生产者-消费者模型,主要由三部分组成:

Tips

1.如果你同时使用了异步任务和计划任务,有一种更简单的启动方式 celery -A website worker -b -l info ,可同时启动worker和beat

本文由分分快三计划发布,转载请注明来源

关键词: 分分快三计划