前一段时间，我需要在R中的retry函数来处理慢速服务器响应。该函数将具有以下行为：（尝试操作（函数或方法），如果失败，请等待一点然后重试）x10
我想到了以下内容：
retry <- function(fun, max_trys = 10, init = 0){
suppressWarnings(tryCatch({
Sys.sleep(0.3);
if(init<max_trys) {fun}
}, error=function(e){retry(fun, max_trys, init = init+1)}))}
它运行良好。现在我在Python3中需要相同的内容，因此我尝试制作相同的代码：
import time
def retry_fun(fun, max_trys = 10, init=0):
try:
time.sleep(0.3)
if(init<max_trys):
fun
except:
retry_fun(fun, max_trys, init = init+1)
但是当我运行它时，它正在崩溃我的内核。由于我在Python中有点是初学者，因此我不确定是什么原因导致崩溃的原因，以及是否可以作为参数传递到另一个函数的函数。
你能帮我吗？
除了能够传递功能并通过添加 ()之外（python的语法调用呼叫）来使用它们，您否 not 需要使用递归;只需将其放在一个循环中：
import time
def retry(fun, max_tries=10):
for i in range(max_tries):
try:
time.sleep(0.3)
fun()
break
except Exception:
continue
except Exception应该更改以捕获该功能可能提出的有意义的例外。使用Exception（就像我在示例中所做的那样）通常是不好的练习，因为它会捕获您可能不会被捕获的大量例外。
除此之外，使用for-loop代替明确的第三计数器和递归（导致大量值的较长呼叫堆栈）更好。
我知道这是一个古老的问题。但是，我想添加我煮熟的解决方案。最好的方法是编写一个retry装饰器，该装饰器将在发生异常时会重试。此外，您还可以设置自定义指数延迟。Docstring解释了如何使用装饰器。在这里，你去：
import logging
import time
from functools import partial, wraps
def retry(func=None, exception=Exception, n_tries=5, delay=5, backoff=1, logger=False):
"""Retry decorator with exponential backoff.
Parameters
----------
func : typing.Callable, optional
Callable on which the decorator is applied, by default None
exception : Exception or tuple of Exceptions, optional
Exception(s) that invoke retry, by default Exception
n_tries : int, optional
Number of tries before giving up, by default 5
delay : int, optional
Initial delay between retries in seconds, by default 5
backoff : int, optional
Backoff multiplier e.g. value of 2 will double the delay, by default 1
logger : bool, optional
Option to log or print, by default False
Returns
-------
typing.Callable
Decorated callable that calls itself when exception(s) occur.
Examples
--------
>>> import random
>>> @retry(exception=Exception, n_tries=4)
... def test_random(text):
... x = random.random()
... if x < 0.5:
... raise Exception("Fail")
... else:
... print("Success: ", text)
>>> test_random("It works!")
"""
if func is None:
return partial(
retry,
exception=exception,
n_tries=n_tries,
delay=delay,
backoff=backoff,
logger=logger,
)
@wraps(func)
def wrapper(*args, **kwargs):
ntries, ndelay = n_tries, delay
while ntries > 1:
try:
return func(*args, **kwargs)
except exception as e:
msg = f"{str(e)}, Retrying in {ndelay} seconds..."
if logger:
logging.warning(msg)
else:
print(msg)
time.sleep(ndelay)
ntries -= 1
ndelay *= backoff
return func(*args, **kwargs)
return wrapper
有几个python软件包：
- 退缩
- penacity
向后的示例
import backoff
@backoff.on_exception(backoff.expo,
(MyPossibleException1,
MyPossibleException2))
def your_function(param1, param2):
# Do something
坚韧的示例
from tenacity import wait_exponential, retry, stop_after_attempt
@retry(wait=wait_exponential(multiplier=2, min=2, max=30), stop=stop_after_attempt(5))
def your_function(param1, param2):
# Do something
还有另一个可安装重试软件包：https://pypi.org/project/retry/
pip install retry
您可以做以下操作：
def retry_call(function_name, args=None, kwargs=None, retries=3):
pass_on_args = args if args else []
pass_on_kwargs = kwargs if kwargs else {}
for index in range(1, retries+1):
try:
return function_name(*pass_on_args, **pass_on_kwargs)
except Exception as error:
if index == retries:
LOGGER.error("Failed %s attempts at calling: %s", retries, function_name)
LOGGER.error(" args: %s", args)
LOGGER.error(" kwargs: %s", kwargs)
LOGGER.error(" error: %s", error)
raise
else:
LOGGER.warning("Failed %d attempt(s) at calling: %s", index, function_name)
LOGGER.warning("Retrying")
def get_variable
with open(FILE) as info:
for line in info:
if 'Mode' in line:
return mode
然后您将调用该功能：
mode = retry(get_variable)
print(mode)