【Django】 Celery 进程执行异步任务（生产者与消费者）

ゞ浴缸里的玫瑰 2022-12-03 11:25 326阅读 0赞

我们在项目中可能会遇到需要多任务执行的情况，例如给用户发送短信验证码、邮件等，但是可能会因为网络问题导致进程阻塞，这时候我们可以开启其他进程。有了 Celery，我们在使用生产者消费者模式时，只需要关注任务本身，极大的简化了程序员的开发流程

celery简单可靠，还可以在多台服务器上运行，单个celery进程每分钟可以处理数以百万计的任务

#### 1.安装celery ####

我们进入到虚拟环境中安装celery

pip install Celery -i https://pypi.tuna.tsinghua.edu.cn/simple

#### 2.创建实例和配置 ####

##### 2.1 新增一个任务包 #####

在Django项目外层目录中新建一个celery\_task包，里面分别新建main.py主文件、config.py配置文件、任务包（这里的包名叫sms），结构如下  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5MTQ3Mjk5_size_16_color_FFFFFF_t_70_pic_center]

##### 2.2 配置数据库 #####

这里以Redis数据库为例

config.py

# 设置中间人，这里是redis三号库
    broker_url = "redis://127.0.0.1:6379/3"
    
    
    # 若是使用rabbitmq
    # broker_url = "amqp://用户名:密码@ip地址:5672"

##### 2.3 设置任务 #####

我们需要在任务包（这里是sms）里 **新建一个tasks.py文件，这个文件名是固定的**，不要用其他名字

然后在tasks.py文件里设置任务  
tasks.py

from celery_tasks.main import celery_app  # 这是在main.py里定义的对象名
    
    # 使用装饰器装饰要执行的函数名，之后可以使用delay()方法
    @celery_app.task(name='ccp_send_sms_code')
    def ccp_send_sms_code(mobile, sms_code):
        '''该函数就是一个任务, 用于发送短信'''
        result = CCP().send_template_sms(mobile,[sms_code, 5],1)
        return result

##### 2.4 main文件 #####

我们要在主文件里实例化Celery对象，并且使用上面的配置和指定任务文件

from celery import Celery
    
    celery_app = Celery('main')
    celery_app.config_from_object('celery_tasks.config')
    
    # 指定任务所在包的位置，它会自动捕捉执行
    celery_app.autodiscover_tasks(['celery_tasks.sms'])

##### 2.5 调用任务 #####

因为我们给函数加上了装饰器，所以我们只要使用原来的函数名加delay()运行，它就会自动开启异步执行了（需要开启celery服务）

from celery_tasks.sms.tasks import ccp_send_sms_code
    
    # 原来的写法:
    # CCP().send_template_sms(mobile, [sms_code, 5], 1)
    
    # 改为现在的写法,注意使用delay()方法
    ccp_send_sms_code.delay(mobile, sms_code)

#### 3.开启服务 ####

我们要开启celery服务才能使用异步  
我们进入项目所在文件夹（这里是处于虚拟环境），然后使用celery命令开启服务

celery -A celery_tasks.main worker -l info

**注意**：celery4之后好像是不支持win系统的，需要使用下面的eventlet运行才可以，不然执行任务时会报错

参数说明

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>-A</td> 
   <td>启动文件，即上面创建的main.py</td> 
  </tr> 
  <tr> 
   <td>worker</td> 
   <td>启动的对象</td> 
  </tr> 
  <tr> 
   <td>-l</td> 
   <td>日志级别，如info</td> 
  </tr> 
  <tr> 
   <td>–concurrency</td> 
   <td>每个CPU开启的进程数</td> 
  </tr> 
  <tr> 
   <td>-P</td> 
   <td>运行的程序，可以是threads、gevent、eventlet等</td> 
  </tr> 
  <tr> 
   <td>-c</td> 
   <td>任务并行度</td> 
  </tr> 
 </tbody> 
</table>

如果想以协程的方式运行，可以下载eventlet

# 安装 eventlet 模块
    $ pip install eventlet -i https://pypi.tuna.tsinghua.edu.cn/simple
    
    # 启用 Eventlet 池
    $ celery -A celery_tasks.main worker -l info -P eventlet -c 1000

#### 4.报错 ####

1.`from kombu.async.timer import Entry, Timer as Schedule, to_timestamp, logger SyntaxError: invalid syntax`  
celery3.x版本最高支持python3.6

2.`not enough values to unpack (expected 3, got 0)`  
celery4以及之后的版本不支持win系统，所以我们需要使用eventlet运行

celery -A XXX.celery_tasks.main worker -l info -P eventlet

3.`django.db.utils.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 XXXXX and this is thread id XXXXX.`

启动命令改换成单任务执行，这也会失去了并发的优势

celery -A XXX.celery_tasks.main worker -l info --pool=solo
    # 发送的任务不会被并发执行，在worker执行任务过程中，再次发送给worker的任务会排队，执行完一个再执行另一个

–pool=solo是单任务的，我们可以使用多线程的方式

celery -A XXX.celery_tasks.main worker -l info -P threads -c 10
    # -c 是任务的并行度

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5MTQ3Mjk5_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/6ad7b6a189a34ff2ac1da710fa324234.png