基于Flask的高并发部署方案
@[toc]
在AI部署方案中,Flask是最常用的方案!本文列举几种最常用基于Flask的部署方案。
Flask方案
简介
Flask 是一个轻量级的 Python Web 框架,它非常适合构建小型到中型的应用程序。下面是对 Flask 的详细解释和简单示例:
Flask 的特点:
- 轻量级:相比于 Django,Flask 更轻量级,更适合小型项目或微服务。
- 简单:Flask 的 API 非常简洁,易于学习和使用。
- 灵活:Flask 提供了基础功能,但并不限制开发者如何实现这些功能。
- 扩展性强:有许多针对 Flask 的扩展,可以轻松地添加新功能。
基本组件:
- 路由:定义了 URL 和处理它们的函数之间的映射关系。
- 模板引擎:用于渲染 HTML 页面。
- URL 构建:用于构建 URL。
- 请求和响应对象:处理 HTTP 请求和响应。
- 会话和 Cookie:用于跟踪用户会话。
- 错误处理:捕获和处理异常。
- 上下文:管理执行环境。
简单示例:
- 安装 Flask:首先需要安装 Flask。可以使用 pip 进行安装。
pip install flask
服务端代码
from flask import request, Flask
import base64
import cv2
import numpy as np
app = Flask(__name__)
@app.route("/get_frame", methods=['POST','GET'])
def get_frame():
#解析图片数据
img_card = base64.b64decode(request.form['image'])
image_data = np.frombuffer(img_card, np.uint8)
image_data = cv2.imdecode(image_data, cv2.IMREAD_COLOR)
cv2.imwrite('01.png', image_data)
print(image_data)
return 'koukou'
if __name__ == "__main__":
app.run("0.0.0.0", port=5005)
然后运行
客户端代码
新建test.py,代码如下:
import json
import requests
import base64
from PIL import Image
import numpy as np
from io import BytesIO
#将图片数据转成base64格式
img_path='./test/䗉螺42.jpg'
imag=Image.open(img_path)
img_buffer = BytesIO()
imag.save(img_buffer, format='JPEG')
byte_data = img_buffer.getvalue()
base64_str = base64.b64encode(byte_data).decode()
res = {"image":base64_str}
#访问服务
info = requests.post("http://127.0.0.1:5005/get_frame",data=res)
print(info.text)
运行客户端,向服务端发请求,结果如下:
Gevent +Flask方案
简介
Gevent 是一个 Python 的并发库,它使用 greenlet(轻量级线程)实现协程,允许开发者以同步的方式编写异步代码,从而提高并发性能。
以下是 Gevent 的一些主要特点:
- 基于 greenlet 的协程:Gevent 利用 greenlet 实现协程,使得代码在等待 I/O 操作时能够被调度到其他 greenlet 上执行。
- 非阻塞 I/O:Gevent 使用 libevent 库实现非阻塞 I/O,可以处理大量并发连接。
- 透明地支持同步和异步代码:使用 Gevent,开发者可以以同步的方式编写异步代码,无需修改现有代码或使用回调函数。
- 事件循环:Gevent 使用了类似 Node.js 的事件循环模型,可以轻松地处理高并发场景。
- 集成其他库:Gevent 可以与许多其他 Python 库集成,如 SQLAlchemy、requests 等,使得这些库也支持异步操作。
使用 Gevent,你可以轻松地编写并发代码,提高应用程序的性能和响应能力。它适用于各种场景,如 Web 开发、网络爬虫、实时通信等。
安装
pip install gevent
示例
下面就是一个简单的gevent 示例:
from gevent import monkey
from gevent.pywsgi import WSGIServer
monkey.patch_all()
from multiprocessing import Process
import flask
from flask import request
import redis
import uuid
import time
# initialize our Flask application and Redis server
app = flask.Flask(__name__)
@app.route("/")
def homepage():
return "Hello World!"
@app.route("/predict", methods=["POST"])
def predict():
if flask.request.method == "POST":
json_data = request.get_data()
params = json.loads(json_data)
img_card = params['image']
return 'koukou'
def run(MULTI_PROCESS):
if MULTI_PROCESS == False:
WSGIServer(('0.0.0.0', 8080), app).serve_forever()
else:
mulserver = WSGIServer(('0.0.0.0', 8080), app)
mulserver.start()
def server_forever():
mulserver.start_accepting()
mulserver._stop_event.wait()
for i in range(15):
p = Process(target=server_forever)
p.start()
if __name__ == "__main__":
# 单进程 + 协程
run(False)
# 多进程 + 协程
#run(True)
注意:多进程+协程只能在Linux上使用,在Win上会报错了!
客户端代码:
import json
import requests
import base64
from PIL import Image
import numpy as np
from io import BytesIO
#将图片数据转成base64格式
img_path='./test/䗉螺42.jpg'
imag=Image.open(img_path)
img_buffer = BytesIO()
imag.save(img_buffer, format='JPEG')
byte_data = img_buffer.getvalue()
base64_str = base64.b64encode(byte_data).decode()
res = {"image":base64_str}
#访问服务
info = requests.post("http://127.0.0.1:8080/predict",data=res)
print(info.text)
gunicorn+Flask 部署服务
简介
Gunicorn 是一个高性能的 Python WSGI HTTP 服务器,主要用于部署 Python Web 应用。以下是 Gunicorn 的主要特点和详解:
- WSGI HTTP 服务器:Gunicorn 是一个 WSGI HTTP 服务器,可以与各种 Web 框架(如 Flask、Django 等)无缝集成。它使用预fork模式,能够在启动时预先fork出指定数量的 worker 进程来处理请求。
- 高性能:由于采用了预fork模式,Gunicorn 在处理请求时具有高性能。它还支持多线程或多进程,可以根据实际需求进行配置。
- 简单易用:安装和使用 Gunicorn 都非常简单。可以通过 pip 安装,然后通过命令行运行。同时,Gunicorn 还支持配置文件,可以方便地配置各种参数。
- 兼容性强:Gunicorn 与大多数 Web 框架兼容,如 Flask、Django、Pyramid 等。它还支持各种WSGI应用,可以与各种 Python 库和工具集成。
- 扩展性高:如果需要更多的功能,可以通过扩展 Gunicorn 或使用其他第三方工具与 Gunicorn 集成。
安装
pip install gunicorn
注意:gunicorn是Linux特有的库,建议使用Linux。
示例
创建gunicorn配置文件,在项目跟目录创建一个gunicorn.py文件,代码如下:
import gevent.monkey
gevent.monkey.patch_all()
import multiprocessing
import os
if not os.path.exists('log'):
os.mkdir('log')
debug = True
loglevel = 'debug'
bind = '127.0.0.1:5005'
pidfile = 'log/gunicorn.pid'
logfile = 'log/debug.log'
errorlog = 'log/error.log'
accesslog = 'log/access.log'
# 启动的进程数
workers = 8
worker_class = 'gunicorn.workers.ggevent.GeventWorker'
x_forwarded_for_header = 'X-FORWARDED-FOR'
使用gevent模式来支持并发
创建程序入口,在使用Pycharm创建Flask项目的时候,会生成一个app.py的入口文件,里面是创建启动App实例,在这里我们创建一个新的程序入口,用来使用Gunicorn服务,在生产环境中使用。
在项目根目录创建一个app.py的文件,内容如下:
from flask import request, Flask
import base64
import cv2
import numpy as np
app = Flask(__name__)
@app.route("/get_frame", methods=['POST','GET'])
def get_frame():
#解析图片数据
img_card = base64.b64decode(request.form['image'])
image_data = np.frombuffer(img_card, np.uint8)
image_data = cv2.imdecode(image_data, cv2.IMREAD_COLOR)
cv2.imwrite('01.png', image_data)
print(image_data)
return 'koukou'
if __name__ == "__main__":
app.run("0.0.0.0", port=5005)
接下来,在项目根目录创建一个start.py的文件,内容如下:
from app import app
import logging
gunicorn_logger = logging.getLogger('gunicorn.error')
app.logger.handlers = gunicorn_logger.handlers
app.logger.setLevel(gunicorn_logger.level)
if __name__ == '__main__':
app.run()
运行程序,执行命令:
gunicorn -c gunicorn.py start:app
参考文章:
- 点赞
- 收藏
- 关注作者
评论(0)