• Flask 数据库 连接池、DBUtils、http 连接池

    1、DBUtils 简介、使用

    DBUtils 简介

    DBUtils 是一套用于管理 数据库 "连接池" 的Python包，为 "高频度、高并发"  的数据库访问提供更好的性能，可以自动管理连接对象的创建和释放。并允许对非线程安全的数据库接口进行线程安全包装和连接。该连接可在各种多线程环境中使用。

    使用场景：如果使用的是流行的对象关系映射器 SQLObject 或 SQLAlchemy 之一，则不需要 DBUtils，因为它们带有自己的连接池。SQLObject 2 (SQL-API) 实际上是从 DBUtils 中借用了一些代码，将池化分离到一个单独的层中。

    DBUtils 提供两种外部接口：

    • PersistentDB ：提供线程专用的数据库连接，并自动管理连接。
    • PooledDB ：提供线程间可共享的数据库连接，并自动管理连接。

    另外，实际使用的数据库驱动也有所依赖，比如SQLite数据库只能使用PersistentDB作连接池。 下载地址：http://www.webwareforpython.org/downloads/DBUtils/

    使用 DBUtils 数据库 连接池

    安装：pip install DBUtils

    示例：MySQLdb 模块使用

    连接池对象只初始化一次，一般可以作为模块级代码来确保。 PersistentDB 的连接例子:

    1. import DBUtils.PersistentDB
    2.  
    3. # maxusage 则为一个连接最大使用次数
    4. persist = DBUtils.PersistentDB.PersistentDB(dbpai=MySQLdb,maxusage=1000,**kwargs)
    5.  
    6. # 获取连接池
    7. conn = persist.connection() 
    8.  
    9. # 关闭连接池
    10. conn.close()

    参数 dbpai 指定使用的数据库模块，兼容 DB-API 。下面是支持 DB-API 2 规范的数据库模块

    pip install pymysql（mysql）
    pip install pymssql（sqlserver）
    pip install cx_Oracle（oracle）
    pip install phoenixdb（hbase）
    pip install sqlite3（sqlite3 python自带）

    DBUtils 仅提供给了连接池管理，实际的数据库操作依然是由符合 DB-API 2 标准的目标数据库模块完成的。

    示例：pymysql 模块使用

    PooledDB 使用方法同 PersistentDB，只是参数有所不同。

    • dbapi ：数据库接口
    • mincached ：启动时开启的空连接数量
    • maxcached ：连接池最大可用连接数量
    • maxshared ：连接池最大可共享连接数量
    • maxconnections ：最大允许连接数量
    • blocking ：达到最大数量时是否阻塞
    • maxusage ：单个连接最大复用次数
    • setsession ：用于传递到数据库的准备会话，如 [”set name UTF-8″] 。

    conn = pooled.connection()
    cur = conn.cursor()
    cur.execute(sql)
    res = cur.fetchone()
    cur.close()   # 或者 del cur
    conn.close()  # 或者 del conn

    1. import pymysql
    2. from dbutils.pooled_db import PooledDB
    3.  
    4. # 定义连接参数
    5. pool = PooledDB(
    6.     creator=pymysql,
    7.     maxconnections=6,
    8.     mincached=2,
    9.     maxcached=5,
    10.     blocking=True,
    11.     host='localhost',
    12.     user='root',
    13.     passwd='123456',
    14.     db='mydb',
    15.     port=3306,
    16.     charset='utf8mb4'
    17. )
    18.  
    19.  
    20. def main():
    21.     # 从连接池获取连接
    22.     conn = pool.connection()
    23.     cursor = conn.cursor()
    24.  
    25.     # 执行 SQL 语句
    26.     sql = "SELECT * FROM students"
    27.     cursor.execute(sql)
    28.     result = cursor.fetchall()
    29.  
    30.     # 处理查询结果
    31.     for row in result:
    32.         print(row)
    33.  
    34.     # 关闭游标和连接
    35.     cursor.close()
    36.     conn.close()
    37.  
    38.  
    39. if __name__ == '__main__':
    40.     main()

    示例：面向对象 使用 DBUtils

    1. """
    2. 使用DBUtils数据库连接池中的连接，操作数据库
    3. """
    4. import json
    5. import pymysql
    6. import datetime
    7. from DBUtils.PooledDB import PooledDB
    8. import pymysql
    9.  
    10.  
    11. class MysqlClient(object):
    12.     __pool = None;
    13.  
    14.     def __init__(self, mincached=10, maxcached=20, maxshared=10, maxconnections=200, blocking=True,
    15.                  maxusage=100, setsession=None, reset=True,
    16.                  host='127.0.0.1', port=3306, db='test',
    17.                  user='root', passwd='123456', charset='utf8mb4'):
    18.         """
    20.         :param mincached:连接池中空闲连接的初始数量
    21.         :param maxcached:连接池中空闲连接的最大数量
    22.         :param maxshared:共享连接的最大数量
    23.         :param maxconnections:创建连接池的最大数量
    24.         :param blocking:超过最大连接数量时候的表现，为True等待连接数量下降，为false直接报错处理
    25.         :param maxusage:单个连接的最大重复使用次数
    26.         :param setsession:optional list of SQL commands that may serve to prepare
    27.             the session, e.g. ["set datestyle to ...", "set time zone ..."]
    28.         :param reset:how connections should be reset when returned to the pool
    29.             (False or None to rollback transcations started with begin(),
    30.             True to always issue a rollback for safety's sake)
    31.         :param host:数据库ip地址
    32.         :param port:数据库端口
    33.         :param db:库名
    34.         :param user:用户名
    35.         :param passwd:密码
    36.         :param charset:字符编码
    37.         """
    38.  
    39.         if not self.__pool:
    40.             self.__class__.__pool = PooledDB(pymysql,
    41.                                              mincached, maxcached,
    42.                                              maxshared, maxconnections, blocking,
    43.                                              maxusage, setsession, reset,
    44.                                              host=host, port=port, db=db,
    45.                                              user=user, passwd=passwd,
    46.                                              charset=charset,
    47.                                              cursorclass=pymysql.cursors.DictCursor
    48.                                              )
    49.         self._conn = None
    50.         self._cursor = None
    51.         self.__get_conn()
    52.  
    53.     def __get_conn(self):
    54.         self._conn = self.__pool.connection();
    55.         self._cursor = self._conn.cursor();
    56.  
    57.     def close(self):
    58.         try:
    59.             self._cursor.close()
    60.             self._conn.close()
    61.         except Exception as e:
    62.             print(e)
    63.  
    64.     def __execute(self, sql, param=()):
    65.         count = self._cursor.execute(sql, param)
    66.         print(count)
    67.         return count
    68.  
    69.     @staticmethod
    70.     def __dict_datetime_obj_to_str(result_dict):
    71.         """把字典里面的datatime对象转成字符串，使json转换不出错"""
    72.         if result_dict:
    73.             result_replace = {k: v.__str__() for k, v in result_dict.items() if isinstance(v, datetime.datetime)}
    74.             result_dict.update(result_replace)
    75.         return result_dict
    76.  
    77.     def select_one(self, sql, param=()):
    78.         """查询单个结果"""
    79.         count = self.__execute(sql, param)
    80.         result = self._cursor.fetchone()
    81.         """:type result:dict"""
    82.         result = self.__dict_datetime_obj_to_str(result)
    83.         return count, result
    84.  
    85.     def select_many(self, sql, param=()):
    86.         """
    87.         查询多个结果
    88.         :param sql: qsl语句
    89.         :param param: sql参数
    90.         :return: 结果数量和查询结果集
    91.         """
    92.         count = self.__execute(sql, param)
    93.         result = self._cursor.fetchall()
    94.         """:type result:list"""
    95.         [self.__dict_datetime_obj_to_str(row_dict) for row_dict in result]
    96.         return count, result
    97.  
    98.     def execute(self, sql, param=()):
    99.         count = self.__execute(sql, param)
    100.         return count
    101.  
    102.     def begin(self):
    103.         """开启事务"""
    104.         self._conn.autocommit(0)
    105.  
    106.     def end(self, option='commit'):
    107.         """结束事务"""
    108.         if option == 'commit':
    109.             self._conn.autocommit()
    110.         else:
    111.             self._conn.rollback()
    112.  
    113.  
    114. if __name__ == "__main__":
    115.     mc = MysqlClient()
    116.     sql1 = 'SELECT * FROM shiji  WHERE  id = 1'
    117.     result1 = mc.select_one(sql1)
    118.     print(json.dumps(result1[1], ensure_ascii=False))
    119.  
    120.     sql2 = 'SELECT * FROM shiji  WHERE  id IN (%s,%s,%s)'
    121.     param = (2, 3, 4)
    122.     print(json.dumps(mc.select_many(sql2, param)[1], ensure_ascii=False))

    不用 连接池

    1. import MySQLdb
    2. conn= MySQLdb.connect(host='localhost',user='root',passwd='pwd',db='myDB',port=3306)  
    3. #import pymysql
    4. #conn = pymysql.connect(host='localhost', port='3306', db='game', user='root', password='123456', charset='utf8')
    5. cur=conn.cursor()
    6. SQL="select * from table1"
    7. r=cur.execute(SQL)
    8. r=cur.fetchall()
    9. cur.close()
    10. conn.close()

    使用 连接池

    1. import MySQLdb
    2. from DBUtils.PooledDB import PooledDB
    3. #5为连接池里的最少连接数
    4. pool = PooledDB(MySQLdb,5,host='localhost',user='root',passwd='pwd',db='myDB',port=3306) 
    5.  
    6. # 以后每次需要数据库连接就是用connection（）函数获取连接就好了
    7. conn = pool.connection()  
    8. cur=conn.cursor()
    9. SQL="select * from table1"
    10. r=cur.execute(SQL)
    11. r=cur.fetchall()
    12. cur.close()
    13. conn.close()

    多线程 使用 连接池

    1. import sys
    2. import threading
    3. import MySQLdb
    4. import DBUtils.PooledDB
    5.  
    6. connargs = { "host":"localhost", "user":"user1", "passwd":"123456", "db":"test" }
    7. def test(conn):
    8.     try:
    9.         cursor = conn.cursor()
    10.         count = cursor.execute("select * from users")
    11.         rows = cursor.fetchall()
    12.         for r in rows: pass
    13.     finally:
    14.         conn.close()
    15.         
    16. def testloop():
    17.     print ("testloop")
    18.     for i in range(1000):
    19.         conn = MySQLdb.connect(**connargs)
    20.         test(conn)
    21.         
    22. def testpool():
    23.     print ("testpool")
    24.     pooled = DBUtils.PooledDB.PooledDB(MySQLdb, **connargs)
    25.     for i in range(1000):
    26.         conn = pooled.connection()
    27.         test(conn)
    28.         
    29. def main():
    30.     t = testloop if len(sys.argv) == 1 else testpool
    31.     for i in range(10):
    32.         threading.Thread(target = t).start()
    33.         
    34. if __name__ == "__main__":
    35.     main()

    虽然测试方式不是很严谨，但从测试结果还是能感受到 DBUtils 带来的性能提升。当然，我们我们也可以在 testloop() 中一直重复使用一个不关闭的 Connection，但这却不适合实际开发时的情形。

    2、Flask 配置，蓝图，数据库连接池，上下文原理

    Flask之配置文件，蓝图，数据库连接池，上下文原理：https://www.cnblogs.com/yunweixiaoxuesheng/p/8418135.html

    配  置

    方式一，使用字典方式配置

    app.config['SESSION_COOKE_NAME'] = 'session_liling'

    方式二，引入文件，设置

    from flask import Flask

    app = Flask(__name__)

    app.config.from_pyfile('settings.py')    # 引用settings.py中的AAAA
    print(app.config['AAAA'])        # 123

    # settings.py
    AAAA = 123

    方法三，使用环境变量设置，推荐使用

    from flask import Flask

    app = Flask(__name__)

    import os

    os.environ['FLASK-SETTINGS'] = 'settings.py'

    app.config.from_envvar('FLASK-SETTINGS')

    方式四，通过对象方式导入使用,可根据不同环境选择不同的配置，推荐使用

    from flask import Flask

    app = Flask(__name__)

    app.config.from_object('settings.BaseConfig')
    print(app.config['NNNN'])  # 123


    # settings.py

    class BaseConfig(object):  # 公用配置
        NNNN = 123


    class TestConfig(object):
        DB = '127.0.0.1'


    class DevConfig(object):
        DB = '192.168.1.1'


    class ProConfig(object):
        DB = '47.18.1.1'

    不同的文件 引用配置

    1. from flask import Flask,current_app
    2.  
    3. app = Flask(__name__)
    4.  
    5. app.secret_key = 'adfadsfhjkhakljsdfh'
    6.  
    7. app.config.from_object('settings.BaseConfig')
    8.  
    9. @app.route('/index')
    10. def index():
    11.     print(current_app.config['NNNN'])
    12.     return 'xxx'
    13.  
    14. if __name__ == '__main__':
    15.     app.run()

    instance_path、instance_relative_config

    1. from flask import Flask,current_app
    2.  
    3. app = Flask(__name__,instance_path=None,instance_relative_config=False)
    4. # 默认 instance_relative_config = False 那么 instance_relative_config和instance_path都不会生效
    5. # instance_relative_config=True，instance_path才会生效，app.config.from_pyfile('settings.py')将会失效
    6. # 配置文件找的路径，按instance_path的值作为配置文件路径
    7. # 默认instance_path=None，None会按照当前路径下的instance文件夹为配置文件的路径
    8. # 如果设置路径，按照设置的路径查找配置文件。
    9.  
    10. app.config.from_pyfile('settings.py')
    11.  
    12. @app.route('/index')
    13. def index():
    14.     print(current_app.config['NNNN'])
    15.     return 'xxx'
    16.  
    17.  
    18. if __name__ == '__main__':
    19.     app.run()

    蓝  图

    对应用程序的目录结构进行分配，一般适用于小中型企业

    代码：

    1. # crm/__init__.py
    2. # 创建flask项目，用蓝图注册不同的模块
    3.  
    4. from flask import Flask
    5. from .views import account
    6. from .views import order
    7.  
    8. app = Flask(__name__)
    9.  
    10. app.register_blueprint(account.account)
    11. app.register_blueprint(order.order)
    12.  
    13. ------------------------------------------------------------------------
    14. # manage.py
    15. # 启动文件
    16. import crm
    17.  
    18. if __name__ == '__main__':
    19.     crm.app.run()
    20.  
    21. ------------------------------------------------------------------------
    22. # crm/views/account.py
    23. # 视图函数模块,Blueprint，将函数引入app
    24. from flask import Blueprint
    25.  
    26. account = Blueprint('account',__name__,url_prefix='/xxx')
    27.  
    28. @account.route('/login')
    29. def login():
    30.     return 'Login'

    Flask 数据库 连接池

    ：https://www.cnblogs.com/TheLand/p/9178305.html

    ORM ( 对象关系映射 )

    ORM（Object-Relational Mapping，对象关系映射）是一种编程技术，用于在关系型数据库和面向对象编程语言之间建立映射关系。它允许开发人员使用面向对象的方式来操作数据库，而无需直接编写或执行 SQL 查询。

    ORM 提供了一个抽象层，将数据库表格映射为对象，并提供了一组方法和工具，以便于进行数据库的增删改查操作。开发人员可以通过使用对象和方法来表示和操作数据，而不必关心底层的 SQL 语句和数据库细节。

    常见的 ORM 框架包括：

    • SQLAlchemy：是 python 操作数据库的一个库，能够进行 orm 映射，是一个功能强大的 Python ORM 框架，支持多种数据库后端，提供了高级的查询功能和事务管理等特性。是为高效和高性能的数据库访问设计，实现了完整的企业级持久模型。SQLAlchemy 的理念是，SQL 数据库的量级和性能重要于对象集合；而对象集合的抽象又重要于表和行。
    • Flask-SQLAlchemy：Flask-SQLAlchemy 是一个与 Flask 框架集成的 SQLAlchemy 扩展，它简化了在 Flask 应用程序中使用 SQLAlchemy 进行数据库操作的过程。它提供了一组简单而强大的工具和功能，使得与数据库的交互变得更加轻松和高效。
    • Django ORM：Django 框架自带的 ORM，提供了简单易用的接口，支持多种数据库后端，并具有强大的查询和模型关联功能。
    • Hibernate：Java 领域中最流行的 ORM 框架，为 Java 对象和关系型数据库之间提供了映射和管理。

    使用 ORM 的好处包括：

    • 提高开发效率：ORM 提供了面向对象的编程接口，使得开发人员能够更快速地进行数据库操作，减少了编写和调试 SQL 语句的工作量。
    • 跨数据库平台：ORM 框架通常支持多种数据库后端，使得开发人员能够轻松地切换或同时使用不同的数据库系统。
    • 数据库抽象和安全性：ORM 隐藏了底层的数据库细节，提供了一层抽象，有助于维护和管理数据库结构，并提供了安全性保护，如参数绑定和防止 SQL 注入。
    • 更好的可维护性和可测试性：使用 ORM 可以提高代码的可读性和可维护性，使得进行单元测试和集成测试更加容易。

    注意：ORM 并不能解决所有数据库问题。在某些情况下，复杂的查询和性能要求可能需要直接使用原生 SQL。因此，根据具体的需求和场景，谨慎选择和使用合适的 ORM 框架。

    为什么 使用 数据库 连接池

    • 多连接：如果不用连接池时，每次操作都要链接数据库，链接次数过多，数据库会耗费过多资源，数量过大的话，数据库会过载，导致程序运行缓慢。
    • 单连接：在程序中全局创建连接，导致程序会一直使用一个连接，避免了反复连接造成的问题。但是多线程时就得加锁。这样就变成串行，没法实现并发

    解决方法：

    • 方式 1：为每一个线程创建一个链接（是基于本地线程来实现的。thread.local），每个线程独立使用自己的数据库链接，该线程关闭不是真正的关闭，本线程再次调用时，还是使用的最开始创建的链接，直到线程终止，数据库链接才关闭。如果线程比较多还是会创建很多连接
    • 方式 2：创建一个链接池，为所有线程提供连接，使用时来进行获取，使用完毕后在放回到连接池。假设最大链接数有10个，其实也就是一个列表，当你pop一个，人家会在append一个，链接池的所有的链接都是按照排队的这样的方式来链接的。链接池里所有的链接都能重复使用，共享的， 即实现了并发，又防止了链接次数太多

    基于 DBUtils 数据库连接池

    数据库连接池避免每次操作都要连接数据库，一直使用一个连接，多线程也会出现问题，可加锁，但变为串行

    1. import pymysql
    2. import threading
    3. from threading import RLock
    4.  
    5. LOCK = RLock()
    6. CONN = pymysql.connect(
    7.     host='127.0.0.1',
    8.     port=3306,
    9.     user='root',
    10.     password='123',
    11.     database='ok1',
    12.     charset='utf8'
    13. )
    14.  
    15.  
    16. def task(arg):
    17.     with LOCK:
    18.         cursor = CONN.cursor()
    19.         cursor.execute('select * from book')
    20.         result = cursor.fetchall()
    21.         cursor.close()
    22.         print(result)
    23.  
    24.  
    25. for i in range(10):
    26.     t = threading.Thread(target=task, args=(i,))
    27.     t.start()

    线程之间 的 数据隔离

    "本地线程" 可以实现线程之间的数据隔离。保证每个线程都只有自己的一份数据，在操作时不会影响别人的，即使是多线程，自己的值也是互相隔离的

    1. import threading
    2. import time
    3.  
    4. # 本地线程对象
    5. local_values = threading.local()
    6.  
    7.  
    8. def func(num):
    9.     """
    10.     # 第一个线程进来，本地线程对象会为他创建一个
    11.     # 第二个线程进来，本地线程对象会为他创建一个
    12.     {
    13.         线程1的唯一标识：{name:1},
    14.         线程2的唯一标识：{name:2},
    15.     }
    16.     :param num: 
    17.     :return: 
    18.     """
    19.     local_values.name = num  # 4
    20.     # 线程停下来了
    21.     time.sleep(2)
    22.     # 第二个线程： local_values.name，去local_values中根据自己的唯一标识作为key，获取value中name对应的值
    23.     print(local_values.name, threading.current_thread().name)
    24.  
    25.  
    26. for i in range(5):
    27.     th = threading.Thread(target=func, args=(i,), name='线程%s' % i)
    28.     th.start()

    模式一：每个线程创建一个连接

    基于threading.local实现创建每个连接。
    每个线程会创建一个连接，该线程没有真正关闭。
    再次调用该线程时，还是使用原有的连接。
    线程真正终止的时候，连接才会关闭。

    1. from DBUtils.PersistentDB import PersistentDB
    2. import pymysql
    3.  
    4. POOL = PersistentDB(
    5.     creator=pymysql,  # 使用链接数据库的模块
    6.     maxusage=None,  # 一个链接最多被重复使用的次数，None表示无限制
    7.     setsession=[],  # 开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."]
    8.     ping=0,
    9.     # ping MySQL服务端，检查是否服务可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = always
    10.     closeable=False,
    11.     # 如果为False时， conn.close() 实际上被忽略，供下次使用，再线程关闭时，才会自动关闭链接。如果为True时， conn.close()则关闭链接，那么再次调用pool.connection时就会报错，因为已经真的关闭了连接（pool.steady_connection()可以获取一个新的链接）
    12.     threadlocal=None,  # 本线程独享值得对象，用于保存链接对象，如果链接对象被重置
    13.     host='127.0.0.1',
    14.     port=3306,
    15.     user='root',
    16.     password='123',
    17.     database='pooldb',
    18.     charset='utf8'
    19. )
    20.  
    21.  
    22. def func():
    23.     # conn = SteadyDBConnection()
    24.     conn = POOL.connection()
    25.     cursor = conn.cursor()
    26.     cursor.execute('select * from tb1')
    27.     result = cursor.fetchall()
    28.     cursor.close()
    29.     conn.close() # 不是真的关闭，而是假的关闭。 conn = pymysql.connect()   conn.close()
    30.  
    31.     conn = POOL.connection()
    32.     cursor = conn.cursor()
    33.     cursor.execute('select * from tb1')
    34.     result = cursor.fetchall()
    35.     cursor.close()
    36.     conn.close()
    37.  
    38. import threading
    39.  
    40. for i in range(10):
    41.     t = threading.Thread(target=func)
    42.     t.start()

    模式二：线程复用连接池 (推荐)

    创建一个连接池，为所有线程提供连接，线程使用连接时获取连接，使用完毕放回连接池。
    线程不断地重用连接池里的连接。

    1. import time
    2. import pymysql
    3. import threading
    4. from DBUtils.PooledDB import PooledDB, SharedDBConnection
    5.  
    6. POOL = PooledDB(
    7.     creator=pymysql,  # 使用链接数据库的模块
    8.     maxconnections=6,  # 连接池允许的最大连接数，0和None表示不限制连接数
    9.     mincached=2,  # 初始化时，链接池中至少创建的空闲的链接，0表示不创建
    10.  
    11.     maxcached=5,  # 链接池中最多闲置的链接，0和None不限制
    12.     maxshared=3,
    13.     # 链接池中最多共享的链接数量，0和None表示全部共享。
    14.     # PS: 无用，因为pymysql和MySQLdb等模块的 threadsafety都为1，所有值无论设置为多少，
    15.     # _maxcached永远为0，所以永远是所有链接都共享。
    16.     blocking=True,  # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
    17.     maxusage=None,  # 一个链接最多被重复使用的次数，None表示无限制
    18.     setsession=[],  # 开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."]
    19.     # ping MySQL服务端，检查是否服务可用。
    20.     # 如：0 = None = never, 1 = default = whenever it is requested, 
    21.     # 2 = when a cursor is created, 4 = when a query is executed, 7 = always
    22.     ping=0,
    23.     host='127.0.0.1',
    24.     port=3306,
    25.     user='root',
    26.     password='123456',
    27.     database='flask_test',
    28.     charset='utf8'
    29. )
    30.  
    31.  
    32. def func():
    33.     # 检测当前正在运行连接数的是否小于最大链接数，如果不小于则：等待或报raise TooManyConnections异常
    34.     # 否则
    35.     # 则优先去初始化时创建的链接中获取链接 SteadyDBConnection。
    36.     # 然后将SteadyDBConnection对象封装到PooledDedicatedDBConnection中并返回。
    37.     # 如果最开始创建的链接没有链接，则去创建一个SteadyDBConnection对象，
    38.     # 再封装到PooledDedicatedDBConnection中并返回。
    39.     # 一旦关闭链接后，连接就返回到连接池让后续线程继续使用。
    40.  
    41.     # PooledDedicatedDBConnection
    42.     conn = POOL.connection()
    43.  
    44.     # print(th, '链接被拿走了', conn1._con)
    45.     # print(th, '池子里目前有', pool._idle_cache, '\r\n')
    46.  
    47.     cursor = conn.cursor()
    48.     cursor.execute('select * from userinfo')
    49.     result = cursor.fetchall()
    50.     print(result)
    51.     conn.close()
    52.  
    53.     conn = POOL.connection()
    54.  
    55.     # print(th, '链接被拿走了', conn1._con)
    56.     # print(th, '池子里目前有', pool._idle_cache, '\r\n')
    57.  
    58.     cursor = conn.cursor()
    59.     cursor.execute('select * from userinfo')
    60.     result = cursor.fetchall()
    61.     conn.close()
    62.  
    63.  
    64. func()

    上下文管理

    所谓上下文，像考试题目根据上下文，回答一下问题。
    程序中，泛指的外部环境，像wsgi来的网络请求，而且通常只有上文。
    flask中的上下文，被使用在 current_app，session，request 上。

    flask 本地线程

    1. from flask import session
    2.  
    3. try:
    4.     from greenlet import getcurrent as get_ident  # grenlet协程模块
    5. except ImportError:
    6.     try:
    7.         from thread import get_ident
    8.     except ImportError:
    9.         from _thread import get_ident  # get_ident()，获取线程的唯一标识
    10.  
    11.  
    12. class Local(object):  # 引用session中的LocalStack下的Local
    13.     __slots__ = ('__storage__', '__ident_func__')  # __slots__该类在外面调用时，只能调用定义的字段，其他的不能调用
    14.  
    15.     def __init__(self):
    16.         # object.__setattr__为self设置值，等价于self.__storage__ = {}
    17.         # 为父类object中包含的__steattr__方法中的self.__storage__ = {}
    18.         # 由于类内包含__steattr__，self.xxx（对象.xxx）时会自动会触发__steattr__，
    19.         # 当前__steattr__中storage = self.__storage__又会像self.xxx要值，故会造成递归
    20.         # 所以在父类中__steattr__方法赋值，避免self.xxx调用__setattr__造成的递归
    21.         object.__setattr__(self, '__storage__', {})
    22.  
    23.         object.__setattr__(self, '__ident_func__', get_ident)  # 赋值为协程
    24.  
    25.     def __iter__(self):
    26.         return iter(self.__storage__.items())
    27.  
    28.     def __release_local__(self):
    29.         self.__storage__.pop(self.__ident_func__(), None)
    30.  
    31.     def __getattr__(self, name):
    32.         try:
    33.             return self.__storage__[self.__ident_func__()][name]
    34.         except KeyError:
    35.             raise AttributeError(name)
    36.  
    37.     def __setattr__(self, name, value):
    38.         ident = self.__ident_func__()  # 获取单钱线程(协程)的唯一标识
    39.         storage = self.__storage__  # {}
    40.         try:
    41.             storage[ident][name] = value  # { 111 : {'stack':[] },222 : {'stack':[] } }
    42.         except KeyError:
    43.             storage[ident] = {name: value}
    44.  
    45.     def __delattr__(self, name):
    46.         try:
    47.             del self.__storage__[self.__ident_func__()][name]
    48.         except KeyError:
    49.             raise AttributeError(name)
    50.  
    51.  
    52. _local = Local()  # flask的本地线程功能，类似于本地线程，如果有人创建Local对象并，设置值，每个线程里一份
    53. _local.stack = []  # _local.stack会调用__setattr__的self.__ident_func__()取唯一标识等

    特殊栈

    1. from flask import session
    2.  
    3. try:
    4.     from greenlet import getcurrent as get_ident
    5. except ImportError:
    6.     try:
    7.         from thread import get_ident
    8.     except ImportError:
    9.         from _thread import get_ident  # 获取线程的唯一标识 get_ident()
    10.  
    11.  
    12. class Local(object):
    13.     __slots__ = ('__storage__', '__ident_func__')
    14.  
    15.     def __init__(self):
    16.         # self.__storage__ = {}
    17.         # self.__ident_func__ = get_ident
    18.         object.__setattr__(self, '__storage__', {})
    19.         object.__setattr__(self, '__ident_func__', get_ident)
    20.  
    21.     def __iter__(self):
    22.         return iter(self.__storage__.items())
    23.  
    24.     def __release_local__(self):
    25.         self.__storage__.pop(self.__ident_func__(), None)
    26.  
    27.     def __getattr__(self, name):
    28.         try:
    29.             return self.__storage__[self.__ident_func__()][name]
    30.         except KeyError:
    31.             raise AttributeError(name)
    32.  
    33.     def __setattr__(self, name, value):
    34.         ident = self.__ident_func__()  # 获取当前线程(协程)的唯一标识
    35.         storage = self.__storage__  # {}
    36.         try:
    37.             storage[ident][name] = value  # { 111:{'stack':[] },222:{'stack':[] }  }
    38.         except KeyError:
    39.             storage[ident] = {name: value}
    40.  
    41.     def __delattr__(self, name):
    42.         try:
    43.             del self.__storage__[self.__ident_func__()][name]
    44.         except KeyError:
    45.             raise AttributeError(name)
    46.  
    47.  
    48. _local = Local()
    49. _local.stack = []

    使用 flask 中的 stack 和 local

    1. from functools import partial
    2. from flask.globals import LocalStack, LocalProxy
    3.  
    4. _request_ctx_stack = LocalStack()
    5.  
    6.  
    7. class RequestContext(object):
    8.     def __init__(self, environ):
    9.         self.request = environ
    10.  
    11.  
    12. def _lookup_req_object(name):
    13.     top = _request_ctx_stack.top
    14.     if top is None:
    15.         raise RuntimeError(_request_ctx_stack)
    16.     return getattr(top, name)
    17.  
    18.  
    19. # 实例化了LocalProxy对象，_lookup_req_object参数传递
    20. session = LocalProxy(partial(_lookup_req_object, 'session'))
    21.  
    22. """
    23. local = {
    24.     “标识”: {'stack': [RequestContext(),]}
    25. }
    26. """
    27. _request_ctx_stack.push(RequestContext('c1'))  # 当请求进来时，放入
    28.  
    29. print(session)  # 获取 RequestContext('c1')， top方法
    30. print(session)  # 获取 RequestContext('c1')， top方法
    31. _request_ctx_stack.pop()  # 请求结束pop

    示例：

    1. from functools import partial
    2. from flask.globals import LocalStack, LocalProxy
    3.  
    4. ls = LocalStack()
    5.  
    6.  
    7. class RequestContext(object):
    8.     def __init__(self, environ):
    9.         self.request = environ
    10.  
    11.  
    12. def _lookup_req_object(name):
    13.     top = ls.top
    14.     if top is None:
    15.         raise RuntimeError(ls)
    16.     return getattr(top, name)
    17.  
    18.  
    19. session = LocalProxy(partial(_lookup_req_object, 'request'))
    20.  
    21. ls.push(RequestContext('c1')) # 当请求进来时，放入
    22. print(session) # 视图函数使用
    23. print(session) # 视图函数使用
    24. ls.pop() # 请求结束pop
    25.  
    26.  
    27. ls.push(RequestContext('c2'))
    28. print(session)
    29.  
    30. ls.push(RequestContext('c3'))
    31. print(session)

    Flask SQLAlchemy 使用 连接池

    Flask SQLAlchemy 提供了内置的连接池功能，可以方便地配置和使用。

    在Flask应用中，我们可以通过配置 SQLALCHEMY_POOL_SIZE 参数来设置连接池的大小。连接池的大小决定了同时打开的数据库连接的数量。例如，我们可以将连接池的大小设置为10：

    app.config['SQLALCHEMY_POOL_SIZE'] = 10

    1. from flask_sqlalchemy import SQLAlchemy
    2.  
    3. app = Flask(__name__)
    4. app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://user:password@localhost/db_name'
    5. db = SQLAlchemy(app)
    6.  
    7. # 创建模型类
    8. class User(db.Model):
    9.     id = db.Column(db.Integer, primary_key=True)
    10.     name = db.Column(db.String(50))
    11.  
    12. # 添加数据到数据库
    13. user = User(name='John')
    14. db.session.add(user)
    15. db.session.commit()
    16.  
    17. # 查询数据
    18. all_users = User.query.all()
    19.  
    20. # 更新数据
    21. user = User.query.filter_by(name='John').first()
    22. user.name = 'Jane'
    23. db.session.commit()
    24.  
    25. # 删除数据
    26. user = User.query.filter_by(name='Jane').first()
    27. db.session.delete(user)
    28. db.session.commit()

    3、flask http 连接池

    Flask 本身并不提供内置的 HTTP 连接池功能，但可以使用第三方库来实现在 Flask 中使用 HTTP 连接池。其中一个常用的库是 urllib3，它提供了高级的连接池管理功能。

    示例：在 Flask 中使用 urllib3 来创建和管理 HTTP 连接池：

    安装：pip install urllib3

    1. from flask import Flask
    2. import urllib3
    3.  
    4. app = Flask(__name__)
    5.  
    6. """
    7. 使用 urllib3.PoolManager() 创建了一个连接池管理器对象 http，然后使用 http.request() 方法发送了一个 GET 请求。
    8. 您可以根据需要进行配置和自定义，例如设置最大连接数、超时时间、重试策略等。以下是一个示例，展示了如何进行自定义设置：
    9. """
    10.  
    11.  
    12. @app.route('/index_1')
    13. def index_1():
    14.     http = urllib3.PoolManager()
    15.     response = http.request('GET', 'http://api.example.com')
    16.     return response.data
    17.  
    18.  
    19. """
    20. 对连接池进行了一些自定义配置，包括最大连接数、每个连接的最大数量、连接和读取的超时时间以及重试策略。
    21. 使用 urllib3 可以更好地控制和管理 HTTP 连接，提高 Flask 应用程序的性能和效率。
    22. """
    23.  
    24.  
    25. @app.route('/index_2')
    26. def index_2():
    27.     http = urllib3.PoolManager(
    28.         num_pools=10,  # 最大连接数
    29.         maxsize=100,  # 每个连接的最大数量
    30.         timeout=urllib3.Timeout(connect=2.0, read=5.0),  # 连接和读取的超时时间
    31.         retries=urllib3.Retry(total=3, backoff_factor=0.1, status_forcelist=[500, 502, 503, 504])  # 重试策略
    32.     )
    33.     response = http.request('GET', 'http://api.example.com')
    34.     return response.data

  • 相关阅读:
    mybatis-plus填充功能，自动设置值
    基于非对称纳什谈判的多微网电能共享运行优化策略（附带MATLAB程序）
    数仓建设教程
    SB树，看这一篇就够了
    QIBOX1-014-栏目的调用2
    前端面试常见问题总结
    【C++】C++11——C++11介绍、初始化列表、声明、auto、decltype、nullptr、范围for循环
    博客网页制作基础大二dw作业 web课程设计网页制作 个人网页设计与实现 我的个人博客网页开发
    NameNode的image和edits里面到底存放了些啥
    docker 部署vue
  • 原文地址：https://blog.csdn.net/freeking101/article/details/132876026