基于Flask的高并发部署方案

文章目录

在AI部署方案中，Flask是最常用的方案！本文列举几种最常用基于Flask的部署方案。

Flask方案

简介

Flask 是一个轻量级的 Python Web 框架，它非常适合构建小型到中型的应用程序。下面是对 Flask 的详细解释和简单示例：

Flask 的特点：

1. 轻量级：相比于 Django，Flask 更轻量级，更适合小型项目或微服务。
2. 简单：Flask 的 API 非常简洁，易于学习和使用。
3. 灵活：Flask 提供了基础功能，但并不限制开发者如何实现这些功能。
4. 扩展性强：有许多针对 Flask 的扩展，可以轻松地添加新功能。

基本组件：

1. 路由：定义了 URL 和处理它们的函数之间的映射关系。
2. 模板引擎：用于渲染 HTML 页面。
3. URL 构建：用于构建 URL。
4. 请求和响应对象：处理 HTTP 请求和响应。
5. 会话和 Cookie：用于跟踪用户会话。
6. 错误处理：捕获和处理异常。
7. 上下文：管理执行环境。

简单示例：

1. 安装 Flask：首先需要安装 Flask。可以使用 pip 进行安装。

pip install flask

服务端代码

新建server.py

from flask import request, Flask
import base64
import cv2
import numpy as np
app = Flask(__name__)@app.route("/get_frame", methods=['POST','GET'])defget_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 的一些主要特点：

1. 基于 greenlet 的协程：Gevent 利用 greenlet 实现协程，使得代码在等待 I/O 操作时能够被调度到其他 greenlet 上执行。
2. 非阻塞 I/O：Gevent 使用 libevent 库实现非阻塞 I/O，可以处理大量并发连接。
3. 透明地支持同步和异步代码：使用 Gevent，开发者可以以同步的方式编写异步代码，无需修改现有代码或使用回调函数。
4. 事件循环：Gevent 使用了类似 Node.js 的事件循环模型，可以轻松地处理高并发场景。
5. 集成其他库：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("/")defhomepage():return"Hello World!"@app.route("/predict", methods=["POST"])defpredict():if flask.request.method =="POST":
        json_data = request.get_data()
        params = json.loads(json_data)
        img_card = params['image']return'koukou'defrun(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()defserver_forever():
            mulserver.start_accepting()
            mulserver._stop_event.wait()for i inrange(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 的主要特点和详解：

1. WSGI HTTP 服务器：Gunicorn 是一个 WSGI HTTP 服务器，可以与各种 Web 框架（如 Flask、Django 等）无缝集成。它使用预fork模式，能够在启动时预先fork出指定数量的 worker 进程来处理请求。
2. 高性能：由于采用了预fork模式，Gunicorn 在处理请求时具有高性能。它还支持多线程或多进程，可以根据实际需求进行配置。
3. 简单易用：安装和使用 Gunicorn 都非常简单。可以通过 pip 安装，然后通过命令行运行。同时，Gunicorn 还支持配置文件，可以方便地配置各种参数。
4. 兼容性强：Gunicorn 与大多数 Web 框架兼容，如 Flask、Django、Pyramid 等。它还支持各种WSGI应用，可以与各种 Python 库和工具集成。
5. 扩展性高：如果需要更多的功能，可以通过扩展 Gunicorn 或使用其他第三方工具与 Gunicorn 集成。

安装

pip install gunicorn

注意：gunicorn是Linux特有的库，建议使用Linux。

示例

创建gunicorn配置文件，在项目跟目录创建一个gunicorn.py文件，代码如下：

import gevent.monkey
gevent.monkey.patch_all()import multiprocessing
import os

ifnot 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'])defget_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

参考文章：

https://zhuanlan.zhihu.com/p/337749105