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.conf1.设置为后台启动 daemonize yes2.redis端口设置 port 6379 # default prot3.日志文件 logfile /home/liuyadong/work/log/redis.log4.数据保存文件 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 6379tcp        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 celery3.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的模拟购置,本次实例中为:tasks4.Running the celery worker server  $ celery -A tasks worker --loglevel=info5.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的traceback7.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

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