> For the complete documentation index, see [llms.txt](https://close.gitbook.io/yun-wei-bi-ji/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://close.gitbook.io/yun-wei-bi-ji/python/django/15celery-yi-bu-ren-wu-ji-cheng.md). # 15、Celery异步任务集成 ### 环境： ```python django-redis = "*" django-celery-results = "==1.1.2" django-celery-beat = "==1.5.0" celery = "==4.3.0" ``` ### 配置 setting\_base.py > 我这里开发环境和生成环境区分， 配置文件修改为： setting\_base.py， 你的环境如果是默认环境，就修改 settings.py ```python # 添加 APP INSTALLED_APPS = [ # ... 'app', # 测试的 app 名称 'django_celery_results', # 查看 celery 执行结果 'django-celery-beat', # 后台管理任务 ] # django 缓存 CACHES = { "default": { "BACKEND": "django_redis.cache.RedisCache", "LOCATION": "redis://:123456@10.10.181.18:6379/1", "OPTIONS": { "CLIENT_CLASS": "django_redis.client.DefaultClient", } } } ############################# # celery 配置信息 start ############################# # celery 定时任务 # 注意，celery4 版本后，CELERY_BROKER_URL 改为 BROKER_URL CELERY_BROKER_URL = 'redis://:123456@10.10.181.18:6379/0' # CELERY_RESULT_BACKEND = 'redis://:123456@10.10.181.18:6379/1' # celery 结果返回，可用于跟踪结果 CELERY_RESULT_BACKEND = 'django-db' # 使用 database 作为结果存储 CELERY_CACHE_BACKEND = 'django-cache' # celery 后端缓存 CELERYD_MAX_TASKS_PER_CHILD = 3 # 每个 worker 最多执行3个任务就会被销毁，可防止内存泄露 # CELERY_STORE_ERRORS_EVEN_IF_IGNORED = True # CELERY_TASK_ALWAYS_EAGER = True # celery 内容等消息的格式设置 if os.name != "nt": # Mac and Centos # worker 启动命令：celery -A joyoo worker -l info CELERY_ACCEPT_CONTENT = ['application/json', ] CELERY_TASK_SERIALIZER = 'json' CELERY_RESULT_SERIALIZER = 'json' else: # windows # pip install eventlet # worker 启动命令：celery -A joyoo worker -l info -P eventlet --pool=solo CELERY_ACCEPT_CONTENT = ['pickle', ] CELERY_TASK_SERIALIZER = 'pickle' CELERY_RESULT_SERIALIZER = 'pickle' ############################# # celery 配置信息 end ############################# ``` ### 项目根目录添加 celery.py ```python from __future__ import absolute_import, unicode_literals import os from celery import Celery from django.conf import settings # 获取当前文件夹名，即为该 Django 的项目名 project_name = os.path.split(os.path.abspath('.'))[-1] # project_settings = '%s.settings' % project_name # 设置环境变量 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.setting_base') # 实例化 app = Celery() # namespace='CELERY'作用是允许你在Django配置文件中对Celery进行配置 # 但所有Celery配置项必须以CELERY开头，防止冲突 app.config_from_object('django.conf:settings', namespace='CELERY') # Celery 加载所有注册的应用 # app.autodiscover_tasks() app.autodiscover_tasks(lambda: settings.INSTALLED_APPS) ``` ### 项目根目录 \_\_init\_\_.py ```python from __future__ import absolute_import, unicode_literals import os from celery import Celery from django.conf import settings # 获取当前文件夹名，即为该 Django 的项目名 project_name = os.path.split(os.path.abspath('.'))[-1] # project_settings = '%s.settings' % project_name # 设置环境变量 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.setting_base') # 实例化 app = Celery() # namespace='CELERY'作用是允许你在Django配置文件中对Celery进行配置 # 但所有Celery配置项必须以CELERY开头，防止冲突 app.config_from_object('django.conf:settings', namespace='CELERY') # Celery 加载所有注册的应用 # app.autodiscover_tasks() app.autodiscover_tasks(lambda: settings.INSTALLED_APPS) ``` ### 在提前创建好的 App 下测试 `app/task.py` ```python from __future__ import absolute_import, unicode_literals from celery import task @task def demo_task(): print('test') ``` `app/views.py` ```python from django.http import JsonResponse from . import tasks def demo_task(request): res = tasks.demo_task.delay() return JsonResponse({'status': 'successful', 'task_id': res.task_id}) ``` `app/urls.py` ```python from django.urls import path from . import views urlpatterns = [ path('demo_task/', views.demo_task), ] ``` `project/urls.py`，根项目的urls.py ```python urlpatterns = [ '''''' # 测试 celery path('domain/', include('app.urls')), ] ``` ### 生成数据库表 ```python python manage.py makemigrations python manage.py migration ``` ### 启动 celery worker ```python # 项目根目录终端执行（project 项目名称） # celery -A project worker -l info celery -A project worker -l info --pool=solo # 守护进程 /root/.virtualenvs/blog/bin/celery multi start w1 -A project -l info --logfile=./celerylog.log ``` ### 启动 celery beat ```python # 项目根目录终端执行（project 项目名称） celery -A project beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler ``` ### 启动项目 ```python python manage.py runserver 0.0.0.0:8000 ``` ![](/files/jrPif66bkslhMUWJH5hr) ![](/files/eZGCtEOblOniPktNov3O) ![](/files/5JRYM0NQ0uYM9qXUAFgd) 参考： 参考： #### 二次开发 　　django-celery-beat插件本质上是对数据库表变化检查，一旦有数据库表改变，调度器重新读取任务进行调度，所以如果想自己定制的任务页面，只需要操作beat插件的四张表就可以了。当然你还可以自己定义调度器，django-celery-beat插件已经内置了model，只需要进行导入便可进行orm操作，以下我用django reset api进行示例： `settings.py` ```python 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` ```python urlpatterns = [ re_path(r'^admin/', admin.site.urls), re_path(r'^index$', views.index), re_path(r'^res$', views.get_res), re_path(r'^tasks$', views.TaskView.as_view({'get':'list'})), ] ``` `views.py` ```python 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 ``` `访问http://127.0.0.1:8000/tasks如下：` ![](/files/uxoJtAg3zx7HjWGMMUDo)