建设企业网站企业网上银行登录官网下载,长沙哪里有网站制作,php做不了大型网站,幸福人寿保险公司官方网站保单查询1 问题说明
1.1 任务简述
在开发Flask应用中一定会遇到执行耗时任务#xff0c;但是Flask是轻量级的同步框架#xff0c;即在单个请求时服务会阻被塞#xff0c;直到任务完成#xff08;注意#xff1a;当前请求被阻塞不会影响到其他请求#xff09;。
解决异步问题有…1 问题说明
1.1 任务简述
在开发Flask应用中一定会遇到执行耗时任务但是Flask是轻量级的同步框架即在单个请求时服务会阻被塞直到任务完成注意当前请求被阻塞不会影响到其他请求。
解决异步问题有两种思路一种是借助外部工具实现异步例如消息队列RabbitMQ、 异步任务队列CeleryRedis另一种借助Python中的进程、线程或协程解决异步。我的是小项目选择因此选择了第二种方法。
经过测试我在Flask中使用协程gevent会被阻塞使用进程multiprocessing不会被阻塞操作数据库出了问题有可能是我没操作正确的问题最后选择使用线程threading。
注意使用线程会出现线程安全问题
1.2 注意的问题
1线程安全 !!! 注意
1 由于flask-SQLAlchemy对SQLAlchemy进行了封装所以是线程安全的可以在线程中直接使用
2 原生SQLAlchemy中# 2.1 直接使用Session是线程不安全的不推荐使用self.session Session(self.engine)# 2.2 直接使用scoped_session是线程安全的推荐使用# 获取sessionmakersession_factory sessionmaker(engine)# scoped_session是线程安全的# 注意此session不能使用Query对象session scoped_session(session_factory)!!!2使用数据库
# 注意高版本的Flask-SQLAlchemy我的版本Flask-SQLAlchemy3.1.1SQLAlchemy2.0.16必须使用”with app.app_context()“本质原因是Flask关联的SQLAlchemy版本太高
# 否则无法插入数据库并报错错误内容如下低版本不会出现此问题This typically means that you attempted to use functionality that needed
the current application. To solve this, set up an application context
with app.app_context(). See the documentation for more information.2 工程布局
项目布局如下
3 源代码
1main.py
from blueprint import init_blueprint
from config_app import app
from config_db import init_mysql_db# 初始化MySQL
init_mysql_db()init_blueprint()if __name__ __main__:app.run(host0.0.0.0, debugTrue)2config_app.py
from flask import Flask
from flask_cors import CORSdef create_app():flask_app Flask(__name__)CORS(flask_app, supports_credentialsTrue)return flask_appapp create_app()
3config_db.py from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow# 添加pymysql驱动连接MySQL数据库
import pymysqlfrom config_app import apppymysql.install_as_MySQLdb()
!!! 注意
1 由于flask-SQLAlchemy对SQLAlchemy进行了封装所以是线程安全的可以在线程中直接使用
2 原生SQLAlchemy中# 2.1 直接使用Session是线程不安全的不推荐使用self.session Session(self.engine)# 2.2 直接使用scoped_session是线程安全的推荐使用# 获取sessionmakersession_factory sessionmaker(engine)# scoped_session是线程安全的# 注意此session不能使用Query对象session scoped_session(session_factory)!!!
# 创建MySQL单实例
mysql_db SQLAlchemy()# 创建Schema
mysql_schema Marshmallow()# 创建数据库create database async default character set utf8mb4 collate utf8mb4_unicode_ci;
class MysqlConf:acc rootpwd 123456host 192.168.108.200port 3306db asyncmysql_conf MysqlConf()# 初始化MySQL数据库
def init_mysql_db():# 配置MySQL数据库urldb_url mysql:// mysql_conf.acc : mysql_conf.pwd mysql_conf.host : str(mysql_conf.port) / mysql_conf.dbapp.config[SQLALCHEMY_DATABASE_URI] db_url# 关闭sqlalchemy自动跟踪数据库app.config[SQLALCHEMY_TRACK_MODIFICATIONS] False# 显示底层执行的SQL语句app.config[SQLALCHEMY_ECHO] True# 解决‘No application found. Either work inside a view function or push an application context.’app.app_context().push()# 初始化appmysql_db.init_app(app)# 初始化schemamysql_schema.init_app(app)# 初始化table
def init_table():# 删除表mysql_db.drop_all()# 创建表mysql_db.create_all()
4dao.py
import datetimefrom config_db import mysql_db as db# Create table of bm_record
class User(db.Model):# Record table__tablename__ as_userid db.Column(us_id, db.Integer, nullableFalse, primary_keyTrue, autoincrementTrue)name db.Column(us_name, db.String(100))age db.Column(us_age, db.Integer)create_time db.Column(us_create_time, db.DateTime, defaultdatetime.datetime.now)# 插入数据
def insert_record_dao(user: User):# Add datadb.session.add(user)# Commit datadb.session.commit()
5blueprint.py
# 构建蓝本
import threading
import time
from concurrent.futures import ThreadPoolExecutorfrom flask import Blueprint, jsonifyfrom config_app import app
from config_db import init_table
from dao import insert_record_dao, Useruser Blueprint(user, __name__)# 注册蓝本
def init_blueprint():app.register_blueprint(user, url_prefix/user)user.route(/initdb)
def init_db():init_table()return jsonify(success)user.route(/add)
def add_user():user: User User()user.name zhangsanuser.age 12time.sleep(10)insert_record_dao(user)return jsonify(user.id)executor ThreadPoolExecutor(3)
user.route(/thread_pool)
def run_task_thread_pool():executor.submit(_run_thread_pool)return jsonify(success)# 执行线程
def _run_thread_pool():print(Run thread pool)time.sleep(2)user: User User()user.name thread pooluser.age 12# 注意必须使用”with app.app_context()“否则无法插入数据库并且不会报错with app.app_context():insert_record_dao(user)print(End thread pool)passuser.route(/thread)
def run_task_thread():task threading.Thread(target_run_thread, namethread-01)task.start()return jsonify(success)# 执行线程
def _run_thread():print(Run thread)time.sleep(2)user: User User()user.name threaduser.age 12# 注意高版本的Flask-SQLAlchemy我的版本Flask-SQLAlchemy3.1.1SQLAlchemy2.0.16必须使用”with app.app_context()“本质原因是Flask关联的SQLAlchemy版本太高# 否则无法插入数据库并报错错误内容如下低版本不会出现此问题This typically means that you attempted to use functionality that neededthe current application. To solve this, set up an application contextwith app.app_context(). See the documentation for more information.with app.app_context():insert_record_dao(user)print(End thread)pass4 请求截图
1初始化数据库 2添加用户 3使用线程池 4使用线程
