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

Posted 2020-08-11

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