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成功消费了每个定时任务：