带有异步计时器的 Python 异步 websocket 客户端 [英] Python async websocket client with async timer

问题描述

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

I need to have a long running websocket client that receives push messages from a websocket server and I need to monitor the client's connection state: if the connection goes down, I need to find out.

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

My approach is to periodically log a constant string, and trigger an alarm if ever the log message is not detected.

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

My idea: 1) have a websocket client that responds to irregularly incoming messages. And 2) at the same time have loop that stops logging a message when the websocket client throws a ConnectionClosed exeption.

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

I am intrigued by the new 3.5 async syntax. This websocket implementation is specifically based on the asyncio. The client in the docs look exactly like what I need.

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

However, I have no idea how to add a second coroutine that does my logging statements and somehow stops when the websocket connection throws the ConnectionClosed.

这里有一些东西可以开始对话，但它不起作用，因为alive 方法阻塞了事件循环.我正在寻找的是同时运行这两种方法的优雅解决方案.

Here is something to start the conversation but that doesn't work because the alive method blocks the event loop. What I am looking for is an elegant solution to run both methods concurrently.

#!/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())

推荐答案

实际上 run_until_complete 在这里阻塞，因为它一直等到 alive 完成.

Actually the run_until_complete is blocking here, since it waits until alive finish.

你可以分两步解决:

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 是不必要的:

As @Vincent mentioned wait accepts tasks, so ensure_future is needless:

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

这篇关于带有异步计时器的 Python 异步 websocket 客户端的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！