遇到的問題:
在日常編碼過程,可能會(huì)遇到這種場(chǎng)景:需要較多次嘗試獲取數(shù)據(jù)隆敢,你寫的程序不知道何時(shí)能拿到數(shù)據(jù)发皿,于是通過for或者while去“詢問”,如以下的偽代碼示意:
def get_data_from_remote():
return data
while True:
data = get_data_from_remote()
if data:
break
這段代碼拂蝎,只要沒有獲取到數(shù)據(jù)穴墅,就會(huì)一直循環(huán)執(zhí)行,主線程一直處于阻塞狀態(tài)匣屡,也就是我們常說的“死循環(huán)”封救。那么我們?cè)趺唇鉀Q這一問題呢?
幾種方案:
如果作為初級(jí)玩家捣作,每次遇到這種場(chǎng)景誉结,都寫以下一段邏輯:
start_time = datetime.datetime.now()
timeout = 5
while True:
data = get_data_from_remote()
if data:
break
if (datetime.datetime.now() -start_time).seconds> timeout:
raise TimeoutException
不錯(cuò)效果達(dá)到了,但不具有通用性券躁,當(dāng)然小編想不到更低級(jí)的寫法了惩坑。所以思考一下,怎么更具通用性也拜?
如果 while關(guān)鍵字能自帶超時(shí)效果就好了撒桨,以小編淺薄的知識(shí)面搬俊,想到了兩種方式:
-
python解釋器
修改python解釋器,把while這個(gè)關(guān)鍵字功能豐富一下(雖然C語言是最好的語言,但我的這塊知識(shí)都全部還給大學(xué)老師了) -
ast
通過ast(抽象語法樹)這個(gè)底層庫履肃,通過代碼替換的方式,這種方式小編幾年前用過抛虫,比如把time.sleep(1000) 悄悄地改成sleep(1)懂缕,再例如python著名測(cè)試框架pytest就是用這種方式把a(bǔ)ssert關(guān)鍵字給重寫了。
然而這兩種方式都要較深的功力键俱,第2種方式可以做出來兰绣,但實(shí)現(xiàn)成本也不低,性能也不會(huì)太高编振。
-
方式3
于是思考有沒有更簡(jiǎn)單的方式缀辩,想到可以“自動(dòng)”修改while的條件, 請(qǐng)看下面的代碼:
# self_while.py
import datetime
import sys
import traceback
from contextlib import contextmanager
class Condition(object):
def __init__(self, cond_str, timeout):
self.start = datetime.datetime.now()
self.cond_str = cond_str
self.timeout = timeout
def __call__(self, *args, **kwargs):
frame = sys._getframe(1)
c = eval(self.cond_str, frame.f_locals)
return (datetime.datetime.now() - self.start).seconds < self.timeout and c
@contextmanager
def ext_while_cond(whl_cond, timeout=5):
cond_str = get_cond_str()
cond = Condition(cond_str, timeout)
yield cond
whl_line_ = 'with ' + ext_while_cond.__name__ + '('
def get_cond_str():
x = traceback.extract_stack()
for i in range(len(x)):
code_line = x[i].line
if whl_line_ in code_line:
break
if whl_line_ not in code_line:
raise Exception('Cannot Find While Statement')
else:
l_idx, r_idx = code_line.find("("), code_line.rfind(")")
return code_line[l_idx + 1, r_idx]
使用這些代碼的例子:
import time
from self_while import ext_while_cond
def test1():
flag = True
with ext_while_cond(flag is True) as cond:
while cond():
# flag = False
time.sleep(1)
print("A")
def test2():
flag1 = True
flag2 = False
with ext_while_cond((flag1 and not flag2), timeout=2) as cond:
while cond():
time.sleep(1)
print("A")
if __name__ == '__main__':
test1()
可以看出,調(diào)用的地方比之前就多一行代碼:with ext_while_cond((flag1 and not flag2), timeout=2) as cond踪央。
self_while里面的代碼臀玄,使用了反射、作用域杯瞻、上下文管理裝飾器等知識(shí)镐牺,可以細(xì)品一下,全展開說太累了魁莉。
這個(gè)工具類沒有實(shí)現(xiàn)超時(shí)后睬涧,拋一個(gè)超時(shí)異常出來募胃,讀者可以自己實(shí)現(xiàn)。
感謝閱讀畦浓。