如何以比较优雅的方式为python-while增加超时?

2020-11-07  本文已影响0人  汪少SZ
遇到的问题:

在日常编码过程,可能会遇到这种场景:需要较多次尝试获取数据,你写的程序不知道何时能拿到数据,于是通过for或者while去“询问”,如以下的伪代码示意:

def get_data_from_remote():
      return data

while True:
      data = get_data_from_remote()
      if data:
          break

这段代码,只要没有获取到数据,就会一直循环执行,主线程一直处于阻塞状态,也就是我们常说的“死循环”。那么我们怎么解决这一问题呢?

几种方案:

如果作为初级玩家,每次遇到这种场景,都写以下一段逻辑:

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

不错效果达到了,但不具有通用性,当然小编想不到更低级的写法了。所以思考一下,怎么更具通用性?
如果 while关键字能自带超时效果就好了,以小编浅薄的知识面,想到了两种方式:

然而这两种方式都要较深的功力,第2种方式可以做出来,但实现成本也不低,性能也不会太高。

# 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()

可以看出,调用的地方比之前就多一行代码:with ext_while_cond((flag1 and not flag2), timeout=2) as cond
self_while里面的代码,使用了反射、作用域、上下文管理装饰器等知识,可以细品一下,全展开说太累了。

这个工具类没有实现超时后,抛一个超时异常出来,读者可以自己实现。
感谢阅读。

上一篇下一篇

猜你喜欢

热点阅读