如何使用socketserver解析TCP客户端的CRLF格式响应？

我用Python和socketserver编写多线程TCP服务器，需要向设备发送命令并读取响应。设备连接后会先发送初始消息，后续命令必须在处理完初始消息后再发送。服务器支持多连接，设备返回的响应是包含CRLF(\r\n)的长字符串，每条消息以$开头、长度不固定，示例响应如下：

b'$\r\n$Name=TEST-Device\r\n$Ext=Test connection\r\n$Security=0\r\n$Software=xxx\r\n$Hardware=xxx\r\n$SUCCESS\r\n$\r\n'

接下来需要解析该消息，检查Name是否为TEST-Device，匹配后发送后续命令。但当前代码无法正确拆分并存储消息，想请教正确的处理方式——是否应使用缓冲区、BytesIO或asyncio？我的目标是实现服务器到设备的命令队列和存储所有响应的响应队列，或许需要单独的Message类。

当前服务器代码如下：

class TCPServerMultiClient:
    ...

    def handle_client(self, client_sock, client_address):

        try:
            time.sleep(5)
            print("- Waiting for initial message -")
            data = client_sock.recv(1024)
            print(data)
            while data:
                # parse the message from client
                data = data.decode('utf-8')
                i = data.rfind("\r\n")
                data, buf = data[:i + 2], data[i + 2:]
                lines = data.split("\r\n")
                lines = filter(None, lines)
                print(lines)
                for line in lines:
                    if line.startswith("$"):
                        symbol = str(line.split(",")[])
                # pseudo code! send next commands only if Name matches (from previous response)
                if Name="TEST-Deivce":

                    print("- Sending first command")
                    cmdToSend=b"first_command"
                    client_sock.sendall(cmdToSend)
                
                    time.sleep(2)            
                    print("- Recieving response")
                    data1 = client_sock.recv(1024)
                    print(data1)

                # check if client closed the connection
                data = client_sock.recv(4024)
            print(f'Connection closed by {client_address}')

        except socket.error as e:
            if e.errno == errno.ECONNABORTED:
                print(os.strerror(e.errno))
            else:
                print(str(e))
                raise

        finally:
            # close socket
            print(f'Closing client socket for {client_address}...')
            client_sock.close()
            print(f'Client socket closed for {client_address}')

    def shutdown_server(self):
        """ Shutdown the server """

        print('Shutting down server...')
        self.sock.close()

解决方案 #

1. 正确的消息拆分与缓冲处理 #

当前代码未处理粘包或不完整消息的情况，需维护字节缓冲区，每次recv后追加到缓冲区，按CRLF拆分完整行，剩余不完整部分留在缓冲区等待下一次数据：

def handle_client(self, client_sock, client_address):
    buffer = bytearray()
    device_name = None
    try:
        print("- Waiting for initial message -")
        while True:
            chunk = client_sock.recv(1024)
            if not chunk:
                break
            buffer.extend(chunk)
            # 拆分完整的行（以\r\n分隔）
            while b'\r\n' in buffer:
                line_end = buffer.find(b'\r\n')
                line = buffer[:line_end].decode('utf-8').strip()
                buffer = buffer[line_end + 2:]  # 移除已处理的行和分隔符
                
                if line.startswith('$'):
                    # 解析行内容
                    if line.startswith('$Name='):
                        device_name = line.split('=', 1)[1]
                        print(f"设备名称：{device_name}")
                    elif line == '$<end>':
                        # 初始消息接收完成，检查设备名称
                        if device_name == 'TEST-Device':
                            self.send_command_queue(client_sock, client_address)
                    # 将响应存入队列
                    self.add_to_response_queue(line, client_address)
    # 保留原有异常处理和关闭逻辑
    except socket.error as e:
        if e.errno == errno.ECONNABORTED:
            print(os.strerror(e.errno))
        else:
            print(str(e))
            raise
    finally:
        print(f'Closing client socket for {client_address}...')
        client_sock.close()
        del self.client_queues[client_address]  # 清理队列
        print(f'Client socket closed for {client_address}')

2. 实现线程安全的命令队列与响应队列 #

使用queue.Queue实现线程安全的队列，为每个客户端维护独立的命令和响应队列：

import queue

class TCPServerMultiClient:
    def __init__(self):
        self.client_queues = {}  # key: client_address, value: (cmd_queue, resp_queue)
        # 其他初始化逻辑...

    def handle_client(self, client_sock, client_address):
        cmd_queue = queue.Queue()
        resp_queue = queue.Queue()
        self.client_queues[client_address] = (cmd_queue, resp_queue)
        # 预添加需要发送的命令到队列
        cmd_queue.put(b"first_command")
        cmd_queue.put(b"second_command")
        # 后续处理逻辑...

    def send_command_queue(self, client_sock, client_address):
        # 从当前客户端的命令队列依次发送命令
        cmd_queue, _ = self.client_queues[client_address]
        while not cmd_queue.empty():
            cmd = cmd_queue.get()
            print(f"- 发送命令：{cmd}")
            client_sock.sendall(cmd + b'\r\n')  # 假设命令需要CRLF结尾
            # 可结合缓冲逻辑等待响应，此处简化用sleep
            time.sleep(2)

    def add_to_response_queue(self, line, client_address):
        _, resp_queue = self.client_queues[client_address]
        resp_queue.put(line)

3. 是否需要使用asyncio？ #

如果服务器需要处理数百上千的并发连接，asyncio的异步IO模型效率更高；若当前并发量不大，多线程模型结合线程安全队列足够满足需求，改造成本更低。

4. 可选：单独的Message类封装解析逻辑 #

若消息结构复杂，可定义DeviceMessage类封装解析和存储逻辑，让代码更清晰：

class DeviceMessage:
    def __init__(self):
        self.name = None
        self.ext = None
        self.security = None
        self.software = None
        self.hardware = None
        self.status = None
        self.complete = False

    def parse_line(self, line):
        if line.startswith('$Name='):
            self.name = line.split('=',1)[1]
        elif line.startswith('$Ext='):
            self.ext = line.split('=',1)[1]
        elif line.startswith('$Security='):
            self.security = line.split('=',1)[1]
        elif line.startswith('$Software='):
            self.software = line.split('=',1)[1]
        elif line.startswith('$Hardware='):
            self.hardware = line.split('=',1)[1]
        elif line == '$SUCCESS':
            self.status = 'SUCCESS'
        elif line == '$<end>':
            self.complete = True

内容的提问来源于stack exchange，提问作者Preeti