|
|
from dbutils.pooled_db import PooledDB
import pymysql
# 创建连接池
mysql_pool = PooledDB(
creator=pymysql,
maxconnections=10,
mincached=2,
maxcached=5,
maxusage=100,
blocking=True,
host='localhost',
user='your_username',
password='your_password',
database='your_database',
charset='utf8mb4',
ping=1, # 每次取出连接时检查
autocommit=True
)
def get_data():
"""从连接池获取连接并执行cha询"""
conn = mysql_pool.connection()
try:
with conn.cursor() as cursor:
cursor.execute("SELECT * FROM your_table LIMIT 5")
return cursor.fetchall()
except Exception as e:
print(f"cha询失败: {e}")
raise
finally:
conn.close()
# 使用示例
if __name__ == "__main__":
try:
data = get_data()
print("cha询结果:", data)
except Exception as e:
print("操作失败:", e)
可以试一下这个 |
|