mysql建立的连接，在8小时内都没有访问请求的话，mysql server将主动断开这条连接。在使用pymysql或MySQLdb操作数据库连接时，当cursor一直处于连接状态，未及时close时，连接池被占用。查看后台日志：
计算
安全
数据库
网络和加速
企业服务
"MySQL server has gone away (%r)" % (e,))
pymysql.err.OperationalError: (2006, "MySQL server has gone away (TimeoutError(110, 'Connection timed out'))")代码中未在query操作及时close cursor，在每个连接中，均要有cursor.close() 和 conn.close()操作。即：
def db_execute(query):
conn = MySQLdb.connect(*)
cur = conn.cursor()
cur.execute(query)
res = cur.fetchall()
cur.close()
conn.close()
return res这样的话会有性能问题，推荐使用SqlAlchemy.pool。那mysql中有办法实现吗？我们试试多线程和协程。
class MysqlConnect(object):
"""
mysql connect 基类
"""
def __init__(self, db_params=cmdb_test_params, maxconn=5):
self.db_params = db_params
self.maxconn = maxconn
self.pool = Queue(maxconn)
for i in range(maxconn):
self.connect = self._connect()
self.commit()
self.cursor = self._cursor()
def _connect(self):
"""
mysql connect
:return cursor:
"""
key = ['host', 'port', 'user', 'password', 'database', 'charset']
if not all([True if k in self.db_params else False for k in key]):
raise Exception(list(self.db_params.keys()), "数据库连接失败，请检查配置参数")
try:
conn = pymysql.connect(**self.db_params)
conn.autocommit(True)
self.pool.put(self.connect)
except pymysql.Error as e:
logutil.Logger().error(e)
traceback.print_exc()
raise pymysql.Error("连接数据库失败 %s" % e)
self.connect = conn
return self.connect
def _cursor(self):
if self.connect:
conn = self.pool.get()
self.cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
else:
self._connect()
conn = self.pool.get()
self.cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
return self.cursor
def close(self):
if self.connect:
self.cursor.close() # 关闭游标，未及时close时，连接池被占用 error code 2006
self.pool.put(self.connect)
self.connect = None
def commit(self):
try:
if self.connect:
self.connect.autocommit(True)
except pymysql.Error as e:
logutil.Logger().error(e)
traceback.print_exc()
raise pymysql.Error("数据库提交失败 %s" % e)
finally:
self.close()
def rollback(self):
try:
if self.connect:
self.connect.rollback()
except pymysql.Error as e:
logutil.Logger().error(e)
traceback.print_exc()
raise pymysql.Error("数据库回滚失败 %s" % e)
finally:
if self.connect:
self.close()
def __del__(self):
self.commit()
def query_execute(self, sql):
try:
if self.connect is None:
self._connect()
self._cursor()
result_list = []
self.cursor.execute(sql)
for row in self.cursor.fetchall():
result_list.append(list(row))
return result_list
except pymysql.Error as e:
logutil.Logger().error(e)
traceback.print_exc()
raise pymysql.Error("数据库查询失败 %s" % e)
finally:
if self.connect:
self.close()
def dml_execute(self, sql):
try:
if self.connect is None:
self._connect()
self._cursor()
if self.cursor is None:
self._cursor()
self.cursor.execute(sql)
self.commit()
except pymysql.Error as e:
logutil.Logger().error(e)
traceback.print_exc()
self.rollback()
raise pymysql.Error("数据库执行dml失败 %s" % e)
finally:
self.close()
def dml_execute_many(self, sql):
try:
if self.connect is None:
self._connect()
self._cursor()
if self.cursor is None:
self._cursor()
self.cursor.executemany(sql)
self.commit()
except pymysql.Error as e:
logutil.Logger().error(e)
traceback.print_exc()
self.rollback()
raise pymysql.Error("数据库执行dml失败 %s" % e)
finally:
self.close()
def testmysqldb(self,ip,user,password,dbname,Strsql):
try:
self.connect = pymysql.connect(host=ip,user=user,passwd=password,charset='utf8')
self.connect.select_db(dbname)
self.query_execute(Strsql)
return True
except Exception as e:
print(("Error %d :%s" %(e.args[0],e.args[1])))
return False亿速云「云数据库 MySQL」免部署即开即用，比自行安装部署数据库高出1倍以上的性能，双节点冗余防止单节点故障，数据自动定期备份随时恢复。点击查看>>
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
