Python MySQL断开重连实现：Connection与Cursor对象解析及自动重连方案

半杯拿铁

发布时间：2026-10-13| 6人浏览过

原创

本文探讨Python中MySQL数据库连接断开后的重连机制。通过解析Connection和Cursor对象，分析超时设置失效原因，并介绍利用Connection.ping()方法实现自动重连的方案，帮助开发者解决数据库连接超时问题。

问题背景与现象

后台服务在运行过程中遇到一个问题，大约运行15分钟后，接口请求开始报错。


pymysql.err.InterfaceError: (0, '')

这个错误提示通常出现在将None赋值给多个变量的情况中，在定位问题时发现。


pymysql.err.OperationalError: (2013, 'Lost connection to MySQL server during query')

那么如何解决这个问题呢？

出现问题的代码如下。


class MysqlConnection(object):

	"""
	mysql操作类，对mysql数据库进行增删改查
	"""

	def __init__(self, config):
		# Connect to the database
		self.connection = pymysql.connect(**config)
		self.cursor = self.connection.cursor()

	def Query(self, sql):
		"""
		查询数据
		:param sql:
		:return:
		"""
		self.cursor.execute(sql)
		return self.cursor.fetchall()

Connection与Cursor对象解析

在分析问题之前，我们先了解一下Python中数据库操作的Connection和Cursor这两个核心对象。

以下是Python数据库交互的流程图。

我们可以将Connection和Cursor形象地比喻为数据库连接的两个关键部分。

Connection()构造函数的参数列表如下：

host，连接的数据库服务器主机名，默认为本地主机(localhost)。
user，连接数据库的用户名，默认为当前用户。
passwd，连接密码，没有默认值。
db，连接的数据库名，没有默认值。
conv，将文字映射到Python类型的字典。
cursorclass，cursor()使用的种类，默认值为MySQLdb.cursors.Cursor。
compress，启用协议压缩功能。
named_pipe，在windows中，与一个命名管道相连接。
init_command，一旦连接建立，就为数据库服务器指定一条语句来运行。
read_default_file，使用指定的MySQL配置文件。
read_default_group，读取的默认组。
unix_socket，在unix中，连接使用的套接字，默认使用TCP。
port，指定数据库服务器的连接端口，默认是3306。

connection对象支持的方法如下。

Cursor对象支持的方法如下。

用于执行查询和获取结果。

execute方法：执行SQL，将结果从数据库获取到客户端。

超时设置与重连尝试

调试代码时，我们将超时时间设置得较长。


self.connection._write_timeout = 10000

发现并没有生效。

使用try...except... 方法捕获失败后重新连接数据库。


try:
	self.cursor.execute(sql)
except:
	self.connection()
	self.cursor.execute(sql)

直接抛出异常，并没有执行except代码段。

打印self.connection ，输出如下：


<pymysql.connections.Connection object at 0x0000000003E2CCC0>

抛出异常重新connect是不行的，因为connections 仍存在未失效。

基于ping()的自动重连方案

找到一种方法可以解决问题，在每次连接之前，判断该链接是否有效，pymysql提供的接口是 Connection.ping()。

这个该方法的源码如下。


 def ping(self, reconnect=True):
    """Check if the server is alive"""
    if self._sock is None:
      if reconnect:
        self.connect()
        reconnect = False
      else:
        raise err.Error("Already closed")
    try:
      self._execute_command(COMMAND.COM_PING, "")
      return self._read_ok_packet()
    except Exception:
      if reconnect:
        self.connect()
        return self.ping(False)
      else:
        raise

在每次请求数据库前执行如下代码。


def reConnect(self):
	try:
		self.connection.ping()
	except:
		self.connection()

不过这样的方式虽然能解决问题，但是感觉相对较low，希望有更好的处理方法。

目前已实现的数据库查询这部分的代码。


import pymysql
class DBManager(object):

  def __init__(self,config):
    self.connection = pymysql.connect(**config) # config为数据库登录验证配置信息
    self.cursor = self.connection.cursor()

  def query(self, sql, params):
    try:
      with self.connection.cursor() as cursor:
        cursor.execute(sql, params)
        result = cursor.fetchall()
        self.connection.commit()
        return result
        # self.connection.close()
    except Exception as e:
      traceback.print_exc()

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持本站。