具有异步计时器的Python异步websocket客户端

Question

我需要有一个长期运行的websocket客户端,它接收来自websocket服务器的推送消息,我需要监控客户端的连接状态:如果连接断开,我需要找出答案.

我的方法是定期记录一个常量字符串,并在未检测到日志消息时触发警报.

我的想法:1)有一个websocket客户端响应不规则传入的消息.2)同时有一个循环,当websocket客户端抛出一个ConnectionClosed执行时停止记录消息.

我对新的3.5异步语法很感兴趣.这个websocket实现特别基于asyncio.文档中的客户端看起来与我需要的完全一样.

但是,我不知道如何添加第二个coroutine来执行我的日志记录语句,并在websocket连接抛出ConnectionClosed时以某种方式停止.

这是开始对话的东西,但这不起作用,因为alive方法阻止事件循环.我正在寻找的是同时运行两种方法的优雅解决方案.

#!/usr/bin/env python

import asyncio
import logging

import websockets

logger = logging.getLogger(__name__)

is_alive = True


async def alive():
    while is_alive:
        logger.info('alive')
        await asyncio.sleep(300)


async def async_processing():
    async with websockets.connect('ws://localhost:8765') as websocket:
        while True:
            try:
                message = await websocket.recv()
                print(message)

            except websockets.exceptions.ConnectionClosed:
                print('ConnectionClosed')
                is_alive = False
                break


asyncio.get_event_loop().run_until_complete(alive())
asyncio.get_event_loop().run_until_complete(async_processing())

Answer 1

实际上run_until_complete这里是封锁,因为它等到alive完成.

您可以通过两个步骤解决它:

1. 安排协同程序asyncio.ensure_future(立即运行而不等待结果),每个返回任务.
2. 等待任务完成 asyncio.wait

代码如:

tasks = [
   asyncio.ensure_future(alive()),
   asyncio.ensure_future(async_processing())
]
asyncio.get_event_loop().run_until_complete(asyncio.wait(tasks))

正如@Vincent提到的那样wait接受任务,所以ensure_future也是不必要的:

asyncio.get_event_loop().run_until_complete(asyncio.wait([   
   alive(),
   async_processing()
]))