python - 异步多个并发服务器

Question

我正在尝试使用 Python 的 asyncio 一起运行多个服务器，并在它们之间传递数据。对于我的具体情况，我需要一个带有 websockets 的 web 服务器，一个到外部设备的 UDP 连接，以及数据库和其他交互。我可以单独找到几乎所有这些示例，但我正在努力找出正确的方法让它们与在它们之间推送的数据同时运行。

我在这里找到的最接近的是:Communicate between asyncio protocol/servers (虽然我一直无法让它在 Python 3.6 上运行)

更具体的示例:我将如何从 https://github.com/aio-libs/aiohttp 中获取以下 aiohttp 示例代码:

from aiohttp import web

async def handle(request):
    name = request.match_info.get('name', "Anonymous")
    text = "Hello, " + name
    return web.Response(text=text)

async def wshandler(request):
    ws = web.WebSocketResponse()
    await ws.prepare(request)

    async for msg in ws:
        if msg.type == web.MsgType.text:
            await ws.send_str("Hello, {}".format(msg.data))
        elif msg.type == web.MsgType.binary:
            await ws.send_bytes(msg.data)
        elif msg.type == web.MsgType.close:
            break

    return ws


app = web.Application()
app.router.add_get('/echo', wshandler)
app.router.add_get('/', handle)
app.router.add_get('/{name}', handle)

web.run_app(app)

和以下 TCP 回显服务器示例 ( http://asyncio.readthedocs.io/en/latest/tcp_echo.html ):

import asyncio

async def handle_echo(reader, writer):
    data = await reader.read(100)
    message = data.decode()
    addr = writer.get_extra_info('peername')
    print("Received %r from %r" % (message, addr))

    print("Send: %r" % message)
    writer.write(data)
    await writer.drain()

    print("Close the client socket")
    writer.close()

loop = asyncio.get_event_loop()
coro = asyncio.start_server(handle_echo, '127.0.0.1', 8888, loop=loop)
server = loop.run_until_complete(coro)

# Serve requests until Ctrl+C is pressed
print('Serving on {}'.format(server.sockets[0].getsockname()))
try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

# Close the server
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()

并将它们组合成一个脚本，其中通过 websockets 或 TCP 回显服务器接收到的任何消息都被发送到其中任何一个的所有客户端？

我如何添加一段代码(比如说)每秒向所有客户端发送一条消息(为了论证当前时间戳)？

Answer 1

首先，您需要将所有协程放入一个事件循环中。您可以从避免为您启动事件循环的便捷 API 开始，例如 run_app。代替 web.run_app(app)，编写如下内容:

runner = aiohttp.web.AppRunner(app)
loop.run_until_complete(runner.setup())
# here you can specify the listen address and port
site = aiohttp.web.TCPSite(runner)    
loop.run_until_complete(site.start())

然后运行 ​​echo 服务器设置，两者都准备好共享 asyncio 事件循环。在脚本末尾，使用 loop.run_forever()(或以您的应用程序中有意义的任何其他方式)启动事件循环。

要向客户端广播信息，创建一个广播协程并将其添加到事件循环中:

# Broadcast data is transmitted through a global Future. It can be awaited
# by multiple clients, all of which will receive the broadcast. At each new
# iteration, a new future is created, to be picked up by new awaiters.
broadcast_data = loop.create_future()

async def broadcast():
    global broadcast_data
    while True:
        broadcast_data.set_result(datetime.datetime.now())
        broadcast_data = loop.create_future()
        await asyncio.sleep(1)

loop.create_task(broadcast())

最后，等待为客户端创建的每个协程中的广播，例如handle_echo:

def handle_echo(r, w):
    while True:
        data = await broadcast_data
        # data contains the broadcast datetime - send it to the client
        w.write(str(data))

修改 websockets 处理程序协同程序以等待和以相同方式中继广播数据应该很简单。