作為著名Python web框架之一的Flask，具有簡單輕量、靈活、擴展豐富且上手難度低的特點，因此成為了機器學習和深度學習模型上線跑定時任務，提供API的首選框架。
眾所周知，Flask默認不支持非阻塞IO的，當請求A還未完成時候，請求B需要等待請求A完成后才能被處理，所以效率非常低。但是線上任務通常需要異步、高並發等需求，本文總結一些在日常使用過程中所常用的技巧。
一、前沿
異步和多線程有什么區別？其實，異步是目的，而多線程是實現這個目的的方法。異步是說，A發起一個操作后（一般都是比較耗時的操作，如果不耗時的操作就沒有必要異步了），可以繼續自顧自的處理它自己的事兒，不用干等着這個耗時操作返回。
實現異步可以采用多線程技術或則交給另外的進程來處理,詳解常見這里。
二、實現方法
- Flask啟動自帶方法
采用gunicorn部署
1、Flask中自帶方法實現
run.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2018-12-01 16:37
# @Author : mokundong
from flask import Flask
import socket
from time import sleep
myhost = socket.gethostbyname(socket.gethostname())
app = Flask(__name__)
@app.route('/job1')
def some_long_task1():
print("Task #1 started!")
sleep(10)
print("Task #1 is done!")
@app.route('/job2')
def some_long_task2(arg1, arg2):
print("Task #2 started with args: %s %s!" % (arg1, arg2))
sleep(5)
print("Task #2 is done!")
if __name__ == '__main__':
app.run(host=myhost,port=5000,threaded=True)
app.run(host=xxx,port=xx,threaded=True)
中threaded開啟后則不需要等隊列。
2、gunicorn部署
Gunicorn 是一個高效的Python WSGI Server,通常用它來運行 wsgi application 或者 wsgi framework(如Django,Paster,Flask),地位相當於Java中的Tomcat。gunicorn 會啟動一組 worker進程，所有worker進程公用一組listener，在每個worker中為每個listener建立一個wsgi server。每當有HTTP鏈接到來時，wsgi server創建一個協程來處理該鏈接，協程處理該鏈接的時候，先初始化WSGI環境，然后調用用戶提供的app對象去處理HTTP請求。
關於gunicorn的詳細說明，可以參考這里。使用命令行啟動gunicorn有兩種方式獲取配置項，一種是在命令行配置，一種是在配置文件中獲取。
run.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2018-12-01 17:00
# @Author : mokundong
from flask import Flask
from time import sleep
app = Flask(__name__)
@app.route('/job1')
def some_long_task1():
print("Task #1 started!")
sleep(10)
print("Task #1 is done!")
@app.route('/job2')
def some_long_task2(arg1, arg2):
print("Task #2 started with args: %s %s!" % (arg1, arg2))
sleep(5)
print("Task #2 is done!")
if __name__ == '__main__':
app.run()
命令行配置
gunicorn --workers=4 --bind=127.0.0.1:8000 run:app
更多配置見官網
配置文件獲取配置
gunicorn_config.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2018-12-01 17:10
# @Author : mokundong
import os
import socket
import multiprocessing
import gevent.monkey
gevent.monkey.patch_all()
myhost = socket.gethostbyname(socket.gethostname())
debug = False
loglevel = 'info'
hosts = get_host_ip()
bind = hosts+":5000"
timeout = 30 #超時
pidfile = "log/gunicorn.pid"
accesslog = "log/access.log"
errorlog = "log/debug.log"
daemon = True #意味着開啟后台運行，默認為False
workers = 4 # 啟動的進程數
threads = 2 #指定每個進程開啟的線程數
worker_class = 'gevent' #默認為sync模式，也可使用gevent模式。
x_forwarded_for_header = 'X-FORWARDED-FOR'
啟動命令如下
gunicorn -c gunicorn_config.py run:app
三、補充
1、關於線程的補充
在工作中我還遇到一種情況，當一個請求過來后，我需要兩種回應，一個是及時返回app運行結果，第二個響應是保存數據到日志或者數據庫。往往我們在寫數據的過程中會花銷一定的時間，導致結果返回會有所延遲，因此我們需要用兩個線程處理這兩個任務，那么我們如下處理。
run.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2018-12-01 17:20
# @Author : mokundong
from flask import Flask,request
from time import sleep
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(2)
app = Flask(__name__)
@app.route('/job')
def run_jobs():
executor.submit(some_long_task1)
executor.submit(some_long_task2, 'hello', 123)
return 'Two jobs was launched in background!'
def some_long_task1():
print("Task #1 started!")
sleep(10)
print("Task #1 is done!")
def some_long_task2(arg1, arg2):
print("Task #2 started with args: %s %s!" % (arg1, arg2))
sleep(5)
print("Task #2 is done!")
if __name__ == '__main__':
app.run()
2、關於獲取IP的補充
上述代碼中通過獲取
hostname，然后再通過hostname反查處機器的IP。這個方法是不推薦的。因為很多的機器沒有規范這個hostname的設置。
另外就是有些服務器會在/etc/hosts中添加本機的hostname的地址，這個做法也不是不可以，但是如果設置成了127.0.0.1，那么獲取出來的IP就都是這個地址了。
這里給出一種優雅的方式獲取IP，利用 UDP 協議來實現的，生成一個UDP包，把自己的 IP 放如到 UDP 協議頭中，然后從UDP包中獲取本機的IP。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2018-12-01 17:30
# @Author : mokundong
# 可以封裝成函數，方便 Python 的程序調用
import socket
def get_host_ip():
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(('8.8.8.8', 80))
ip = s.getsockname()[0]
finally:
s.close()
return ip
總結
當然推薦使用gunicorn部署多線程，Flask自帶的，emmmm，測試玩兒玩兒吧。
在寫作過程中才發現自己知識漏洞不是一般多，共勉！