前言:由于工作需要使用flask写了一个简易的http服务用来提供接口,按照接口文档demo写好以后本地测试一切正常,但是发布到服务器以后有一串警告:WARNING:This is a developnent server. Do not use it in a production deploynent,如下图:
意思是我的这个启动方式不能在生产环境上使用,然后带着疑问上网查了一下,我的启动方式是
app.run(host="0.0.0.0", port=80)
只适用于开发模式,因为它是单线程的,生产环境影响性能,替代方案是可以用uWSGI或者pywsgi
三者的区别如下:
1.app.run 启动的是单线程服务,性能很低
2.pywsgi服务器使用的是gevent的pywsgi模块,性能不错,配置也很简单,但是它只是把单线程改造成了单线程异步方式
3.uWSGI性能最好,配置稍微比上面难一点,但是它是支持多进程、多线程、和多协程的方式,简直就是完美,所以我选择尝试使用uWSGI服务器来替代
一、pywsgi方式实现简单,下面我也一下步骤,非常简单
1.安装gevent模块
pip install gevent
2.在启动类里引入模块
from gevent import pywsgi
3.main方法里将app.run替换
server = pywsgi.WSGIServer(('0.0.0.0', 80), app)
server.serve_forever()
复制代码
二、uWSGI实现方式以前需要引入nginx,现在也不需要了,很简单的实现,如下
1.安装uWSGI模块
pip install uwsgi
复制代码
2.在根目录下创建uWSGI配置文件 【uwsgi.ini】
[uwsgi]
# 地址端口
http = 0.0.0.0:80
# 项目路径
chdir = /root/projectname
# 项目启动文件
wsgi-file = manage.py
# 项目需要调用的启动类
callable = app
# 进程线程设置
processes = 4
threads = 10
# 日志文件
daemonize = /app/logs/uwsgi.log
# 保存主进程pid文件
pidfile = uwsgi.pid
# 是否需要主进程
master = true
复制代码
3.相关指令
#启动
uwsgi --ini uwsgi.ini
#重新加载
uwsgi --reload uwsgi.pid
#停止
uwsgi --stop uwsgi.pid
复制代码
4.附代码
manage.py
# coding:utf-8from app import create_app
app = create_app()if __name__ =='__main__':
app.run(host="0.0.0.0", port=80)
复制代码
app/init.py
# coding:utf-8import os
import sys
from apscheduler.schedulers.background import BackgroundScheduler
from flask import Flask as _Flask, request, jsonify
from flask_apscheduler import APScheduler
from app.extensions import*from app import config
from app.api import api
import fcntl
import atexit
classFlask(_Flask):defwsgi_app(self, environ, start_response):
ctx = self.request_context(environ)
ctx.push()
error =Nonetry:try:
response = self.full_dispatch_request()except Exception as e:
error = e
response = self.handle_exception(e)except:
error = sys.exc_info()[1]raisereturn response(environ, start_response)finally:if self.should_ignore_error(error):
error =None
ctx.auto_pop(error)defcreate_app():
app = Flask(__name__, static_folder=os.path.join(os.path.dirname(os.path.abspath(__file__)),'..','static'))
app.config.from_object(config)
blueprints_fabric(app)
run_task(app)
extensions_fabric(app)
process_request(app)
app.json_encoder = CustomJSONEncoder
return app
classCustomJSONEncoder(JSONEncoder):defdefault(self, obj):ifisinstance(obj,bytearray):returnstr(obj, encoding='utf-8')ifisinstance(obj, datetime):return obj.strftime('%Y-%m-%dT%H:%M:%S')else:return JSONEncoder.default(self, obj)defjob1():print("job1")defrun_task(app):# 执行一次"""Configure Scheduler"""
f =open("scheduler.lock","wb")try:
fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)print("********************************run_task**********************************")
scheduler = APScheduler(scheduler=BackgroundScheduler(timezone='Asia/Shanghai'))
scheduler.init_app(app)
scheduler.start()@scheduler.task('interval',id='do_job_1', seconds=10, misfire_grace_time=900)defdo_job_1():
job1()except:passdefunlock():
fcntl.flock(f, fcntl.LOCK_UN)
f.close()
atexit.register(unlock)defblueprints_fabric(app):
blueprints =[api]for blueprint in blueprints:
app.register_blueprint(blueprint)defextensions_fabric(app):
csrf.init_app(app)
cors.init_app(app, origins=config.CORS_ORIGINS, supports_credentials=True)defprocess_request(app):@app.before_requestdeftest():if request.path =='/test':return jsonify({"code":200,"message":"success"})
复制代码
app/extensions.py
from flask_wtf.csrf import CSRFProtect
cors = CORS()
csrf = CSRFProtect()
复制代码
版权归原作者 unhejing 所有, 如有侵权,请联系我们删除。