Python异步强制时间ou

Question

使用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))

因为time.sleep(10)没有被取消，所以这需要10秒以上的时间来运行。在这种情况下，是否可以强制取消联程？

如果应该使用asyncio来解决这个问题，我该怎么做呢？

Answer 1

不，除非将控制权交回事件循环，否则不能中断协程，这意味着它需要在一个yield from调用中。asyncio是单线程的，因此当您在第二个示例中阻塞time.sleep(10)调用时，事件循环无法运行。这意味着当您使用wait_for设置的超时过期时，事件循环将无法对其执行操作。直到coro退出，事件循环才有机会再次运行，此时太晚了。

这就是为什么在一般情况下，您应该始终避免任何非异步的阻塞调用；每当调用阻塞而不屈服于事件循环时，程序中的任何其他内容都无法执行，这可能不是您想要的。如果确实需要执行长时间的阻塞操作，则应尝试使用^{}在线程或进程池中运行该操作，这将避免阻塞事件循环：

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))

Answer 2

谢谢你的回答。如果运行coroutine不是一个硬性要求，那么这里有一个经过修改的、更紧凑的版本

import asyncio, time, concurrent

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 concurrent.futures.TimeoutError:
    print('I need more sleep !')

奇怪的是，我的Python 3.5.2中的一点调试表明，将None作为执行器传递会导致创建_default_executor，如下所示：

# _MAX_WORKERS = 5
self._default_executor = concurrent.futures.ThreadPoolExecutor(_MAX_WORKERS)

Answer 3

我看到的超时处理示例非常简单。考虑到现实情况，我的应用程序要复杂一些。顺序是：

1. 当客户端连接到服务器时，让服务器创建到内部服务器的另一个连接
2. 当内部服务器连接正常时，等待客户端发送数据。基于这些数据，我们可以向内部服务器进行查询。
3. 当有数据要发送到内部服务器时，发送它。由于内部服务器有时响应不够快，请将此请求包装为超时。
4. 如果操作超时，则折叠所有连接以向客户端发出错误信号

为了实现上述所有功能，在保持事件循环运行的同时，生成的代码包含以下代码：

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()