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

Question

我是一个 Python 新手，我的第一个任务是创建一个小型服务器程序，它将事件从网络单元转发到 REST API。我的代码的整体结构似乎有效，但我有一个问题。我收到第一个包裹后，没有任何反应。我的循环是否有问题，导致不接受新包（来自同一客户端）？

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

我的代码（我跳过了不相关的休息初始化等，但主循环是完整的代码）：

# 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调用。不，你对此无能为力。由于某些其他代理服务器或路由器或网络延迟或宇宙辐射或其他原因，可能会发生分块。你必须为此做好准备。

因此，为了正确使用它，您需要一个适当的成帧协议。例如，您必须以某种方式知道传入数据包有多大（以便您可以回答“我是否阅读了我需要的所有内容？”的问题）。实现此目的的一种方法是为每个帧添加（例如）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)

现在您知道传入缓冲区的长度为 length 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)

重要提示：这只是处理为每个帧添加其长度前缀的协议的示例。请注意，客户端也必须进行调整。您要么必须定义这样的协议，要么如果您有给定的协议，则必须阅读规范并尝试了解框架的工作原理。例如，这与 HTTP 的实现方式非常不同。在 HTTP 中，您会一直阅读，直到遇到\r\n\r\n表示标头结束的信号。然后你检查Content-Length或Transfer-Encoding标头（更不用说协议切换等硬核内容）来确定下一步操作。但这变得相当复杂。我只是想让你知道还有其他选择。尽管如此，框架还是必要的。

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