本文共 4840 字,大约阅读时间需要 16 分钟。
(1)Celery 4.0要求Django1.8或者比1.8 更新的版本,如果Django低于1.8版本,请使用Celery 3.1版本。
(2)俺是个小气的人,免费听哔哩上几年前的Django教程视频。让安装三个包:celery、celery-with-redis、django-celery。依次安装,发现celery-with-redis、django-celery支持的celery版本最高到4.0。也就是最新版本的Celery已经实现了 celery-with-redis、django-celery包的功能,不在需要安装这两个包。
ERROR: django-celery 3.3.1 has requirement celery<4.0,>=3.1.15, but you'll have celery 4.1.1 which is incompatible.ERROR: celery-with-redis 3.0 has requirement celery<4.0,>=3.0, but you'll have celery 4.1.1 which is incompatible.
(1)自建的Django项目目录结构如下。 proj为项目名称。
- proj/ - manage.py - proj/ - __init__.py - settings.py - urls.py - app1 - __init__.py - admin.py - app1.py - models.py - tests.py - urls.py - views.py
(2)在proj/proj/目录下建立celery.py。 proj需要更改成自己项目名字。
from __future__ import absolute_import, unicode_literalsimport osfrom celery import Celery# set the default Django settings module for the 'celery' program.os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')app = Celery('proj')# 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))
上面内容解释:
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings') 为Celery命令设置默认的环境变量。当执行Celery命令时,会自动使用指定的settings作为配置文件,而不必每次执行时都要指定配置文件。
app = Celery('proj') 生成Celery实例,对于Django,只需要一个
app.config_from_object('django.conf:settings', namespace='CELERY') 设置Django项目的settings作为Celery的一部分配置文件。这样你就可以直接将Celery的配置项配置在Django项目的settings文件中,而不必为Celery单独写一个配置文件。但是Celery支持单独定义配置文件。
app.autodiscover_tasks() 最佳项目实践是将不同app的task放在不同的task.py模块中。自动加载经过注册的app根目录下的task。app的根目录如下:。这样就不需要配置CELERY_IMPORTS来指定每个app具体task.py路径。
- app1/ - tasks.py - models.py- app2/ - tasks.py - models.py
(3)在proj/proj/__init__.py文件中导入Celery app
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',)
(4)在proj的settings.py中添加配置项
# root为随意的一个名字CELERY_BROKER_URL = 'redis://root:password@127.0.0.1:6379/0'
(5)编写task代码
# Create your tasks herefrom __future__ import absolute_import, unicode_literalsfrom celery import shared_task# 自己建立的模型类from demoapp.models import Widget@shared_taskdef add(x, y): return x + y@shared_taskdef mul(x, y): return x * y@shared_taskdef xsum(numbers): return sum(numbers)import time@shared_taskdef count_widgets(): time.sleep(5) return Widget.objects.count()@shared_taskdef rename_widget(widget_id, name): w = Widget.objects.get(id=widget_id) w.name = name w.save()
(6)打开两个命令行窗口。一个运行Celery命令启动Worker,另一个窗口运行Django项目。
# 启动celery。windows系统需要 -P参数。 eventlet需要单独pip installcelery -A demo worker -l info -P eventlet# 启动Django项目. 项目根目录下python manage.py runserver
(7)views视图函数中使用启动task任务。 当访问此视图函数时,会立即返回celery.html页面,任务会被提交到celery去异步执行。
from .task import count_widgetsdef celery(req): res = count_widgets.delay() # print(res.get()) return render(req, 'app1/celery.html')
(1)版本问题 只需要安装Celery一个包即可
(2)错误:Celery ValueError: not enough values to unpack (expected 3, got 0)
解决方法:
第一步: pip install eventlet 第二步:运行celery命令时增加P参数。 celery -A proj worker -l info -P eventlet
参数说明:
-A
,
--app 使用到的项目名
-P
, --pool Celery使用的实现池。可以有
prefork (default), eventlet, gevent, threads ,solo. -l
, --loglevel 日志级别,可以有
DEBUG, INFO, WARNING, ERROR, CRITICAL,FATAL.
(2)错误:DatabaseError: DatabaseWrapper objects created in a thread can only be used in that same thread. The object with alias 'default' was created in thread id 140107533682432 and this is thread id 65391024.
解决办法:
在Pycharm全文搜索(连按shift两次快捷键),搜索monkey_patch() 函数,其为eventlet源码函数,对其进行修改
def monkey_patch(on={"thread":False}):
问题原因:
经过分析django的db模块的代码,发现进行数据库操作关闭时,会对创建这个连接进行验证是否是同 一个thread进行操作,如果不是一个操作,就会报错。
经过排查eventlet的monkey_patch(),发现eventlet在提供绿色协程时,为了性能提升,竟然强悍的部分原生模块(os、socket、thread)进行了修改,并且通过monkey_patch的方式进行替换了。这里受影响的就是eventlet 对原生的thread 进行了补丁操作,对thread的获取线程id的方法get_ident()进行了重写。
def get_ident(gr=None): if gr is None: return id(greenlet.getcurrent()) else: return id(gr)
这样,就会引起,之前thread.get_ident() 和monkey_patch 后的thread.get_ident() 获取的值不同,monkey_patch() 后获取的是线程中绿色协程的id。
想要解决这个问题,可以在monkey_patch() 时,不对thread进行打补丁,monkey_patch((on={"thread":False}),就可以解决。不过,这样不知道会不会影响到eventlet提供的写成并发能力。如果有大牛,可以做进一步补充。
(4)参考资料
官网资料
别人资料
转载地址:http://xnglz.baihongyu.com/