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