【理论】python使用celery异步处理请求

时间：2021-03-03 09:06:37

Flask中使用celery队列处理执行时间较长的请求。

一. 安装celery

pip install celery flask  redis

二. celery简介

Celery是个异步分布式任务队列

通过Celery在后台跑任务并不像线程那么简单，但是用Celery的话，能够是应用有较好的扩展性，因为Celery是个分布式架构，下面介绍Celery的三个核心组件：

1. 生产者(Celery client): 生产者发送消息，在Flask上工作时，生产者在Flask应用内运行

2. 消费者(Celert worker): 消费者用于处理后台任务。消费者可以是本地的也可以是远程的。我们可以在运行Flask的server上运行一个单一的消费者，当业务量上涨之后再去添加更多的消费者

3. 消息传递着(Celery broker): 生产者和消费者的信息交互使用的是消息队列，Celery支持若干方式的消息队列，其中最长用的是RabbitMQ和Redis, 我们在使用过程中使用的Redis

三. redis配置与使用

redis配置文件/etc/redis.conf

1.设置为后台启动

 daemonize yes

2.redis端口设置

 port 6379 # default prot

3.日志文件

 logfile /home/liuyadong/work/log/redis.log

4.数据保存文件

 dir /home/liuyadong/data/redisData

通过下面命令指定配置文件启动redis:

redis-server /etc/redis.conf

通过下面命令测试是否启动成功:

redis-cli -p 6379

下面这样表示成功(进入了命令行模式)：

redis 127.0.0.1:6379> 

查看启动端口:

sudo netstat -ntlp | grep 6379

tcp        0      0 127.0.0.1:6379              0.0.0.0:*                   LISTEN      49380/redis-server

四. celery使用简介

1.Choosing a broker

 最常用的broker包括: RabbitMQ 和 Redis, 我们使用Redis, Redis的安装及启动等查看第二部分

2.intall celery

 pip install celery

3.Application

 使用celery的第一步是创建一个application, 通常叫做'app'。具体的创建一个app的代码如下:

 $ cat tasks.py

 #!/usr/bin/env python

 from celery import Celery

 app = Celery('tasks', broker='redis://localhost')

 @app.tasks

 def add(x, y):

     return x + y

 Note: Celery第一个参数必须是当前module的模拟购置,本次实例中为:tasks

4.Running the celery worker server

  $ celery -A tasks worker --loglevel=info

5.Calling the tasks

 可以通过delay()或者apply_sync()方法来调用一个task

 >>> from tasks import add

 >>> add.delay(4, 4)

6. Keeping Results

 我们可以将task的执行状态保存起来,可以保存到broker中, 可以通过CELERY_RESULT_BACKEND字段来设置保存结果。

 也可以通过Celery的backend参数来设置

 app.Celery('tasks', broker='redis://localhost', backend='redis://localhost')

 >>> result = add.delay(4, 4)

 可以通过ready()方法来判断程序执行是否完成,执行完成返回True.

 >>> result.ready()

 False

 下面是AsyncResult对象的其他调用方法介绍:

 1) AsyncResult.get(timeout=None, propagate=True, interval=0.5, no_ack=True, follow_parents=True)

 timeout    : 设置一个等待的预操作时间,单位是s, 方法返回执行结果

 propagate  : 如果task执行失败,则Re-taise Exception

 interval   : 等待一定时间重新执行操作,如果使用amqp来存储backend则此参数无效

 no_ack     : Enable amqp no ack (automatically acknowledge message)

              If this is False then the message will not be acked

 follow_parents :  Reraise any exception raised by parent task

 2) AsyncResult.state 或 status属性

 方法返回当前task的执行状态,返回值包括下面多种情况:

 PENDING: task正在等待执行

 STARTED: task已经开始执行了

 RETRY  : task重新执行了,这可能是由于第一次执行失败引起的

 FAILURE: task执行引发了异常,并且结果的属性当中包括了异常是由哪个task引起的

 SUCCESS: task执行成功,结果的属性当中包括执行结果

 3) AsyncResult.success()

 如果返回True,则表示task执行成功

 4) AsyncResult.traceback()

 得到一个执行失败的task的traceback

7.Configuration celert

 默认的配置对于大多数用户来说已经足够好了，但是我们仍有许多想让celery按照我们的想法去work,通过configuration实现是一个好的方式。 configutation可以通过app设置,也可以通过一个单独的模块进行设置。

 比如,通过app设置CELERY_TASK_SERIALIZER属性:app.conf.CELERY_TASK_SERIALIZER = 'json'

 如果你一次性有许多需要配置,则可以通过update()方法实现:

 app.conf.update(

    CELERY_TASK_SERIALIZER='json',

    CELERY_ACCEPT_CONTENT=['json'], # Ignore other content

    CELERY_RESULT_SERIALIZER='json',

    CELERY_TIMEZONE='Europe/Oslo',

    CELERY_ENABLE_UTC=True,

 )

 你也可以通过app.config_from_object() method告诉Celery通过一个模块来生成configuration: app.config_from_object('celeryconfig') 这个模块通常叫做 celeryconfig,但实际上你可以叫任何名字。

  $ cat celeryconfig.py

  CELERY_ROUTES = {'tasks.add': 'low-priority', 'tasks.add': {'rate_limit': '10/m'}

8.Where to go from here

 如果你想了解更多请阅读: http://docs.celeryproject.org/en/latest/getting-started/next-steps.html#next-steps

 之后阅读: http://docs.celeryproject.org/en/latest/userguide/index.html#guide

相关文章