為了避免由于一些網絡或等其他不可控因素,而引起的功能性問題。比如在發送請求時,會因為網絡不穩定,往往會有請求超時的問題。
這種情況下,我們通常會在代碼中加入重試的代碼。重試的代碼本身不難實現,但如何寫得優雅、易用,是我們要考慮的問題。
這里要給大家介紹的是一個第三方庫 - Tenacity ,它實現了幾乎我們可以使用到的所有重試場景,比如:
- 在什么情況下才進行重試?
- 重試幾次呢?
- 重試多久后結束?
- 每次重試的間隔多長呢?
- 重試失敗后的回調?
在使用它之前 ,先要安裝它
$ pip install tenacity
1. 最基本的重試
無條件重試,重試之間無間隔
from tenacity import retry
@retry
def test_retry():
print("等待重試,重試無間隔執行...")
raise Exception
test_retry()
無條件重試,但是在重試之前要等待 2 秒
from tenacity import retry, wait_fixed
@retry(wait=wait_fixed(2))
def test_retry():
print("等待重試...")
raise Exception
test_retry()
2. 設置停止基本條件
只重試7 次
from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(7))
def test_retry():
print("等待重試...")
raise Exception
test_retry()
重試 10 秒后不再重試
from tenacity import retry, stop_after_delay
@retry(stop=stop_after_delay(10))
def test_retry():
print("等待重試...")
raise Exception
test_retry()
或者上面兩個條件滿足一個就結束重試
from tenacity import retry, stop_after_delay, stop_after_attempt
@retry(stop=(stop_after_delay(10) | stop_after_attempt(7)))
def test_retry():
print("等待重試...")
raise Exception
test_retry()
3. 設置何時進行重試
在出現特定錯誤/異常(比如請求超時)的情況下,再進行重試
from requests import exceptions
from tenacity import retry, retry_if_exception_type
@retry(retry=retry_if_exception_type(exceptions.Timeout))
def test_retry():
print("等待重試...")
raise exceptions.Timeout
test_retry()
在滿足自定義條件時,再進行重試。
如下示例,當 test_retry 函數返回值為 False 時,再進行重試
from tenacity import retry, stop_after_attempt, retry_if_result
def is_false(value):
return value is False
@retry(stop=stop_after_attempt(3),
retry=retry_if_result(is_false))
def test_retry():
return False
test_retry()
4. 重試后錯誤重新拋出
當出現異常后,tenacity 會進行重試,若重試后還是失敗,默認情況下,往上拋出的異常會變成 RetryError,而不是最根本的原因。
因此可以加一個參數(reraise=True),使得當重試失敗后,往外拋出的異常還是原來的那個。
from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(7), reraise=True)
def test_retry():
print("等待重試...")
raise Exception
test_retry()
5. 設置回調函數
當最后一次重試失敗后,可以執行一個回調函數
from tenacity import *
def return_last_value(retry_state):
print("執行回調函數")
return retry_state.outcome.result() # 表示返回原函數的返回值
def is_false(value):
return value is False
@retry(stop=stop_after_attempt(3),
retry_error_callback=return_last_value,
retry=retry_if_result(is_false))
def test_retry():
print("等待重試中...")
return False
print(test_retry())
輸出如下
等待重試中...
等待重試中...
等待重試中...
執行回調函數
False
聲明:本文內容及配圖由入駐作者撰寫或者入駐合作網站授權轉載。文章觀點僅代表作者本人,不代表電子發燒友網立場。文章及其配圖僅供工程師學習之用,如有內容侵權或者其他違規問題,請聯系本站處理。
舉報投訴
-
模塊
+關注
關注
7文章
2837瀏覽量
53283 -
參數
+關注
關注
11文章
1869瀏覽量
33945 -
代碼
+關注
關注
30文章
4968瀏覽量
73960
發布評論請先 登錄
相關推薦
熱點推薦
wifi (STA) 最大重試次數是多少?
您好,目前 STA 應用程序一直在嘗試連接到接入點。我需要限制重試次數。這個限制是可配置的嗎?萬一,怎么辦?“esp_wifi_connect()”是否不斷重試并生成事件“WIFI_EVENT_STA_DISCONNECTED”?萬一,我該如何設置最大
發表于 03-02 08:46
K32W041自SDK v2.6.8起不再重試是怎么回事?
我們將 K32W041 與我們自己的 802.15.4 協議一起使用,即不是 Zigbee 或 Thread,并且自 SDK 發布 v2.6.8 以來,我們觀察到傳輸重試的行為發生了變化
發表于 04-21 07:30
HBase客戶端實踐重試機制
現在,網易視頻云與大家分享HBase客戶端實踐重試機制。 在運維HBase的這段時間里,發現業務用戶一方面比較關注HBase本身服務的讀寫性能:吞吐量以及讀寫延遲,另一方面也會比較關注HBase
發表于 10-10 10:15
?0次下載
如何在RocketMQ中合理使用重試機制
RocketMQ 的重試機制包括三部分,分別是生產者重試,服務端內部數據復制遇到非預期問題時重試,消費者消費重試。
一個Spring注解輕松搞定循環重試功能!
重試等待策略,默認使用@Backoff,@Backoff的value默認為1000L,我們設置為2000L;multiplier(指定延遲倍數)默認為0,表示固定暫停1秒后進行重試,如果把multiplier設置為1.5,則第一次重試
Python 在什么情況下才進行重試
如何寫得優雅、易用,是我們要考慮的問題。 這里要給大家介紹的是一個第三方庫 - Tenacity (標題中的重試機制并并不準確,它不是 Python 的內置模塊,因此并不能稱之為機制),它實現了幾乎我們可以使用到的所有
Python中retrying庫的有參數重試
有參數重試 (1) stop_max_attempt_number 在retry中傳入stop_max_attempt_number參數后可以指定失敗重試的次數 @retry
Tenacity重試模塊實踐
評論