Python mysql（使用pymysql）自动重连
我不太确定这是否可行，但我想找一种方法，当与mysql数据库的连接断开时，能够重新连接。所有的连接都保存在一个gevent队列里，但我觉得这应该没什么关系。我相信如果我花点时间，我能想出一种重新连接数据库的方法。不过，我在看pymysql的代码时，发现Connection类里有一个'ping'方法，但我不太确定这个方法该怎么用。
这个方法看起来第一次可以重新连接，但之后又把重新连接的标志改回了False？我可以用这个方法吗，还是说有其他方法可以在连接断开时重新建立连接？即使不是用pymysql，大家通常是怎么处理数据库服务器宕机的情况，以及如何重新连接mysql服务器的呢？
def ping(self, reconnect=True):
''' Check if the server is alive '''
if self.socket is None:
if reconnect:
self._connect()
reconnect = False
else:
raise Error("Already closed")
try:
self._execute_command(COM_PING, "")
return self._read_ok_packet()
except Exception:
if reconnect:
self._connect()
return self.ping(False)
else:
raise
6 个回答
1
虽然这个问题有点老，但我遇到了一个类似的情况，就是在程序中访问托管的数据库。最后我找到的解决办法是创建一个装饰器，这样在进行查询时可以自动重新连接。
假设有一个连接数据库的函数：
def connect(self):
self.conn = mysql.connector.connect(host=self.host, user=self.user,
database=self.database, password=self.password)
self.cursor = self.conn.cursor()
print("Established connectionn...")
我创建了
def _reconnect(func):
@wraps(func)
def rec(self,*args,**kwargs):
try:
result = func(self,*args,**kwargs)
return result
except (mysql.connector.Error, mysql.connector.Warning) as e:
self.connect()
result = func(self,*args,**kwargs)
return result
return rec
这样，任何使用这个连接的函数现在都可以像这样被装饰
@_reconnect
def check_user_exists(self,user_id):
self.cursor.execute("SELECT COUNT(*) FROM _ where user_id={};".format(user_id))
if self.cursor.fetchall()[0][0]==0:
return False
else:
return True
这个装饰器会重新建立连接，并重新执行任何涉及数据库查询的函数。
1
这个逻辑很简单，如果连接断开了，就尝试重新连接几次。在这个例子中，我设置了最多尝试15次来重新连接或者发送信号。
import pymysql, pymysql.cursors
conn = pymysql.connect(
host=hostname,
user=username,
password=password,
db=dbname,
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor,
)
cursor = conn.cursor()
# you can do transactions to database and when you need conn later, just make sure the server is still connected
if conn.open is False:
max_try = 15
try = 0
while conn.open is False:
if try < max_try:
conn.ping() # autoreconnect is true by default
try +=1
# check the conn again to make sure it connected
if conn.open:
# statements when conn is successfully reconnect to the server
else:
# it must be something wrong : server, network etc
3
最简单的方法是在发送查询之前检查一下连接。
你可以通过创建一个小类来实现，这个类里包含两个方法：connect 和 query：
import pymysql
import pymysql.cursors
class DB:
def connect(self):
self.conn = pymysql.connect(
host=hostname,
user=username,
password=password,
db=dbname,
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor,
port=3306)
def query(self, sql):
try:
cursor = self.conn.cursor()
cursor.execute(sql)
except pymysql.OperationalError:
self.connect()
cursor = self.conn.cursor()
cursor.execute(sql)
return cursor
db = DB()
现在，每当你使用 db.query("example SQL") 发送查询时，这个请求会自动准备好处理连接错误。如果需要，它会用 self.connect() 重新连接。
记住： 这是一个简化的例子。通常情况下，你会希望让 PyMySQL 帮你处理查询中的特殊字符。要做到这一点，你需要在 query 方法中添加第二个参数，然后再继续。
23
我在我的应用程序中也遇到了同样的问题，我在PyMySQL的文档上找到了一种方法，可以向服务器发送一个“信号”，检查连接是否已经关闭。如果连接关闭了，它会自动重新连接。
from pymysql import connect
from pymysql.cursors import DictCursor
# create the connection
connection = connect(host='host', port='port', user='user',
password='password', db='db',
cursorclass=DictCursor)
# get the cursor
cursor = connection.cursor()
# if the connection was lost, then it reconnects
connection.ping(reconnect=True)
# execute the query
cursor.execute(query)
希望这对你有帮助。
15
终于找到一个可用的解决方案，希望能对某些人有所帮助。
from gevent import monkey
monkey.patch_socket()
import logging
import gevent
from gevent.queue import Queue
import pymysql as db
logging.basicConfig(level=logging.DEBUG)
LOGGER = logging.getLogger("connection_pool")
class ConnectionPool:
def __init__(self, db_config, time_to_sleep=30, test_run=False):
self.username = db_config.get('user')
self.password = db_config.get('password')
self.host = db_config.get('host')
self.port = int(db_config.get('port'))
self.max_pool_size = 20
self.test_run = test_run
self.pool = None
self.time_to_sleep = time_to_sleep
self._initialize_pool()
def get_initialized_connection_pool(self):
return self.pool
def _initialize_pool(self):
self.pool = Queue(maxsize=self.max_pool_size)
current_pool_size = self.pool.qsize()
if current_pool_size < self.max_pool_size: # this is a redundant check, can be removed
for _ in xrange(0, self.max_pool_size - current_pool_size):
try:
conn = db.connect(host=self.host,
user=self.username,
passwd=self.password,
port=self.port)
self.pool.put_nowait(conn)
except db.OperationalError, e:
LOGGER.error("Cannot initialize connection pool - retrying in {} seconds".format(self.time_to_sleep))
LOGGER.exception(e)
break
self._check_for_connection_loss()
def _re_initialize_pool(self):
gevent.sleep(self.time_to_sleep)
self._initialize_pool()
def _check_for_connection_loss(self):
while True:
conn = None
if self.pool.qsize() > 0:
conn = self.pool.get()
if not self._ping(conn):
if self.test_run:
self.port = 3306
self._re_initialize_pool()
else:
self.pool.put_nowait(conn)
if self.test_run:
break
gevent.sleep(self.time_to_sleep)
def _ping(self, conn):
try:
if conn is None:
conn = db.connect(host=self.host,
user=self.username,
passwd=self.password,
port=self.port)
cursor = conn.cursor()
cursor.execute('select 1;')
LOGGER.debug(cursor.fetchall())
return True
except db.OperationalError, e:
LOGGER.warn('Cannot connect to mysql - retrying in {} seconds'.format(self.time_to_sleep))
LOGGER.exception(e)
return False
# test (pytest compatible) -------------------------------------------------------------------------------------------
import logging
from src.py.ConnectionPool import ConnectionPool
logging.basicConfig(level=logging.DEBUG)
LOGGER = logging.getLogger("test_connection_pool")
def test_get_initialized_connection_pool():
config = {
'user': 'root',
'password': '',
'host': '127.0.0.1',
'port': 3305
}
conn_pool = ConnectionPool(config, time_to_sleep=5, test_run=True)
pool = conn_pool.get_initialized_connection_pool()
# when in test run the port will be switched back to 3306
# so the queue size should be 20 - will be nice to work
# around this rather than test_run hack
assert pool.qsize() == 20