Python asyncio强制超时
使用asyncio，可以在超时的情况下执行协程，这样在超时之后它就会被取消：
@asyncio.coroutine
def coro():
yield from asyncio.sleep(10)
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait_for(coro(), 5))上面的例子按照预期工作(5秒后超时)。
然而，当协程没有使用asyncio.sleep() (或其他异步协程)时，它似乎不会超时。示例：
@asyncio.coroutine
def coro():
import time
time.sleep(10)
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait_for(coro(), 1))这需要10秒以上的时间才能运行，因为time.sleep(10)未取消。在这种情况下，是否可以强制取消协程？
如果应该使用asyncio来解决这个问题，我该怎么做呢？
回答 3
Stack Overflow用户
发布于 2015-02-19 23:22:11
不，您不能中断协程，除非它将控制权交还给事件循环，这意味着它需要在yield from调用中。asyncio是单线程的，因此当您在第二个示例中阻塞time.sleep(10)调用时，事件循环将无法运行。这意味着当您使用wait_for设置的超时时间到期时，事件循环将无法对其执行操作。在coro退出之前，事件循环不会有机会再次运行，到那时就太晚了。
这就是为什么在一般情况下，您应该始终避免任何非异步的阻塞调用；任何时候，当调用阻塞而不向事件循环让步时，程序中的其他任何东西都不能执行，这可能不是您想要的。如果你真的需要做一个长的阻塞操作，你应该尝试使用BaseEventLoop.run_in_executor在线程或进程池中运行它，这样可以避免阻塞事件循环：
import asyncio
import time
from concurrent.futures import ProcessPoolExecutor
@asyncio.coroutine
def coro(loop):
ex = ProcessPoolExecutor(2)
yield from loop.run_in_executor(ex, time.sleep, 10) # This can be interrupted.
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait_for(coro(loop), 1))Stack Overflow用户
发布于 2016-10-19 02:26:18
Thx @dano感谢您的回答。如果运行coroutine不是一个很难的要求，这里有一个修改过的、更紧凑的版本
import asyncio, time
timeout = 0.5
loop = asyncio.get_event_loop()
future = asyncio.wait_for(loop.run_in_executor(None, time.sleep, 2), timeout)
try:
loop.run_until_complete(future)
print('Thx for letting me sleep')
except asyncio.exceptions.TimeoutError:
print('I need more sleep !')出于好奇，我的Python 3.8.2中的一些调试显示，将None作为执行器传递会导致创建一个_default_executor，如下所示：
self._default_executor = concurrent.futures.ThreadPoolExecutor()Stack Overflow用户
发布于 2018-01-31 23:50:40
我所见过的超时处理示例非常简单。考虑到现实，我的应用程序要复杂一些。顺序为：
- 当客户端连接到服务器时，让服务器创建另一个到内部服务器的连接
- 当内部服务器连接正常时，等待客户端发送数据。基于此数据，我们可以向内部服务器发出查询。
- 当有数据要发送到内部服务器时，将其发送。由于内部服务器有时响应不够快，请将此请求包装在超时中。
- 如果操作超时，请折叠所有连接以向客户端发出有关error
的信号
为了实现上述所有功能，同时保持事件循环运行，生成的代码包含以下代码：
def connection_made(self, transport):
self.client_lock_coro = self.client_lock.acquire()
asyncio.ensure_future(self.client_lock_coro).add_done_callback(self._got_client_lock)
def _got_client_lock(self, task):
task.result() # True at this point, but call there will trigger any exceptions
coro = self.loop.create_connection(lambda: ClientProtocol(self),
self.connect_info[0], self.connect_info[1])
asyncio.ensure_future(asyncio.wait_for(coro,
self.client_connect_timeout
)).add_done_callback(self.connected_server)
def connected_server(self, task):
transport, client_object = task.result()
self.client_transport = transport
self.client_lock.release()
def data_received(self, data_in):
asyncio.ensure_future(self.send_to_real_server(message, self.client_send_timeout))
def send_to_real_server(self, message, timeout=5.0):
yield from self.client_lock.acquire()
asyncio.ensure_future(asyncio.wait_for(self._send_to_real_server(message),
timeout, loop=self.loop)
).add_done_callback(self.sent_to_real_server)
@asyncio.coroutine
def _send_to_real_server(self, message):
self.client_transport.write(message)
def sent_to_real_server(self, task):
task.result()
self.client_lock.release()https://stackoverflow.com/questions/28609534
复制相似问题
腾讯云开发者
