我有这个装饰器：
def timed_out(timeout):
def decorate(f):
if not hasattr(signal, "SIGALRM"):
return f
def handler(signum, frame):
raise TimedOutExc()
@functools.wraps(f)
def new_f(*args, **kwargs):
old = signal.signal(signal.SIGALRM, handler)
signal.alarm(timeout)
try:
result = f(*args, **kwargs)
finally:
signal.signal(signal.SIGALRM, old)
signal.alarm(0)
return result
new_f.func_name = f.func_name
return new_f
return decorate
该代码仅在 Linux 上执行任何操作，但与 Windows 上一样，没有
SIGALRM这不是很漂亮，但我必须以跨平台的方式做类似的事情，所以我想出了使用单独的线程。基于信号的系统无法在所有平台上可靠地工作。
此类的使用可以包装在装饰器中，或者制作成
withYMMV.
#!/usr/bin/env python2.7
import time, threading
class Ticker(threading.Thread):
"""A very simple thread that merely blocks for :attr:`interval` and sets a
:class:`threading.Event` when the :attr:`interval` has elapsed. It then waits
for the caller to unset this event before looping again.
Example use::
t = Ticker(1.0) # make a ticker
t.start() # start the ticker in a new thread
try:
while t.evt.wait(): # hang out til the time has elapsed
t.evt.clear() # tell the ticker to loop again
print time.time(), "FIRING!"
except:
t.stop() # tell the thread to stop
t.join() # wait til the thread actually dies
"""
# SIGALRM based timing proved to be unreliable on various python installs,
# so we use a simple thread that blocks on sleep and sets a threading.Event
# when the timer expires, it does this forever.
def __init__(self, interval):
super(Ticker, self).__init__()
self.interval = interval
self.evt = threading.Event()
self.evt.clear()
self.should_run = threading.Event()
self.should_run.set()
def stop(self):
"""Stop the this thread. You probably want to call :meth:`join` immediately
afterwards
"""
self.should_run.clear()
def consume(self):
was_set = self.evt.is_set()
if was_set:
self.evt.clear()
return was_set
def run(self):
"""The internal main method of this thread. Block for :attr:`interval`
seconds before setting :attr:`Ticker.evt`
.. warning::
Do not call this directly! Instead call :meth:`start`.
"""
while self.should_run.is_set():
time.sleep(self.interval)
self.evt.set()
...并且它可能不再适用于最新版本的 Cygwin...
我发现这个超时装饰器代码也非常方便。 （我最初在这个问题的答案中找到它：如何限制函数调用的执行时间？）
为了使其在 Windows 上运行，我使用随 Cygwin 一起安装的 Python。
我运行 setup-x86_64.exe，然后从 Python 文件夹中选择
python3python要将 python3 重命名为 python2，我定义了别名
alias python=python3
从 Cygwin 命令提示符。由于我不经常使用此功能，因此我可能不会将其放入 .bashrc 或其他任何内容中。
有一种方法可以在 Windows 上创建 SIGALRM 类型事件。它利用多媒体定时器来触发事件。这是一个通用计时器，可以触发单个事件 (
periodic=Falsen_reps >= 1.stop()可以指定此计时器来生成间隔非常精确的事件，这些事件不依赖于 Python 运行时。此处，时间间隔以秒为单位指定（
intervalresolutionimport ctypes as ct
from ctypes.wintypes import UINT, DWORD
from typing import Callable, Optional
class mmPeriodicTimer:
"""Periodic timer based upon the Windows Multimedia library
This produces a timed callback with high precision ( ~ < 1ms difference)
executed outside of a threaded structure (based on Windows runtime ).
The advantage of this is that the Python GIL limitation is avoided,
and an aligned "lag" time between all Python processes is not generated;
The disadvantage is that since this is thread independent, the callback
procedure must complete its tasks before the next callback is executed
otherwise collisions may occur
This is based on the example:
https://stackoverflow.com/questions/10717589/how-to-implement-high-speed-consistent-sampling
"""
timeproc = ct.WINFUNCTYPE(None, ct.c_uint, ct.c_uint, DWORD, DWORD, DWORD)
timeSetEvent = ct.windll.winmm.timeSetEvent
timeKillEvent = ct.windll.winmm.timeKillEvent
def Tick(self):
self.i_reps += 1
self.tickFunc()
if not self.periodic:
self.stop()
if self.n_reps is not None and self.i_reps >= self.n_reps:
self.stop()
def CallBack(self, uID, uMsg, dwUser, dw1, dw2):
if self.running:
self.Tick()
def __init__(
self,
interval: int,
tickfunc: Callable,
resolution: Optional[int] = 0,
stopFunc: Optional[Callable] = None,
periodic: Optional[bool] = True,
n_reps: Optional[int] = None,
):
self.interval = UINT(int(interval * 1000))
self.resolution = UINT(int(resolution * 1000))
self.tickFunc = tickfunc
self.stopFunc = stopFunc
self.periodic = periodic
self.n_reps = n_reps
if not self.periodic and self.n_reps is not None:
raise ValueError("n_reps must be None if periodic is False")
self.i_reps = 0
self.id = None
self.running = False
self.calbckfn = self.timeproc(self.CallBack)
def start(self, instant: bool = False):
if not self.running:
self.running = True
if instant:
self.Tick()
self.id = self.timeSetEvent(
self.interval,
self.resolution,
self.calbckfn,
ct.c_ulong(0),
ct.c_uint(self.periodic),
)
def stop(self):
if self.running:
self.timeKillEvent(self.id)
self.running = False
if self.stopFunc:
self.stopFunc()
单个事件示例：
import time
from datetime import datetime
def my_func():
print("current time=", datetime.now())
t = mmPeriodicTimer(interval=1, tickfunc=my_func, periodic=False)
t.start()
输出：
current time= 2024-03-24 22:42:16.475416多个事件示例（然后停止）：
t = mmPeriodicTimer(interval=1, tickfunc=my_func, n_reps=5)
t.start()
输出：
current time= 2024-03-24 22:42:11.474459
current time= 2024-03-24 22:42:12.475201
current time= 2024-03-24 22:42:13.474462
current time= 2024-03-24 22:42:14.475113
current time= 2024-03-24 22:42:15.474802
给定时间后手动终止示例：
t = mmPeriodicTimer(interval=1, tickfunc=my_func, periodic=True)
t.start()
time.sleep(5)
t.stop()
输出：
current time= 2024-03-24 22:46:32.864423
current time= 2024-03-24 22:46:33.865319
current time= 2024-03-24 22:46:34.864554
current time= 2024-03-24 22:46:35.864762
current time= 2024-03-24 22:46:36.864778