Python 3 - 多个 AsyncIO 连接

Question

我正在尝试学习如何在 Python 3.7 中使用 AsyncIO，但我仍然对其原理有些困惑。

我的目标是编写一个简单的聊天程序，但是我需要使用环形网络拓扑——一个节点只知道它的两个邻居。当消息被发送时，它会被节点传递，直到它再次到达发送者。这意味着每个节点基本上同时是客户端和服务器。

我还需要能够检测到死节点，以便我的环不会损坏。

我认为每个节点为每个邻居建立单独的连接可能是一个很好的解决方案 -successor和predecessor.

class Node:
    ...
    def run():
        ...
        s = loop.create_connection(lambda: Client(...), addr1, port1)
        p = loop.create_server(lambda: Server(...), addr2, port2)
        successor = loop.run_until_complete(s)
        predecessor = loop.run_until_complete(p)
        loop.run_forever()
        ...
    ...

Server并且Client是实现asyncio.Protocol.

我想这样做的原因是，如果有消息通过圈子发送，它总是从 发送predecessor到successor。在I 的connection_lost方法中，predecessor我可以检测到它已断开连接并发送它predecessor的消息（通过整个环）以连接到我。

我希望能够将我从我predecessor进一步收到的消息发送到我的successor. 我还希望能够将包含我的地址的消息发送给我successor，以防万一我predecessor死了（此消息将从predecessor's发送Server.connection_lost()并一直传递到我死者predecessor的predecessor）。

我的问题是：我可以将接收到的数据从predecessorto传递过来successor吗？如果不是，那么使用 AsyncIO 和环形拓扑的这个程序的更好实现是什么？

Answer 1

对于遇到相同问题的 AsyncIO 新手，我自己找到了解决方案。

首先，最好使用 AsyncIO 的高级方面 -- streams。调用loop.create_connction和loop.create_server被认为是低级的（我一开始理解错了）。

高层次的替代create_connection是asyncio.open_connection，这将提供给您一个元组组成的asyncio.StreamReader和asyncio.StreamWriter你可以用它来读取和写入到打开的连接。您还可以在从StreamReaderequals读取数据时b''或在ConnectionError尝试写入StreamWriter.

的高级替代方案create_server是asyncio.start_server，需要提供一个回调函数，每次与服务器建立连接（打开连接，接收数据......）时都会调用该函数。回调有StreamReader和StreamWriter作为参数。也可以通过接收b''或ConnectionError写入writer.

协程可以处理多个连接。服务器部分可以有一个协程（它接受来自环拓扑中一个邻居的连接）和一个客户端部分（打开到环中另一个邻居的连接）的协程。本Node类可以是这样的：

import asyncio

class Node:
    ...
    async def run(self):
        ...
        self.next_reader, self.next_writer = await asyncio.open_connection(self.next_IP, self.next_port)
        server_coro = asyncio.create_task(self.server_init())
        client_coro = asyncio.create_task(self.client_method())
        await client_coro
        await server_coro
        ...

    async def server_init(self):
        server = await asyncio.start_server(self.server_callback, self.IP, self.port)
        async with server:
            await server.serve_forever()

    async def client_method(self):
        ...
        try:
            data = await self.next_reader.read()
        except ConnectionError:
            ...
    ...

请注意，我使用asyncio.create_task的是协程 and （不在代码清单中）asyncio.run(node.run())，它们被认为是asyncio.ensure_future()and的高级替代品loop.run_forever()。这两个都是在 Python 3.7 中添加的，asyncio.run()据说是临时的，所以当你阅读这篇文章的时候，它可能已经被其他东西取代了。

我不是 AsyncIO 专家，所以可能有更好、更简洁的方法来做到这一点（如果你知道，请分享）。