Python使用backoff如何更優(yōu)雅的實(shí)現(xiàn)輪詢

這篇文章將為大家詳細(xì)講解有關(guān)Python使用backoff如何更優(yōu)雅的實(shí)現(xiàn)輪詢，文章內(nèi)容質(zhì)量較高，因此小編分享給大家做個(gè)參考，希望大家閱讀完這篇文章后對(duì)相關(guān)知識(shí)有一定的了解。

我們經(jīng)常在開(kāi)發(fā)中會(huì)遇到這樣一種場(chǎng)景，即輪循操作。今天介紹一個(gè)Python庫(kù)，用于更方便的達(dá)到輪循的效果——backoff。

backoff 模塊簡(jiǎn)介及安裝

這個(gè)模塊主要提供了是一個(gè)裝飾器，用于裝飾函數(shù)，使得它在遇到某些條件時(shí)會(huì)重試（即反復(fù)執(zhí)行被裝飾的函數(shù)）。通常適用于我們?cè)讷@取一些不可靠資源，比如會(huì)間歇性故障的資源等。

此外，裝飾器支持正常的同步方法，也支持異步asyncio代碼。

backoff 模塊的安裝也很簡(jiǎn)單，通過(guò) pip 即可安裝完成：

pip install backoff

backoff 用法及簡(jiǎn)單源碼分析

backoff 提供兩個(gè)主要的裝飾器，通過(guò) backoff. 調(diào)用，通過(guò)提示我們可以看到這兩個(gè)裝飾器，分別是：

backoff.on_predicatebackoff.on_exception

通過(guò) github 查看 backoff 的源碼，源碼目錄 backoff/_decorator.py，定義如下：

def on_predicate(wait_gen, predicate=operator.not_, max_tries=None, max_time=None, jitter=full_jitter, on_success=None, on_backoff=None, on_giveup=None, logger='backoff',
 **wait_gen_kwargs):
 # 省略具體代碼
 # 每個(gè)參數(shù)的定義在源碼中都給出了明確的解釋
 pass
def on_exception(wait_gen,
 exception, max_tries=None, max_time=None, jitter=full_jitter, giveup=lambda e: False, on_success=None, on_backoff=None, on_giveup=None, logger='backoff',
 **wait_gen_kwargs):
 # 省略具體代碼
 # 每個(gè)參數(shù)的定義在源碼中都給出了明確的解釋
 pass

可以看到，定義了很多的參數(shù)，這些參數(shù)在源碼中都給出了比較詳細(xì)的解釋，這里做簡(jiǎn)單的介紹：

首先，wait_gen：表示每次循環(huán)等待的時(shí)長(zhǎng)，以秒為單位。它的類型是一個(gè)生成器，在 backoff 中內(nèi)置了三個(gè)生成器。我們查看下源碼，目錄為 backoff/_wait_gen.py。我們?nèi)∑渲幸粋€(gè)的詳細(xì)實(shí)現(xiàn)來(lái)看下：

# 省略實(shí)現(xiàn)代碼# base * factor * ndef expo(base=2, factor=1, max_value=None):
 """Generator for exponential decay.
 Args:
 base: The mathematical base of the exponentiation operation
 factor: Factor to multiply the exponentation by.
 max_value: The maximum value to yield. Once the value in the
 true exponential sequence exceeds this, the value
 of max_value will forever after be yielded.
 """
 n = 0
 while True:
 a = factor * base ** n if max_value is None or a < max_value: yield a
 n += 1
 else: yield max_value# 通過(guò)斐波那契數(shù)列控制def fibo(max_value=None):
 pass# 常量數(shù)值def constant(interval=1):
 pass

從源碼不難看出，通過(guò)一些策略，每次 yield 返回不同的數(shù)值，這些數(shù)值就是重試等待秒數(shù)。當(dāng)然因?yàn)檫@個(gè)參數(shù)類型是生成器，顯然我們也是可以自定義的。同時(shí)我們會(huì)發(fā)現(xiàn)每個(gè) wait_gen 都是參數(shù)控制的，所以我們理應(yīng)是可以修改這個(gè)參數(shù)的初始值的。

顯然，wait_gen_kwargs就是用來(lái)傳遞這些參數(shù)的，它是通過(guò)可變關(guān)鍵字參數(shù)控制的，可以直接用 key=value 的形式進(jìn)行傳參，簡(jiǎn)單示例如下：

@backoff.on_predicate(backoff.constant, interval=5)def main3():
 print("time is {} retry...".format(time.time()))

predict 與 exception。這兩個(gè)相對(duì)比較簡(jiǎn)單，predict 接受一個(gè)函數(shù)，當(dāng)這個(gè)函數(shù)返回 True 時(shí)會(huì)進(jìn)行重試，否則停止，同時(shí)這個(gè)函數(shù)接受一個(gè)參數(shù)，這個(gè)參數(shù)的值是被裝飾函數(shù)的返回值。這個(gè)參數(shù)的默認(rèn)值是：operator._not。這個(gè)函數(shù)的源碼如下：

def not_(a): "Same as not a."
 return not a

所以默認(rèn)返回的是 not 被裝飾函數(shù)的返回值。如果當(dāng)被裝飾函數(shù)并沒(méi)有返回值時(shí)，返回 True，會(huì)進(jìn)行重試。

示例代碼如下：

import backoffimport time@backoff.on_predicate(backoff.fibo)def test2():
 print("time is {}, retry...".format(time.time()))if __name__ == "__main__":
 test2()# 等價(jià)于：# 必須接受一個(gè)參數(shù)，這個(gè)參數(shù)的值是被裝飾函數(shù)的返回值def condition(r):
 return True
 @backoff.on_predicate(backoff.fibo, condition)def test2():
 print("time is {}, retry...".format(time.time()))if __name__ == "__main__":
 test2()

執(zhí)行結(jié)果如下：

$ python3 backoff_test.pytime is 1571801845.834578, retry...time is 1571801846.121314, retry...time is 1571801846.229812, retry...time is 1571801846.533237, retry...time is 1571801849.460303, retry...time is 1571801850.8974788, retry...time is 1571801856.498335, retry...time is 1571801861.56931, retry...time is 1571801872.701226, retry...time is 1571801879.198495, retry...
...

需要注意幾點(diǎn)：

• 如果自定義這個(gè)參數(shù)對(duì)應(yīng)的函數(shù)，這個(gè)函數(shù)是需要接受一個(gè)參數(shù)的，這個(gè)參數(shù)的值是被裝飾函數(shù)的返回值。我們可以通過(guò)控制這個(gè)返回值來(lái)做一些條件判斷，當(dāng)達(dá)到某些特殊條件時(shí)重試結(jié)束。

• 示例中 wait_gen 用的是 backoff.fibo，注意觀察輸出的時(shí)間單隔，這里的時(shí)間間隔好像并不像我們想象中按 fibo 返回的時(shí)間間隔數(shù)，實(shí)際上如果想達(dá)到這個(gè)效果，我們需要將 jitter 參數(shù)設(shè)置為 None，后面介紹 jitter 參數(shù)時(shí)再做說(shuō)明。

而 exception 則是接受異常類型的實(shí)例，可以是單個(gè)異常，也可以是元組形式的多個(gè)異常。簡(jiǎn)單示例如下：

import timeimport randomimport backofffrom collections import dequeclass MyException(Exception):
 def __init__(self, message, status):
 super().__init__(message, status)
 self.message = message
 self.status = statusclass MyException2(Exception):
 pass@backoff.on_exception(backoff.expo, (MyException, MyException2))def main():
 random_num = random.randint(0, 9)
 print("retry...and random num is {}".format(random_num)) if random_num % 2 == 0: raise MyException("my exception", int("1000" + str(random_num))) raise MyException2()

max_tries 與 max_time 也比較簡(jiǎn)單，分別代表最大重試次數(shù)與最長(zhǎng)重試時(shí)間。這里就不做演示了。

@backoff.on_exception 中的 giveup，它接受一個(gè)異常實(shí)例，通過(guò)對(duì)這個(gè)實(shí)例做一些條件判斷，達(dá)到判斷是否需要繼續(xù)循環(huán)的目的。如果返回 True，則結(jié)束，反之繼續(xù)。默認(rèn)值一直是返回 False，即會(huì)一直循環(huán)。示例如下：

import randomimport backoffclass MyException(Exception):
 def __init__(self, message, status):
 super().__init__(message, status)
 self.message = message
 self.status = statusdef exception_status(e):
 print('exception status code is {}'.format(e.status)) return e.status % 2 == 0
 @backoff.on_exception(backoff.expo, MyException, giveup=exception_status)def main():
 random_num = random.randint(0, 9)
 print("retry...and random num is {}".format(random_num)) raise MyException("my exception", int("1000" + str(random_num)))if __name__ == "__main__":
 main()

運(yùn)行結(jié)果：

retry...and random num is 5exception status code is 10005retry...and random num is 0exception status code is 10000# 會(huì)再走一遍 raise 的代碼，所以異常仍然會(huì)拋出來(lái)
Traceback (most recent call last):
 File "backoff_test.py", line 36, in <module>
 main()
 File "/Users/ruoru/code/python/exercise/.venv/lib/python3.7/site-packages/backoff/_sync.py", line 94, in retry
 ret = target(*args, **kwargs)
 File "backoff_test.py", line 32, in main
 raise MyException("my exception", int("1000" + str(random_num)))
__main__.MyException: ('my exception', 10000)

需要注意兩點(diǎn)：

• 這個(gè)參數(shù)接受的函數(shù)仍然只有一個(gè)參數(shù)，這個(gè)參數(shù)的值是一個(gè)異常實(shí)例對(duì)象

• 從結(jié)果我們可以看出，當(dāng)拋出異常時(shí)，會(huì)先進(jìn)入 giveup 接受的函數(shù)，如果函數(shù)判斷需要 giveup 時(shí)，當(dāng)前的異常仍然會(huì)拋出。所以有需要，代碼仍然需要做異常邏輯處理。

on_success、on_backoff 與 on_giveup 這三個(gè)是一類的參數(shù)，用于做事件處理：

• on_sucess 事件會(huì)比較難理解一點(diǎn)，它表示的是被裝飾函數(shù)成功結(jié)束輪循則會(huì)退出，對(duì)于 on_exception 來(lái)說(shuō)即當(dāng)被裝飾函數(shù)沒(méi)有發(fā)生異常時(shí)則會(huì)調(diào)用 on_success。而對(duì)于 on_predicate 來(lái)說(shuō)即是通過(guò) predicate 關(guān)鍵字返回為 False 結(jié)束循環(huán)則會(huì)調(diào)用。

• on_backoff 即當(dāng)程序產(chǎn)生循環(huán)時(shí)會(huì)調(diào)用

• on_giveup 當(dāng)程序是達(dá)到當(dāng)前可嘗試最大次數(shù)后，會(huì)調(diào)用。對(duì)于 on_predicate 如果是通過(guò) max_tries 或者 max_time 會(huì)調(diào)用，而對(duì)于 on_exception ，對(duì)于 exception 參數(shù)返回 True 時(shí)也會(huì)調(diào)用 on_giveup

總結(jié)來(lái)說(shuō)，max_tries 和 max_time 這種直接控制結(jié)束的，調(diào)用的是 on_giveup，而 exception 參數(shù)也是通過(guò)返回 True 則程序就結(jié)束，它是用來(lái)控制程序結(jié)束的，所以也會(huì)調(diào)用 on_giveup。而 predicate 參數(shù)返回 True 則程序繼續(xù)，它是用來(lái)控制程序是否繼續(xù)徨的，所以當(dāng)它結(jié)束時(shí)，調(diào)用的是 on_success。

實(shí)驗(yàn)代碼如下：

'''
@Author: ruoru
@Date: 2019-10-22 15:30:32
@LastEditors: ruoru
@LastEditTime: 2019-10-23 14:37:13
@Description: backoff
'''import timeimport randomimport backoffclass MyException(Exception):
 def __init__(self, status, message):
 super().__init__(status, message)
 self.status = status
 self.message = messagedef backoff_hdlr(details):
 print("Backing off {wait:0.1f} seconds afters {tries} tries "
 "calling function {target} with args {args} and kwargs "
 "{kwargs}".format(**details))def success_hdlr(details):
 print("Success offafters {tries} tries "
 "calling function {target} with args {args} and kwargs "
 "{kwargs}".format(**details))def giveup_hdlr(details):
 print("Giveup off {tries} tries "
 "calling function {target} with args {args} and kwargs "
 "{kwargs}".format(**details))@backoff.on_predicate(
 backoff.constant, # 當(dāng) random num 不等 10009 則繼續(xù)
 # 當(dāng) random_num 等于 10009 后，會(huì)調(diào)用 on_success
 lambda x: x != 10009,
 on_success=success_hdlr,
 on_backoff=backoff_hdlr,
 on_giveup=giveup_hdlr,
 max_time=2)def main():
 num = random.randint(10000, 10010)
 print("time is {}, num is {}, retry...".format(time.time(), num)) return num@backoff.on_exception(
 backoff.constant,
 MyException, # 當(dāng) Exception 實(shí)例對(duì)象的 status 為 10009 成立時(shí)退出
 # 當(dāng)條件成立時(shí)，調(diào)用的是 on_giveup
 giveup=lambda e: e.status == 10009,
 on_success=success_hdlr,
 on_backoff=backoff_hdlr,
 on_giveup=giveup_hdlr,
 )def main2():
 num = random.randint(10000, 10010)
 print("time is {}, num is {}, retry...".format(time.time(), num)) # 如果是通過(guò)這個(gè)條件成立退出，調(diào)用的是 on_success
 if num == 10010: return
 raise MyException(num, "hhh")if __name__ == "__main__": #main()
 main2()

logger 參數(shù)，很顯然就是用來(lái)控制日志輸出的，這里不做詳細(xì)介紹。copy 官方文檔的一個(gè)示例：

my_logger = logging.getLogger('my_logger')
my_handler = logging.StreamHandler()
my_logger.add_handler(my_handler)
my_logger.setLevel(logging.ERROR)
@backoff.on_exception(backoff.expo,
 requests.exception.RequestException,
 logger=my_logger)# ...

最后一個(gè)參數(shù)，jitter，開(kāi)始也不是很明白這個(gè)參數(shù)的作用，文檔的解釋如下：

jitter: A function of the value yielded by wait_gen returning the actual time to wait. This distributes wait times stochastically in order to avoid timing collisions across concurrent clients. Wait times are jittered by default using the full_jitter function. Jittering may be disabled altogether by passing jitter=None.

有點(diǎn)暈，于是去看了下源碼，明白了用法，截取關(guān)鍵源碼如下：

# backoff/_decorator.pydef on_predicate(wait_gen,
 predicate=operator.not_,
 max_tries=None,
 max_time=None,
 jitter=full_jitter,
 on_success=None,
 on_backoff=None,
 on_giveup=None,
 logger='backoff',
 **wait_gen_kwargs):
 pass # 省略
 # 因?yàn)闆](méi)有用到異步，所以會(huì)進(jìn)到這里
 if retry is None:
 retry = _sync.retry_predicate# backoff/_sync# 分析可以看到有一句獲取下次 wait 時(shí)長(zhǎng)seconds = _next_wait(wait, jitter, elapsed, max_time_)# backoff/_commondef _next_wait(wait, jitter, elapsed, max_time): value = next(wait) try: if jitter is not None: seconds = jitter(value) else: seconds = value
 except TypeError:
 warnings.warn( "Nullary jitter function signature is deprecated. Use "
 "unary signature accepting a wait value in seconds and "
 "returning a jittered version of it.",
 DeprecationWarning,
 stacklevel=2,
 ) seconds = value + jitter() # don't sleep longer than remaining alloted max_time
 if max_time is not None: seconds = min(seconds, max_time - elapsed) return seconds

看前面幾行代碼應(yīng)該就會(huì)比較清晰了，如果 jitter 為 None，則會(huì)使用第一個(gè)參數(shù)返回的 value 值，而如果使用了，則會(huì)在這個(gè) value 值上再做一次算法，默認(rèn)為 full_jitter(value)。backoff/_jitter.py 提供了兩個(gè)算法，代碼不長(zhǎng)，貼上來(lái)看看：

import randomdef random_jitter(value):
 """Jitter the value a random number of milliseconds.
 This adds up to 1 second of additional time to the original value.
 Prior to backoff version 1.2 this was the default jitter behavior.
 Args:
 value: The unadulterated backoff value.
 """
 return value + random.random()def full_jitter(value):
 """Jitter the value across the full range (0 to value).
 This corresponds to the "Full Jitter" algorithm specified in the
 AWS blog's post on the performance of various jitter algorithms.
 (http://www.awsarchitectureblog.com/2015/03/backoff.html)
 Args:
 value: The unadulterated backoff value.
 """
 return random.uniform(0, value)

關(guān)于Python使用backoff如何更優(yōu)雅的實(shí)現(xiàn)輪詢就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，可以學(xué)到更多知識(shí)。如果覺(jué)得文章不錯(cuò)，可以把它分享出去讓更多的人看到。