django中配置使用celery

环境版本：

windows7 x64

django 1.11.6

django-celery 3.2.2

工程结构说明：源码下载请访问https://i.cnblogs.com/Files.aspx

1、新建django项目DjangoCelery（具体请参考https://www.cnblogs.com/apple2016/p/11434107.html），并启动，启动方式包括2种：

　　1）通过eclipse，右击项目名称，Run As----Pydev:Django

　　2）在manage.py文件所在路径执行命令：python manage.py runserver 0.0.0.0:8000

本次启动采用方式1，启动成功后，控制台日志打印信息如下：

2、在...\testdj\testdj\testdj下新建celery配置文件celeryconfig.py：

# -*- coding: utf-8 -*-

'''

Created on 2019年8月28日

@author: lenovo

'''

import celery

import djcelery

#当djcelery.setup_loader()运行时，Celery便会去查看INSTALLD_APPS下包含的所有app目录中的tasks.py文件，找到标记为task的方法，将它们注册为celery task。

djcelery.setup_loader()

#设置不同的队列，不要只使用默认的队列，这样当任务比较多的时候任务之间会相互影响（例如将普通任务和定时任务混在一起），

CELERY_QUEUES={

    #定时任务队列

    'beat_tasks':{

        'exchange':'beat_tasks',

        'exchange_type':'direct',

        'binding_key':'beat_tasks'

        },

    'work_queue':{

        'exchange':'work_queue',

        'exchange_type':'direct',

        'binding_key':'work_queue'

        }

    #普通任务队列

    }

CELERY_DEFAULT_QUEUE='work_queue' #设置默认队列，若不指定队列则使用该队列

CELERY_IMPORTS=(

    'course.tasks',

    )

#参数配置可参考官网：http://docs.celeryproject.org/en/latest/userguide/configuration.html

CELERY_ACKS_LATE=True #允许重试

CELERYD_FORCE_EXECV=True #可以让Celery更加可靠,只有当worker执行完任务后,才会告诉MQ,消息被消费，防治死锁

CELERYD_CONCURRENCY=4 #设置并发的worker数量

CELERYD_MAX_TASKS_PRE_CHILD=100 #每个worker最多执行100个任务被销毁，可以防止内存泄露

CELERYD_TASK_TIME_LIMIT=12*30 #单个任务的最大运行时间为6分钟，超过的话就被杀掉

3、在项目路径...\testdj\testdj下新建app，命名为course

右击项目名称testdj，选择Django---create application(mange.py startapp)，输入app名称完成创建；

4、将新建的course这个app和djcelery模块注册到settings.py中：

INSTALLED_APPS = [

    'django.contrib.admin',

    'django.contrib.auth',

    'django.contrib.contenttypes',

    'django.contrib.sessions',

    'django.contrib.messages',

    'django.contrib.staticfiles',

    'djcelery',

    'course',

]

注册djcelery是为了后续通过manage.py使用其相关命令；

5、建立Celery与django的关系，包括broker和backend设置，在settings.py中完成相关配置：

#Celery

from celeryconfig import *

BROKER_BACKEND='redis'

BROKER_URL='redis://localhost:6379/1'

CELERY_RESULT_BACKEND='redis://localhost:6379/2'

6、在...\testdj\testdj\course下新建tasks.py文件

# -*- coding: utf-8 -*-

'''

Created on 2019年8月28日

@author: lenovo

'''

import  time

from celery.task import Task

class CourseTask(Task):

    name='course-task' #给该任务起一个名字，这样可以用它在其他地方表示这个任务

    def run(self,*args,**kwargs):

        print 'start course task'

        time.sleep(4)

        print 'args={},kwargs={}'.format(args, kwargs)

        print 'end course task'

7、新建views.py

# -*- coding: utf-8 -*-

'''

Created on 2019年8月29日

@author: lenovo

'''

from course.tasks import CourseTask

from django.http import JsonResponse

def do(request):

    #执行异步任务

    print 'start do request'

    CourseTask.delay()

    print 'end do request'

    return JsonResponse({'result':'ok'})

8、修改urls.py文件：

from django.conf.urls import url

from django.contrib import admin

from course import views

urlpatterns = [

    url(r'^admin/', admin.site.urls),

    url(r'^do/$',views.do,name='do'),

]

9、启动worker：在manage.py文件所在路径下执行命令python manage.py celery worker -l INFO

10、打开浏览器访问http://127.0.0.1:8000/do/完成请求发送，查看runserver启动的服务日志显示通过views.py已发送任务成功，查看worker日志显示worker已接收到任务并进行了消费

并且我们发起请求时，

start do request
end do request

是实时打印的（而非等待4S才打印end do request），说明任务未出现阻塞情况，证明任务是异步的。

11、增加定时任务设置：修改celeryconfig.py文件，增加定时任务，并指定定时任务运行在beat_tasks队列中：

# -*- coding: utf-8 -*-

'''

Created on 2019年8月28日

@author: lenovo

'''

import celery

import djcelery

#当djcelery.setup_loader()运行时，Celery便会去查看INSTALLD_APPS下包含的所有app目录中的tasks.py文件，找到标记为task的方法，将它们注册为celery task。

djcelery.setup_loader()

#设置不同的队列，不要只使用默认的队列，这样当任务比较多的时候任务之间会相互影响（例如将普通任务和定时任务混在一起），

CELERY_QUEUES={

    #定时任务队列

    'beat_tasks':{

        'exchange':'beat_tasks',

        'exchange_type':'direct',

        'binding_key':'beat_tasks'

        },

    'work_queue':{

        'exchange':'work_queue',

        'exchange_type':'direct',

        'binding_key':'work_queue'

        }

    #普通任务队列

    }

CELERY_DEFAULT_QUEUE='work_queue' #设置默认队列，若不指定队列则使用该队列

CELERY_IMPORTS=(

    'course.tasks',

    )

#参数配置可参考官网：http://docs.celeryproject.org/en/latest/userguide/configuration.html

CELERY_ACKS_LATE=True #允许重试

CELERYD_FORCE_EXECV=True #可以让Celery更加可靠,只有当worker执行完任务后,才会告诉MQ,消息被消费，防治死锁

CELERYD_CONCURRENCY=4 #设置并发的worker数量

CELERYD_MAX_TASKS_PRE_CHILD=100 #每个worker最多执行100个任务被销毁，可以防止内存泄露

CELERYD_TASK_TIME_LIMIT=12*30 #单个任务的最大运行时间为6分钟，超过的话就被杀掉

#设置定时任务

from datetime import timedelta

CELERYBEAT_SCHEDULE={

    'task1':{

        'task':'course-task',

        'schedule':timedelta(seconds=5),

        'options':{

            'queue':'beat_tasks'

            }

        }

    }

12、启动beat：执行命令python manage.py celery beat -l INFO

13、查看beat日志定时任务发送情况以及worker日志任务消费情况，当前定时任务跑在beat_tasks队列里正常

　　beat日志显示按照定时任务设置要求每隔5s发送一次任务：

worker日志显示worker成功消费了每个定时任务：

巴特西

django中配置使用celery

最新文章

热门文章