非轮询/非阻塞定时器？

Question

到目前为止，我找到的最好解决方案就是使用sleep()函数。我想在计时器到期事件发生时运行自己的回调函数。是否有任何基于事件驱动的方法可以实现？

from time import sleep

# Sleep for a minute
time.sleep(60)

- tarabyte

---

有几个事件/队列计时器库可供选择。我使用Twisted的反应堆，但那是因为我需要Twisted做其他事情。如果您的应用程序还有其他任务要完成，绝对不要让线程睡眠。 - Geoff Genz

这个问题对你来说已经解决了吗？如果是的话，你会接受其中一个发布的答案吗？如果不是，你有自己的答案要发布吗？ - Bach

Answer 1

使用Python中的threading模块可以轻松解决此问题：

import threading

timer = threading.Timer(60.0, callback)
timer.start()  # after 60 seconds, 'callback' will be called

## (in the meanwhile you can do other stuff...)

你也可以将参数和关键字参数传递给回调函数。请参见此处。

- Bach

---

我似乎无法重置计时器。我不想让它一直运行。我想在另一个事件上启动它，然后获取到期事件。当我尝试使用timer.cancel()时，我会得到以下错误信息：raise RuntimeError("threads can only be started once") RuntimeError: 线程只能启动一次。 - tarabyte

计时器无法重置/不需要重置，它只会运行一次。 - Bach

Answer 2

我认为这可能非常简单。请看这个例子。它甚至可以在Python控制台中运行！

from threading import Thread
from time import sleep

# Function to be called when the timer expires
def myFunction():
    print 'Did anyone call me?'

# Function with the timer
def myTimer(seconds):
    sleep(seconds)
    myFunction()

# Thread that will sleep in background and call your function
# when the timer expires.
myThread = Thread(target=myTimer, args=(4,))
myThread.start()

输入你想要的秒数，并继续使用控制台或运行主线程/程序。当计时器结束时，你会注意到函数被调用。

编辑

另一个很好的例子是仅根据某些变量或标志的值来调用函数。我希望这就是@tarabyte正在寻找的答案。

from threading import Thread
from time import sleep

myFlag = False

# Function to be called when the flag turns on
def myFunction():
    print 'Did anyone call me?'

def myTimer():
    global myFlag
    while True:
        if myFlag:
            myFunction()
            myFlag = False
        else:
            sleep(1)

# Thread that will sleep in background and call your function
# when the myFlag turns to be True
myThread = Thread(target=myTimer)
myThread.start()

# Then, you can do whatever you want and later change the value of myFlag.
# Take a look at the output inside ipython when the value of myFlag is changed.


In [35]: myFlag
Out[35]: False

In [36]: myFlag = True

In [37]: Did anyone call me?

- Javier

---

很好的例子，但我应该澄清一下，我对于重复过期不感兴趣，只是想要自己启动并在第一次运行后结束。我猜我可以通过将myTimer的内容包装在if (some_var_i_set_elsewhere)中来实现一个解决方案。 - tarabyte

嗨@tarabyte。请看一下我帖子的编辑版本。我认为这更或多或少是你想要做的事情。您可以修改myFlag的类型和可能的值，以使其适应您的需求。如果您同意我的意见，请接受答案并标记它。 - Javier

@Javier “它甚至在Python控制台中也能工作” - 所有的Python代码在控制台和脚本中都是一样的。只有一些代码仅在控制台中工作，而不在Python脚本中。例如，在控制台中输入somevar会打印出其值（如果之前已定义），但在Python脚本中则不会有任何反应。 - TheEagle

Answer 3

有时候，即使它会轮询时间，但简单的解决方案也是最好的。我以前用过这个方法，取得了巨大的成功——如果您的线程没有停止在它上面，它不会阻塞。

我认为，最简单的方法是通过检查时间来管理它，因为这比计算一个单独的线程解决方案更加简单和节约资源。

def event_minute_later(event):
    print(time.time()) # use for testing, comment out or delete for production
    return event + 60 < time.time()

使用方法：

>>> event = time.time()
>>> print(event)
1393962502.62

>>> event_minute_later(event)
1393962526.73
False
>>> event_minute_later(event)
1393962562.9
True

- Russia Must Remove Putin

Answer 4

自 Python 3.7 开始（旧版本已经到达生命周期终点），内置模块 asyncio 允许您异步添加 Python 的 sleep() 调用：

import asyncio

async def test():
    print("Hello ... but wait, there is more!")
    await asyncio.sleep(3)
    print("... in the async world!")

这里有一些证明它是非阻塞的（由RealPython提供）：

import asyncio
# Jupyter Notebook users need to allow
# nesting of the asyncio event loop
import nest_asyncio
nest_asyncio.apply()
import time

async def workload(text, duration):
    while duration > 0:
        # run sleep and yield control 
        # back to the event loop (for one cycle)
        await asyncio.sleep(1)
        print(f'{text} counter: sleeping {duration} seconds')
        duration -= 1

async def main():
    # send the workload() coroutine to the background,
    # to let it run concurrently with other tasks, 
    # switching between them at await points
    task_1 = asyncio.create_task(workload('First', 2))
    task_2 = asyncio.create_task(workload('Second', 4))
    task_3 = asyncio.create_task(workload('Third', 8))
    print(f"Started: {time.strftime('%X')}")
    # create await points for each 
    # of the concurrent tasks
    await task_1
    await task_2
    await task_3
    print(f"Ended: {time.strftime('%X')}")

if __name__ == '__main__':
    asyncio.run(main())

输出：

Started: 09:07:21
First counter: sleeping 2 seconds
Second counter: sleeping 4 seconds
Third counter: sleeping 8 seconds
First counter: sleeping 1 seconds
Second counter: sleeping 3 seconds
Third counter: sleeping 7 seconds
Second counter: sleeping 2 seconds
Third counter: sleeping 6 seconds
Second counter: sleeping 1 seconds
Third counter: sleeping 5 seconds
Third counter: sleeping 4 seconds
Third counter: sleeping 3 seconds
Third counter: sleeping 2 seconds
Third counter: sleeping 1 seconds
Ended: 09:07:29

- mirekphd