我只是资助自己实现一个基于计时器的"将事件列表处理成一堆"的版本，以便再次保护资源 - 我想知道是否有一种很好的通用pythonic方法。
您可能知道这一点：您正在处理重复发生的事件，如鼠标移动、文件系统更改等，并且您必须进行一些计算作为对这些事件的反应，但如果您可以在事件流中使用一点休息来处理它们，那就太好了一堆。也许是因为较旧的事件被较新的事件使无效(足以处理最旧的事件(，或者因为事件可以以某种方式挤在一起。
例如：鼠标移动(仅绘制最新位置(，编辑器中的"自动保存"或在文件系统上自动同步，或(在我的示例中(监视文件系统更改并重新编译某些内容。
通常，我会查找如何使用Timer，并思考如何避免额外的线程，并为一个非常简单的问题提出一些半成品但复杂的解决方案。 出现了很多问题：
- 如何避免并发处理(例如，如果我使用
threading.Timer并启动线程来执行工作( - 如何确保处理事件有时间限制(在连续传入事件不间断的情况下(
- 如果可能，如何避免线程
- 如何避免创建过于复杂的框架
- (你的名字(
我想要的是这样工作的东西：
timer = SomeContinuousTimer()
new_events = []
while True:
event = wait_for(inotify_adapter.event_gen(), timer.timeout())
if event == timer.TIMEOUT:
my_handler_func(new_events)
else:
new_events.append(event)
timer.restart(1500)
但是wait_for必须像select一样行事，为此我需要文件描述符，上面的代码已经比我实际预期的要多一点。
我真的很高兴拥有的东西会像这样使用：
bunch_handler = BunchHandler()
new_events = []
def read_events():
for event in inotify_adapter.event_gen():
new_events.append(event)
while True:
# will run `read_events` asynchronously until 1.5sec have passed since the
# last event
bunch_handler.read(read_fn=read_events, bunch_wait=1500)
handle_events(new_events)
这是我应该使用async/await的典型场景吗？对于无法选择async的情况，是否有框架？是否有针对此确切方案的异步框架？
这并不好，但它可以做我想要的，并且可以作为一个例子来显示我在说什么:)
import asyncio
import time
async def event_collector(*, listener_fn, bunch_wait=1.0, max_wait=2.0):
"""Wait for (but don't handle) events and wait for a maximum of @bunch_wait seconds after the
last event before returning. Force return after @max_wait seconds"""
max_time_task = asyncio.Task(asyncio.sleep(max_wait))
while True:
resetable = asyncio.Task(asyncio.sleep(bunch_wait))
done, _ = await asyncio.wait(
{listener_fn.__anext__(), resetable, max_time_task},
return_when=asyncio.FIRST_COMPLETED)
if resetable in done or max_time_task in done:
return
resetable.cancel()
async def generate_events(events):
"""Simulates bursts of events with side-effects"""
while True:
for i in range(5):
await asyncio.sleep(.01)
events.append(i)
print("*" * len(events))
yield
await asyncio.sleep(3.200)
def handle_events(events):
"""Simulates an event handler operating on a given structure"""
print("Handle %d events" % len(events))
events.clear()
async def main():
new_events = []
t = time.time()
while True:
await event_collector(listener_fn=generate_events(new_events), bunch_wait=1.1, max_wait=2.2)
now = time.time()
print("%.2f" % (now - t))
t = now
handle_events(new_events)
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
这种方法有一些缺点： * 您需要使用async异步侦听事件 * 无论是否看到任何事件，event_collector 都将在max_wait秒后返回(因此，如果没有事件发生，它就像超时( * 不是重置计时器，而是每次创建一个新计时器