随时间验证连续状态 [英] Verify continuous condition with time
问题描述
我想开发一个Python程序，从某个时刻开始，等待60秒再执行操作。该程序必须具有的另一个功能是，如果我更新了初始时间，则必须开始检查条件。我曾考虑过使用线程来做，但是我不知道如何停止线程并在新的开始时间重新启动它。
I would like to develop a python program that, starting from a moment of time, wait 60 seconds before performing an action. Another feature that must have the program, is that if I update the initial time, it must start to check the condition. I thought about doing it with threads, but I do not know how to stop the thread and get it started again with the new start time.
import thread
import time
# Define a function for the thread
def check_message (last, timer):
oldtime = time.time()
print oldtime
# check
while time.time() - oldtime <= 60:
print (time.time() - oldtime)
print "One minute"+ str(time.time())
return 1
# Create two threads as follows
try:
named_tuple = time.localtime() # get struct_time
time_string = time.strftime("%H:%M:%S", named_tuple)
thread.start_new_thread(check_message , (time_string, 60))
except:
print "Error: unable to start thread"
while 1:
pass
谢谢！
推荐答案
在这里循环检查时间可能没有必要，而且浪费时间，因为您c一个线程进入睡眠状态，如果时间到了，让内核将其唤醒。
线程库提供 线程。计时器 。您遇到的困难是，您不能中断这样的睡眠线程来调整执行指定功能的时间间隔。
Checking times in a loop is probably not necessary here and wasteful since you can put a thread to sleep and let the kernel wake it up if the time has come.
The threading library provides threading.Timer for such use cases. The difficulty in your case is, that you cannot interrupt such a sleeping thread to adjust the interval after which the specified function should be executed.
我正在使用自定义管理器在下面的示例中-class TimeLord 可以克服此限制。 TimeLord 可以通过取消当前计时器并将其替换为新计时器来重置计时器。
为此， TimeLord 包含一个包装的中间worker函数和一个 token属性，必须由正在运行的timer-instance弹出它们才能执行指定的目标功能。
I'm using a custom manager-class TimeLord in my example below, to overcome this limitation. TimeLord enables "resetting" the timer by canceling the current timer and replacing it with a new one.
For this purpose TimeLord contains a wrapping intermediate worker-function and a "token"-attribute, which must be popped by a running timer-instance to execute the specified target-function.
由于 dict.pop（）是原子操作，因此该设计保证了指定目标函数的唯一执行。只要当前计时器尚未启动其线程并弹出 _token ， timelord.reset（）就有效。这种方法不能完全防止尝试重置时新计时器线程的潜在无效启动，但由于目标函数只能执行一次，因此它是不重要的冗余。
This design guarantees unique execution of the specified target-function since dict.pop() is an atomic operation. timelord.reset() is effective as long the current timer has not started its thread and popped the _token. This approach can not totally prevent potentially ineffective starts of new timer-threads when trying to "reset", but it's an uncritical redundancy when it happens since the target function can only be executed once.
此代码在Python 2和3上运行。
This code runs with Python 2 and 3:
import time
from datetime import datetime
from threading import Timer, current_thread
def f(x):
print('{} {}: RUNNING TARGET FUNCTION'.format(
datetime.now(), current_thread().name)
)
time.sleep(x)
print('{} {}: EXITING'.format(datetime.now(), current_thread().name))
class TimeLord:
"""
Manager Class for threading.Timer instance. Allows "resetting" `interval`
as long execution of `function` has not started by canceling the old
and constructing a new timer instance.
"""
def worker(self, *args, **kwargs):
try:
self.__dict__.pop("_token") # dict.pop() is atomic
except KeyError:
pass
else:
self.func(*args, **kwargs)
def __init__(self, interval, function, args=None, kwargs=None):
self.func = function
self.args = args if args is not None else []
self.kwargs = kwargs if kwargs is not None else {}
self._token = True
self._init_timer(interval)
def _init_timer(self, interval):
self._timer = Timer(interval, self.worker, self.args, self.kwargs)
self._timer.daemon = True
def start(self):
self._timer.start()
print('{} {}: STARTED with `interval={}`'.format(
datetime.now(), self._timer.name, self._timer.interval)
)
def reset(self, interval):
"""Cancel latest timer and start a new one if `_token` is still there.
"""
print('{} {}: CANCELED'.format(datetime.now(), self._timer.name))
self._timer.cancel()
# reduces, but doesn't prevent, occurrences when a new timer
# gets created which eventually will not succeed in popping
# the `_token`. That's uncritical redundancy when it happens.
# Only one thread ever will be able to execute `self.func()`
if hasattr(self, "_token"):
self._init_timer(interval)
self.start()
def cancel(self):
self._timer.cancel()
def join(self, timeout=None):
self._timer.join(timeout=timeout)
def run_demo(initial_interval):
print("*** testing with initial interval {} ***".format(initial_interval))
tl = TimeLord(interval=initial_interval, function=f, args=(10,))
tl.start()
print('*** {} sleeping two seconds ***'.format(datetime.now()))
time.sleep(2)
tl.reset(interval=6)
tl.reset(interval=7)
tl.join()
print("-" * 70)
if __name__ == '__main__':
run_demo(initial_interval=5)
run_demo(initial_interval=2)
示例输出：
*** testing with initial interval 5 ***
2019-06-05 20:58:23.448404 Thread-1: STARTED with `interval=5`
*** 2019-06-05 20:58:23.448428 sleeping two seconds ***
2019-06-05 20:58:25.450483 Thread-1: CANCELED
2019-06-05 20:58:25.450899 Thread-2: STARTED with `interval=6`
2019-06-05 20:58:25.450955 Thread-2: CANCELED
2019-06-05 20:58:25.451496 Thread-3: STARTED with `interval=7`
2019-06-05 20:58:32.451592 Thread-3: RUNNING TARGET FUNCTION
2019-06-05 20:58:42.457527 Thread-3: EXITING
----------------------------------------------------------------------
*** testing with initial interval 2 ***
2019-06-05 20:58:42.457986 Thread-4: STARTED with `interval=2`
*** 2019-06-05 20:58:42.458033 sleeping two seconds ***
2019-06-05 20:58:44.458058 Thread-4: RUNNING TARGET FUNCTION
2019-06-05 20:58:44.459649 Thread-4: CANCELED
2019-06-05 20:58:44.459724 Thread-4: CANCELED
2019-06-05 20:58:54.466342 Thread-4: EXITING
----------------------------------------------------------------------
Process finished with exit code 0
注意，interval = 2，两秒钟后取消无效，因为计时器已经在执行目标功能。
Note, with interval=2 the cancelations after two seconds had no effect, since the timer was already executing the target-function.
这篇关于随时间验证连续状态的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
