python - 什么是 Python 中正确的无限套接字服务器循环

Question

我是 Python 新手，我的第一个任务是创建一个小型服务器程序，它将事件从网络单元转发到 rest api。 我的代码的整体结构似乎可行，但我有一个问题。在我收到第一个包裹后，没有任何反应。我的循环是否有问题，以至于不接受新包(来自同一客户端)？

包看起来像这样:EVNTTAG 20190219164001132%0C%3D%E2%80h%90%00%00%00%01%CBU%FB%DF ...这并不重要，但我分享只是为了清晰度。

我的代码(我跳过了rest等不相关的初始化，但主循环是完整的代码):

# Configure TAGP listener
ipaddress = ([l for l in ([ip for ip in socket.gethostbyname_ex(socket.gethostname())[2] if not ip.startswith("127.")][:1], [[(s.connect(('8.8.8.8', 53)), s.getsockname()[0], s.close()) for s in [socket.socket(socket.AF_INET, socket.SOCK_DGRAM)]][0][1]]) if l][0][0])
server_name = ipaddress
server_address = (server_name, TAGPListenerPort)
print ('starting TAGP listener on %s port %s' % server_address)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(server_address)
sock.listen(1)

sensor_data = {'tag': 0}

# Listen for TAGP data and forward events to ThingsBoard
try:
    while True:
        data = ""
        connection, client_address = sock.accept()
        data = str(connection.recv(1024))
        if data.find("EVNTTAG") != -1:
            timestamp = ((data.split())[1])[:17]
            tag = ((data.split())[1])[17:]
            sensor_data['tag'] = tag
            client.publish('v1/devices/me/telemetry', json.dumps(sensor_data), 1)
            print (data)
except KeyboardInterrupt:
    # Close socket server (TAGP)
    connection.shutdown(1)
    connection.close()
    # Close client to ThingsBoard
    client.loop_stop()
    client.disconnect()

Answer 1

您的代码存在多个问题:

首先，您需要对客户端发送的内容进行循环。所以你首先 connection, client_address = sock.accept() 现在你有了一个客户端。但是在循环的下一次迭代中，您再次执行 .accept() 并用新客户端覆盖旧的 connection。如果没有新客户，这将永远等待。这就是您所观察到的。

所以这可以这样解决:

while True:
    conn, addr = sock.accept()
    while True:
        data = conn.recv(1024)

但是这段代码还有另一个问题:在旧客户端断开连接之前，新客户端无法连接(好吧，目前它只是无限循环，无论客户端是否存活，我们稍后会处理它)。要克服它，您可以使用线程(或异步编程)并独立处理每个客户端。例如:

from threading import Thread

def client_handler(conn):
    while True:
        data = conn.recv(1024)

while True:
    conn, addr = sock.accept()
    t = Thread(target=client_handler, args=(conn,))
    t.start()

异步编程更难，我不打算在这里解决它。请注意，与线程相比，异步有很多优势(您可以通过谷歌搜索这些优势)。

现在每个客户端都有自己的线程，主线程只关心接受连接。事情同时发生。到目前为止一切顺利。

让我们关注client_handler 函数。您误解了套接字的工作原理。这:

data = conn.recv(1024)

不从缓冲区中读取 1024 字节。它实际上读取最多 1024 个字节，其中 0 也是可能的。即使您发送 1024 个字节，它仍然可以读取 3。当您收到长度为 0 的缓冲区时，这表明客户端已断开连接。所以首先你需要这个:

def client_handler(conn):
    while True:
        data = conn.recv(1024)
        if not data:
            break

现在真正的乐趣开始了。即使 data 是非空的，它也可以是 1 到 1024 之间的任意长度。您的数据可以分块并且可能需要多次 .recv 调用。不，对此您无能为力。由于某些其他代理服务器或路由器或网络滞后或宇宙辐射或其他原因，可能会发生分块。你必须为此做好准备。

因此，为了正确地使用它，您需要一个合适的框架协议(protocol)。例如，您必须以某种方式知道传入数据包有多大(以便您可以回答“我是否阅读了我需要的所有内容？”的问题)。一种方法是在每个帧前加上(比如)2 个字节，这些字节组合成帧的总长度。代码可能如下所示:

def client_handler(conn):
    while True:
        chunk = conn.recv(1)  # read first byte
        if not chunk:
            break
        size = ord(chunk)
        chunk = conn.recv(1)  # read second byte
        if not chunk:
            break
        size += (ord(chunk) << 8)

现在您知道传入缓冲区的长度为size。这样你就可以循环阅读所有内容:

def handle_frame(conn, frame):
    if frame.find("EVNTTAG") != -1:
        pass  # do your stuff here now

def client_handler(conn):
    while True:
        chunk = conn.recv(1)
        if not chunk:
            break
        size = ord(chunk)
        chunk = conn.recv(1)
        if not chunk:
            break
        size += (ord(chunk) << 8)
        # recv until everything is read
        frame = b''
        while size > 0:
            chunk = conn.recv(size)
            if not chunk:
                return
            frame += chunk
            size -= len(chunk)
        handle_frame(conn, frame)

重要提示:这只是一个处理协议(protocol)的示例，该协议(protocol)为每个帧添加长度前缀。请注意，客户端也必须调整。您要么必须定义这样的协议(protocol)，要么如果您有给定的协议(protocol)，则必须阅读规范并尝试理解框架的工作原理。例如，这与 HTTP 的做法非常不同。在 HTTP 中，您一直阅读直到遇到 \r\n\r\n ，它表示 header 结束。然后检查 Content-Length 或 Transfer-Encoding header (更不用说协议(protocol)切换等硬核内容)以确定下一步操作。但这变得相当复杂。我只是想让你知道还有其他选择。然而，框架是必要的。

网络编程也很难。我不会深入探讨安全性(例如对抗 DDOS)和性能等问题。上面的代码应该被视为极度简化，而不是生产就绪。我建议使用一些现有的软件。