celery与django的配合使用

一、流程

二、准备工作

1. celery4.4.0文档https://docs.celeryproject.org/en/stable/

2. 安装模块 pip install celery 4.4.0

3. 在根目录下新建celery_tasks/celery.py

4. celery.py 文件

   from __future__ import absolute_import, unicode_literals
   import os
   from celery import Celery
   # 1. 设置环境变量
   os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')
   # 2. 创建celery对象
   app = Celery('proj')
   # 3. 加载配置文件
   app.config_from_object('celery_tasks.config', namespace='CELERY')
   # 4. 加载任务模块
   app.autodiscover_tasks()
   # 5. 任务
   @app.task(bind=True,name='xx')
   def debug_task(self):
       print('Request: {0!r}'.format(self.request))
   

5. 配置文件config.py

   # 存放结果
   result_backend = 'redis://127.0.0.1:6379/15'
   # 存放任务
   broker_url = 'redis://127.0.0.1:6379/14' 
   

三、启动

1. 启动命令

   celery -A celery_tasks.celery worker -l info

   fuqiang:test009 fuqiang$ celery -A celery_tasks.celery worker -l info
    
    -------------- celery@fuqiang.local v4.3.0 (rhubarb)
   ---- **** ----- 
   --- * ***  * -- Darwin-18.7.0-x86_64-i386-64bit 2020-02-17 13:19:28
   -- * - **** --- 
   - ** ---------- [config]
   - ** ---------- .> app:         proj:0x104c09940
   - ** ---------- .> transport:   redis://127.0.0.1:6379/14
   - ** ---------- .> results:     redis://127.0.0.1:6379/15
   - *** --- * --- .> concurrency: 8 (prefork)
   -- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
   --- ***** ----- 
    -------------- [queues]
                   .> celery           exchange=celery(direct) key=celery
                   
   [tasks]
     . xx
   
   [2020-02-17 13:19:28,432: INFO/MainProcess] Connected to redis://127.0.0.1:6379/14
   [2020-02-17 13:19:28,442: INFO/MainProcess] mingle: searching for neighbors
   [2020-02-17 13:19:29,463: INFO/MainProcess] mingle: all alone
   [2020-02-17 13:19:29,481: INFO/MainProcess] celery@fuqiang.local ready.
   [2020-02-17 13:20:03,732: INFO/MainProcess] Received task: xx[e53b1786-a8c3-4eed-83e5-95bf6693f88e]  
   [2020-02-17 13:20:03,734: WARNING/ForkPoolWorker-8] ok
   
   
   

2. 测试 在开一个终端

   进入django的交互模式 python manager.py shell

   fuqiang:test009 fuqiang$ python manage.py shell
   Python 3.6.8 (v3.6.8:3c6b436a57, Dec 24 2018, 02:04:31) 
   Type 'copyright', 'credits' or 'license' for more information
   IPython 7.6.1 -- An enhanced Interactive Python. Type '?' for help.
   
   In [1]: from celery_tasks.celery import debug_task                                                                           
   
   In [2]: debug_task.delay()                                                                                                   
   Out[2]: <AsyncResult: e53b1786-a8c3-4eed-83e5-95bf6693f88e>
   

3. 查看任务

   127.0.0.1:6379[14]> keys *
   1) "unacked_mutex"
   2) "_kombu.binding.celeryev"
   3) "_kombu.binding.celery.pidbox"
   4) "_kombu.binding.celery"
   
   127.0.0.1:6379[14]> get unacked_mutex
   "09c47a60-201c-40e3-9b66-eb458e7d1d7f"
   

4. 查看结果

   127.0.0.1:6379[15]> keys *
   1) "celery-task-meta-e53b1786-a8c3-4eed-83e5-95bf6693f88e"
   2) "celery-task-meta-337bd385-29d5-4b99-9a62-ee445707dcc8"
   3) "celery-task-meta-e796c575-b8c1-4f62-b83c-bfa2b13d51ad"
   4) "celery-task-meta-52f9f207-a3e0-4338-8da6-7b86add24cf0"
   
   127.0.0.1:6379[15]> get celery-task-meta-e53b1786-a8c3-4eed-83e5-95bf6693f88e
   "{"status": "SUCCESS", "result": null, "traceback": null, "children": [], "task_id": "e53b1786-a8c3-4eed-83e5-95bf6693f88e", "date_done": "2020-02-17T13:20:03.735324"}"
   127.0.0.1:6379[15]>