时间戳

首页 > 解决方案 > 具有多个客户端和无限循环的 Python 异步协议行为

python-3.x - 具有多个客户端和无限循环的 Python 异步协议行为

问题描述

我很难理解我改变的回声服务器的行为，它试图利用 python 3 的asyncio模块。

本质上我有一个无限循环（假设我想在建立连接的同时将一些数据从服务器无限期地流式传输到客户端）例如MyServer.py：

#! /usr/bin/python3
import asyncio
import os
import time

class MyProtocol(asyncio.Protocol):

    def connection_made(self, transport):
        peername = transport.get_extra_info('peername')
        print('Connection from {}'.format(peername))
        self.transport = transport

    def connection_lost(self, exc):
        asyncio.get_event_loop().stop()

    def data_received(self, data):
        i = 0
        while True:
            self.transport.write(b'>> %i' %i)
            time.sleep(2)
            i+=1

loop = asyncio.get_event_loop()
coro = loop.create_server(MyProtocol, 
    os.environ.get('MY_SERVICE_ADDRESS', 'localhost'), 
    os.environ.get('MY_SERVICE_PORT', 8100))
server = loop.run_until_complete(coro)

try:
    loop.run_forever()
except:
    loop.run_until_complete(server.wait_closed())
finally:
    loop.close()

接下来，当我连接nc ::1 8100并发送一些文本（例如“测试”）时，我得到以下信息：

user@machine$ nc ::1 8100
*** Connection from('::1', 58503, 0, 0) ***
testing
>> 1
>> 2
>> 3
^C

现在，当我尝试nc再次使用连接时，我没有收到任何欢迎消息，并且在我尝试向服务器发送一些新文本后，我收到了源源不断的以下错误：

user@machine$ nc ::1 8100
Is there anybody out there?
socket.send() raised exception
socket.send() raised exception
...
^C

只是为了在伤口上加盐，socket.send() raised exception消息继续向我的终端发送垃圾邮件，直到我杀死 python 服务器进程......

由于我是网络技术的新手（作为桌面恐龙太久了！），我不确定为什么会出现上述行为，而且我不知道如何产生预期的行为，这很松散看起来像这样：

  1. 服务器启动
  2. 客户端 1 连接到服务器
  3. 服务器向客户端 4 发送欢迎消息 客户端 1 发送任意消息
  4. 只要客户端连接，服务器就会将消息发送回客户端 1
  5. 客户端 1 断开连接（假设电缆被拔出）
  6. 客户端 2 连接到服务器
  7. 对客户端 2 重复步骤 3-6

任何启蒙都将非常受欢迎！

标签： python-3.xserverclient-servermicroservicespython-asyncio

解决方案

代码存在多个问题。

首先，data_received 永不返回。在传输/协议级别，异步编程是单线程和基于回调的。应用程序代码分散在回调中，例如data_received，事件循环运行节目，监控文件描述符并根据需要调用回调。每个回调只允许执行一个简短的计算，调用传输方法，并安排进一步的回调执行。回调不能做的是花费大量时间来完成或阻止等待某事。永不退出的while循环尤其糟糕，因为它根本不允许事件循环运行。

这就是为什么代码只在客户端断开连接时才吐出异常的原因：connection_lost从不调用。它应该由事件循环调用，并且永不返回data_received不会让事件循环有机会恢复。由于事件循环被阻塞，程序无法响应其他客户端，并data_received不断尝试向断开连接的客户端发送数据，并记录其失败。

表达想法的正确方式可以如下所示：

def data_received(self, data):
    self.i = 0
    loop.call_soon(self.write_to_client)

def write_to_client(self):
    self.transport.write(b'>> %i' % self.i)
    self.i += 1
    loop.call_later(2, self.write_to_client)

注意两者都data_received做write_to_client很少的工作并迅速返回。没有对 的调用time.sleep()，也绝对没有无限循环——“循环”隐藏在对 的递归调用中write_to_client。

此更改揭示了代码中的第二个问题。它MyProtocol.connection_lost停止整个事件循环并退出程序。这使得程序无法响应第二个客户端。解决方法可能是替换loop.stop()为在以下位置设置标志connection_lost：

def data_received(self, data):
    self._done = False
    self.i = 0
    loop.call_soon(self.write_to_client)

def write_to_client(self):
    if self._done:
        return
    self.transport.write(b'>> %i' % self.i)
    self.i += 1
    loop.call_later(2, self.write_to_client)

def connection_lost(self, exc):
    self._done = True

这允许多个客户端连接。

与上述问题无关，基于回调的代码写起来有点累，尤其是考虑到复杂的代码路径和异常处理时。（想象一下，试图用回调来表达嵌套循环，或者在深度嵌入的回调中传播发生的异常。）asyncio 支持基于协程的流作为基于回调的传输和协议的替代方案。

协程允许编写包含循环的看起来自然的代码，并且看起来它包含阻塞调用，这些调用在后台被转换为使事件循环能够恢复的暂停点。使用流，问题中的代码如下所示：

async def talk_to_client(reader, writer):
    peername = writer.get_extra_info('peername')
    print('Connection from {}'.format(peername))

    data = await reader.read(1024)
    i = 0
    while True:
        writer.write(b'>> %i' % i)
        await writer.drain()
        await asyncio.sleep(2)
        i += 1

loop = asyncio.get_event_loop()
coro = asyncio.start_server(talk_to_client, 
    os.environ.get('MY_SERVICE_ADDRESS', 'localhost'), 
    os.environ.get('MY_SERVICE_PORT', 8100))
server = loop.run_until_complete(coro)

loop.run_forever()

talk_to_client看起来非常像 的原始实现data_received，但没有缺点。await如果数据不可用，则在它使用事件循环的每个点恢复。time.sleep(n)替换await asyncio.sleep(n)为相当于loop.call_later(n, <resume current coroutine>). 等待writer.drain()确保协程在对等方无法处理它获得的输出时暂停，并在对等方断开连接时引发异常。

推荐阅读