python-如果事件循环已经在运行,如何等待协程在方法中同步完成？
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了python-如果事件循环已经在运行,如何等待协程在方法中同步完成？，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含5376字，纯文字阅读大概需要8分钟。
内容图文
我正在尝试创建一个基于Python的CLI,该CLI通过websockets与Web服务进行通信.我遇到的一个问题是CLI间歇性地向Web服务发出的请求无法得到处理.查看来自Web服务的日志,我可以看到问题是由以下事实引起的：套接字关闭的同时(甚至在关闭之后)经常发出这些请求：
2016-09-13 13:28:10,930 [22 ] INFO DeviceBridge - Device bridge has opened
2016-09-13 13:28:11,936 [21 ] DEBUG DeviceBridge - Device bridge has received message
2016-09-13 13:28:11,937 [21 ] DEBUG DeviceBridge - Device bridge has received valid message
2016-09-13 13:28:11,937 [21 ] WARN DeviceBridge - Unable to process request: {"value": false, "path": "testcube.pwms[0].enabled", "op": "replace"}
2016-09-13 13:28:11,936 [5 ] DEBUG DeviceBridge - Device bridge has closed
在我的CLI中,我定义了一个CommunicationService类,该类负责处理与Web服务的所有直接通信.在内部,它使用websockets程序包来处理通信,该程序包本身是建立在asyncio之上的.
CommunicationService包含以下发送请求的方法：
def send_request(self, request: str) -> None:
logger.debug('Sending request: {}'.format(request))
asyncio.ensure_future(self._ws.send(request))
…其中ws是先前通过另一种方法打开的websocket：
self._ws = await websockets.connect(websocket_address)
我想要的是能够等待asyncio.ensure_future返回的未来,并且如果有必要的话,可以在之后等待一小段时间,以便在关闭websocket之前为Web服务腾出时间来处理请求.
但是,由于send_request是同步方法,因此不能简单地等待这些将来.使它异步将是没有意义的,因为没有什么可以等待它返回的协程对象.我也不能使用loop.run_until_complete,因为循环在调用时已经在运行.
我发现有人描述了一个非常类似于我在mail.python.org处的问题.在该线程中发布的解决方案是,在循环已经运行的情况下,使该函数返回协程对象：
def aio_map(coro, iterable, loop=None):
if loop is None:
loop = asyncio.get_event_loop()
coroutines = map(coro, iterable)
coros = asyncio.gather(*coroutines, return_exceptions=True, loop=loop)
if loop.is_running():
return coros
else:
return loop.run_until_complete(coros)
这对我来说是不可能的,因为我正在使用PyRx(反应性框架的Python实现),并且send_request仅作为可观察到的Rx的订户调用,这意味着返回值将被丢弃并且对我的代码不可用：
class AnonymousObserver(ObserverBase):
...
def _on_next_core(self, value):
self._next(value)
附带说明一下,我不确定这是否是asyncio经常遇到的某种问题,或者我是否没有得到它,但是我发现使用它非常令人沮丧.在C#中(例如),我所需要做的可能只是以下内容：
void SendRequest(string request)
{
this.ws.Send(request).Wait();
// Task.Delay(500).Wait(); // Uncomment If necessary
}
同时,asyncio的“ wait”版本只会返回另一个我被迫丢弃的协程.
更新资料
我找到了解决该问题的方法,该方法似乎可行.我有一个异步回调,该回调在命令执行后且CLI终止之前被执行,因此我从此进行了更改…
async def after_command():
await comms.stop()
…对此：
async def after_command():
await asyncio.sleep(0.25) # Allow time for communication
await comms.stop()
不过,我仍然很高兴收到有关此问题的任何答案,以供将来参考.在其他情况下,我可能无法依靠这种变通方法,并且我仍然认为,最好在send_request内部执行延迟,这样CommunicationService的客户端不必担心计时问题,这是更好的做法.
关于文森特的问题：
Does your loop run in a different thread, or is send_request called by some callback?
一切都在同一线程中运行-由回调调用.发生的是,我定义了所有命令以使用异步回调,并且在执行其中的某些命令后,它们将尝试向Web服务发送请求.由于它们是异步的,因此它们只有在通过CLI顶层的loop.run_until_complete调用执行后才执行此操作-这意味着循环将在执行和执行过程中途运行.发出此请求(通过间接调用send_request).
更新2
这是基于Vincent建议添加“完成”回调的解决方案.
将一个新的布尔字段_busy添加到CommunicationService,以表示是否正在进行通讯活动.
在发送请求之前,对CommunicationService.send_request进行了修改,以将_busy设置为true,然后为_ws.send提供回调以在完成后重置_busy：
def send_request(self, request: str) -> None:
logger.debug('Sending request: {}'.format(request))
def callback(_):
self._busy = False
self._busy = True
asyncio.ensure_future(self._ws.send(request)).add_done_callback(callback)
现在实现了CommunicationService.stop,以等待在继续进行操作之前将该标志设置为false：
async def stop(self) -> None:
"""
Terminate communications with TestCube Web Service.
"""
if self._listen_task is None or self._ws is None:
return
# Wait for comms activity to stop.
while self._busy:
await asyncio.sleep(0.1)
# Allow short delay after final request is processed.
await asyncio.sleep(0.1)
self._listen_task.cancel()
await asyncio.wait([self._listen_task, self._ws.close()])
self._listen_task = None
self._ws = None
logger.info('Terminated connection to TestCube Web Service')
这似乎也可行,并且至少以这种方式,所有通信时序逻辑都应按原样封装在CommunicationService类中.
更新3
基于Vincent提议的更好的解决方案.
而不是self._busy,我们有self._send_request_tasks = [].
新的send_request实现：
def send_request(self, request: str) -> None:
logger.debug('Sending request: {}'.format(request))
task = asyncio.ensure_future(self._ws.send(request))
self._send_request_tasks.append(task)
新的停止实施：
async def stop(self) -> None:
if self._listen_task is None or self._ws is None:
return
# Wait for comms activity to stop.
if self._send_request_tasks:
await asyncio.wait(self._send_request_tasks)
...
解决方法:
您可以使用一组任务：
self._send_request_tasks = set()
使用sure_future计划任务,并使用add_done_callback进行清理：
def send_request(self, request: str) -> None:
task = asyncio.ensure_future(self._ws.send(request))
self._send_request_tasks.add(task)
task.add_done_callback(self._send_request_tasks.remove)
并等待一组任务完成：
async def stop(self):
if self._send_request_tasks:
await asyncio.wait(self._send_request_tasks)
内容总结
以上是互联网集市为您收集整理的python-如果事件循环已经在运行,如何等待协程在方法中同步完成？全部内容，希望文章能够帮你解决python-如果事件循环已经在运行,如何等待协程在方法中同步完成？所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。