我有一个 python 脚本，它正在查询共享 Linux 主机上的 MySQL 服务器。由于某种原因，对 MySQL 的查询经常返回“服务器已消失”错误：
_mysql_exceptions.OperationalError: (2006, 'MySQL server has gone away')
如果您随后立即再次尝试查询，通常会成功。所以，我想知道 python 中是否有一种合理的方法来尝试执行查询，如果失败，则重试，最多尝试固定次数。也许我希望在完全放弃之前尝试 5 次。
这是我的代码：
conn = MySQLdb.connect(host, user, password, database)
cursor = conn.cursor()
try:
cursor.execute(query)
rows = cursor.fetchall()
for row in rows:
# do something with the data
except MySQLdb.Error, e:
print "MySQL Error %d: %s" % (e.args[0], e.args[1])
显然，我可以通过在 except 子句中再次尝试来做到这一点，但这非常丑陋，而且我有一种感觉，必须有一种不错的方法来实现这一目标。
怎么样：
conn = MySQLdb.connect(host, user, password, database)
cursor = conn.cursor()
attempts = 0
while attempts < 3:
try:
cursor.execute(query)
rows = cursor.fetchall()
for row in rows:
# do something with the data
break
except MySQLdb.Error, e:
attempts += 1
print "MySQL Error %d: %s" % (e.args[0], e.args[1])
根据 Dana 的答案，您可能想作为装饰器来执行此操作：
def retry(howmany):
def tryIt(func):
def f():
attempts = 0
while attempts < howmany:
try:
return func()
except:
attempts += 1
return f
return tryIt
然后...
@retry(5)
def the_db_func():
# [...]
decoratorimport decorator, time
def retry(howmany, *exception_types, **kwargs):
timeout = kwargs.get('timeout', 0.0) # seconds
@decorator.decorator
def tryIt(func, *fargs, **fkwargs):
for _ in xrange(howmany):
try: return func(*fargs, **fkwargs)
except exception_types or Exception:
if timeout is not None: time.sleep(timeout)
return tryIt
然后...
@retry(5, MySQLdb.Error, timeout=0.5)
def the_db_func():
# [...]
decorator$ easy_install decorator
更新：重试库有一个维护得更好的分支，称为tenacity，它支持更多功能，并且通常更灵活。
API 略有变化：
@retry(stop=stop_after_attempt(7))
def stop_after_7_attempts():
print("Stopping after 7 attempts")
@retry(wait=wait_fixed(2))
def wait_2_s():
print("Wait 2 second between retries")
@retry(wait=wait_exponential(multiplier=1, min=4, max=10))
def wait_exponential_1000():
print("Wait 2^x * 1000 milliseconds between each retry,")
print("up to 10 seconds, then 10 seconds afterwards")
是的，有重试库，它有一个装饰器，可以实现多种可以组合的重试逻辑：
一些例子：
@retry(stop_max_attempt_number=7)
def stop_after_7_attempts():
print("Stopping after 7 attempts")
@retry(wait_fixed=2000)
def wait_2_s():
print("Wait 2 second between retries")
@retry(wait_exponential_multiplier=1000, wait_exponential_max=10000)
def wait_exponential_1000():
print("Wait 2^x * 1000 milliseconds between each retry,")
print("up to 10 seconds, then 10 seconds afterwards")
conn = MySQLdb.connect(host, user, password, database)
cursor = conn.cursor()
for i in range(3):
try:
cursor.execute(query)
rows = cursor.fetchall()
for row in rows:
# do something with the data
break
except MySQLdb.Error, e:
print "MySQL Error %d: %s" % (e.args[0], e.args[1])
像 S.Lott 一样，我喜欢用一个标志来检查我们是否完成了：
conn = MySQLdb.connect(host, user, password, database)
cursor = conn.cursor()
success = False
attempts = 0
while attempts < 3 and not success:
try:
cursor.execute(query)
rows = cursor.fetchall()
for row in rows:
# do something with the data
success = True
except MySQLdb.Error, e:
print "MySQL Error %d: %s" % (e.args[0], e.args[1])
attempts += 1
我会像这样重构它：
def callee(cursor):
cursor.execute(query)
rows = cursor.fetchall()
for row in rows:
# do something with the data
def caller(attempt_count=3, wait_interval=20):
""":param wait_interval: In seconds."""
conn = MySQLdb.connect(host, user, password, database)
cursor = conn.cursor()
for attempt_number in range(attempt_count):
try:
callee(cursor)
except MySQLdb.Error, e:
logging.warn("MySQL Error %d: %s", e.args[0], e.args[1])
time.sleep(wait_interval)
else:
break
分解出
callee您可以使用带有
forelseconn = MySQLdb.connect(host, user, password, database)
cursor = conn.cursor()
for n in range(3):
try:
cursor.execute(query)
except MySQLdb.Error, e:
print "MySQL Error %d: %s" % (e.args[0], e.args[1])
else:
rows = cursor.fetchall()
for row in rows:
# do something with the data
break
else:
# All attempts failed, raise a real error or whatever
关键是查询成功就跳出循环。仅当循环在没有
elsebreakdef successful_transaction(transaction):
try:
transaction()
return True
except SQL...:
return False
succeeded = any(successful_transaction(transaction)
for transaction in repeat(transaction, 3))
1.定义：
def try_three_times(express):
att = 0
while att < 3:
try: return express()
except: att += 1
else: return u"FAILED"
2.用途：
try_three_times(lambda: do_some_function_or_express())
我用它来解析 html 上下文。
这是我的通用解决方案：
class TryTimes(object):
''' A context-managed coroutine that returns True until a number of tries have been reached. '''
def __init__(self, times):
''' times: Number of retries before failing. '''
self.times = times
self.count = 0
def __next__(self):
''' A generator expression that counts up to times. '''
while self.count < self.times:
self.count += 1
yield False
def __call__(self, *args, **kwargs):
''' This allows "o() calls for "o = TryTimes(3)". '''
return self.__next__().next()
def __enter__(self):
''' Context manager entry, bound to t in "with TryTimes(3) as t" '''
return self
def __exit__(self, exc_type, exc_val, exc_tb):
''' Context manager exit. '''
return False # don't suppress exception
这允许使用如下代码：
with TryTimes(3) as t:
while t():
print "Your code to try several times"
也可以：
t = TryTimes(3)
while t():
print "Your code to try several times"
我希望这可以通过以更直观的方式处理异常来改进。接受建议。