Python Socket实现WebSocket：压缩帧无效问题排查与优化

我正在用Python的socket库构建WebSocket服务器，运行在无法安装第三方包的嵌入式Python环境中。目前已完成握手并建立连接，服务器与React客户端可互传数据，但部分载荷过大，分批传输速度过慢。于是我用zlib压缩数据，却出现“Invalid WebSocket Frame”错误。

可复现代码（Python 3.9-3.11兼容） #

import socket
import struct
import base64
import hashlib
import zlib
import logging
import json
from threading import Thread

class WebSocketServer(Thread):
    def __init__(self):
        Thread.__init__(self)
        self.connection = None
        self.logger = logging.getLogger('WebSocketServer')

    def run(self):
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            sock.bind(('', 5558))
            self.logger.info("Server started, waiting for connections...")
            sock.listen(1)
            while True:
                connection, _ = sock.accept()
                if connection:
                    self.connection = connection
                    self.logger.info("Client connected")
                    Thread(target=self.handle_connection, args=[connection]).start()
        except Exception as e:
            self.logger.error(f'Run error: {e}')
        finally:
            if self.connection:
                self.connection.close()
                self.logger.info('Server socket closed')

    def handle_connection(self, connection):
        try:
            if self.perform_handshake(connection):
                while True:
                    msg = self.receive_message(connection)
                    if msg:
                        self.logger.info(f'Received message: {msg}')
                        # Echo messages back
                        self.send_message(json.dumps(msg))
                    else:
                        break
        except Exception as e:
            self.logger.error(f'Connection error: {e}')
        finally:
            connection.close()
            self.logger.info('Connection closed')

    def perform_handshake(self, connection):
        try:
            self.logger.info("Performing handshake...")
            request = connection.recv(1024).decode('utf-8')
            self.logger.info(f"Handshake request: {request}")

            headers = self.parse_headers(request)
            websocket_key = headers['Sec-WebSocket-Key']
            websocket_accept = self.generate_accept_key(websocket_key)

            response = (
                'HTTP/1.1 101 Switching Protocols\r\n'
                'Upgrade: websocket\r\n'
                'Connection: Upgrade\r\n'
                f'Sec-WebSocket-Accept: {websocket_accept}\r\n\r\n'
            )

            connection.send(response.encode('utf-8'))
            self.logger.info("Handshake response sent")
            return True
        except Exception as e:
            self.logger.error(f'Handshake error: {e}')
            return False

    def parse_headers(self, request):
        headers = {}
        lines = request.split('\r\n')
        for line in lines[1:]:
            if line:
                key, value = line.split(': ', 1)
                headers[key] = value
        return headers

    def generate_accept_key(self, websocket_key):
        magic_string = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
        accept_key = base64.b64encode(hashlib.sha1((websocket_key + magic_string).encode()).digest()).decode('utf-8')
        return accept_key

    def receive_message(self, connection):
        try:
            data = connection.recv(1024)
            if not data:
                return None

            byte1, byte2 = struct.unpack('BB', data[:2])
            fin = byte1 & 0b10000000
            opcode = byte1 & 0b00001111
            masked = byte2 & 0b10000000
            payload_length = byte2 & 0b01111111

            if masked != 0b10000000:
                self.logger.error('Client data must be masked')
                return None

            if payload_length == 126:
                extended_payload_length = data[2:4]
                payload_length = int.from_bytes(extended_payload_length, byteorder='big')
                masking_key = data[4:8]
                payload_data = data[8:]
            elif payload_length == 127:
                extended_payload_length = data[2:10]
                payload_length = int.from_bytes(extended_payload_length, byteorder='big')
                masking_key = data[10:14]
                payload_data = data[14:]
            else:
                masking_key = data[2:6]
                payload_data = data[6:]

            decoded_bytes = bytearray()
            for i in range(payload_length):
                decoded_bytes.append(payload_data[i] ^ masking_key[i % 4])

            if opcode == 0x1:  # Text frame
                return decoded_bytes.decode('utf-8')
            elif opcode == 0x8:  # Connection close frame
                self.logger.info('Connection closed by client')
                return None
            else:
                self.logger.error(f'Unsupported frame type: {opcode}')
                return None
        except Exception as e:
            self.logger.error(f'Error receiving message: {e}')
            return None

    def send_message(self, message):
        try:
            if self.connection and isinstance(message, str):
                # Compress the message using zlib
                compressed_message = zlib.compress(message.encode('utf-8'))

                # Determine chunk size based on network conditions
                max_chunk_size = 1024  # Adjust as needed

                # Split the compressed message into smaller chunks
                chunks = [compressed_message[i:i+max_chunk_size] for i in range(0, len(compressed_message), max_chunk_size)]

                for chunk in chunks:
                    frame = bytearray()
                    frame.append(0b10000001)  # Text frame opcode

                    length = len(chunk)
                    if length <= 125:
                        frame.append(length)
                    elif length <= 65535:
                        frame.append(126)
                        frame.extend(struct.pack('!H', length))
                    else:
                        frame.append(127)
                        frame.extend(struct.pack('!Q', length))

                    # Append the chunk to the frame
                    frame.extend(chunk)

                    # Send the framed chunk
                    self.connection.sendall(frame)
            else:
                self.logger.error("Connection closed or invalid message")
        except Exception as e:
            self.logger.error(f'Error sending message: {e}')
            
            
# Configure logging
logging.basicConfig(level=logging.INFO)

# Create an instance of WebSocketServer
server = WebSocketServer()

# Start the server
server.start()

测试步骤与错误信息 #

使用wscat测试：

wscat -c ws://127.0.0.1:5558

输入任意消息后收到错误：
error: Invalid WebSocket frame: invalid UTF-8 sequence

三个问题 #

1. 为何出现该错误？
2. 正确的WebSocket消息帧格式应该是怎样的？
3. 是否有更高效的实现方式？

补充背景：传输的是MIDI SysEx数据，为整数数组的数组，每个数组以0xF0开头、0xF7结尾，消息发送频繁且初始消息可能包含数千个SysEx数组，此前通过类似WebSocket传输过该数据。

1. 错误原因分析 #

你把压缩后的二进制数据当作文本帧（opcode=0x1）发送了。WebSocket的文本帧要求载荷必须是合法的UTF-8编码，但zlib压缩后的字节流是二进制数据，包含大量非UTF-8的字节序列，客户端解析时自然会抛出“invalid UTF-8 sequence”错误。

另外，你分批发送压缩数据时，每一块都设置了FIN=1（帧首字节的最高位是1），这会让客户端认为每一个chunk都是独立的完整消息，而不是一个大消息的分片，客户端无法将多个chunk拼接后解压，也会导致解析失败。

2. 正确的WebSocket消息帧格式 #

WebSocket帧的字节结构如下：

• 第1字节：
  • 最高位FIN：1表示这是消息的最后一帧，0表示后续还有分片
  • 接下来3位RSV1/RSV2/RSV3：用于扩展（比如启用压缩时需设置RSV1=1）
  • 最后4位opcode：0x1=文本帧，0x2=二进制帧，0x0=续帧，0x8=关闭帧等
• 第2字节：
  • 最高位MASK：服务器发送给客户端时必须设为0（客户端发送给服务器必须设为1）
  • 最后7位Payload Length：0-125直接表示长度；126表示后续2字节是16位长度；127表示后续8字节是64位长度
• 扩展长度字段：根据Payload Length的值决定是否存在（126对应2字节，127对应8字节）
• 掩码字段：仅当MASK=1时存在（服务器发送不需要）
• 载荷数据：文本帧为UTF-8字节，二进制帧为任意字节流

针对你的压缩场景，正确处理逻辑：

• 使用**二进制帧（opcode=0x2）**发送压缩数据
• 分片发送时：第一帧设FIN=0+opcode=0x2，中间分片设FIN=0+opcode=0x0，最后一帧设FIN=1+opcode=0x0
• 若要使用标准压缩，需在握手阶段协商Sec-WebSocket-Extensions: permessage-deflate头，客户端和服务器都支持后才能发送压缩帧

3. 更高效的实现方式 #

结合嵌入式环境和MIDI SysEx数据特性，优化方向如下：

（1）改用二进制帧传输原生数据 #

MIDI数据是二进制格式，无需转成JSON字符串，直接将整数数组转成字节流发送，节省序列化/反序列化开销：

# 示例：将MIDI SysEx整数数组转为字节
sysex_data = [0xF0, 0x41, 0x10, 0x42, 0x12, 0x40, 0x00, 0x7F, 0x00, 0x41, 0xF7]
byte_data = bytes(sysex_data)
# 发送时使用二进制帧格式

（2）修复分片与压缩逻辑 #

如果必须压缩，先在握手阶段协商permessage-deflate扩展（修改握手代码，解析客户端的Sec-WebSocket-Extensions头并返回对应响应），然后：

• 对完整的二进制消息进行zlib压缩
• 分片发送时，仅第一帧用opcode=0x2，后续分片用opcode=0x0，最后一帧设置FIN=1
• 若不协商扩展，可让客户端接收二进制数据后自行解压，避免帧格式错误

（3）完善消息接收逻辑 #

当前receive_message仅接收一次1024字节，无法处理超过1024字节的完整消息，需循环接收直到获取全部载荷：

def receive_message(self, connection):
    try:
        data = connection.recv(1024)
        if not data:
            return None

        byte1, byte2 = struct.unpack('BB', data[:2])
        fin = byte1 & 0b10000000
        opcode = byte1 & 0b00001111
        masked = byte2 & 0b10000000
        payload_length = byte2 & 0b01111111

        if masked != 0b10000000:
            self.logger.error('Client data must be masked')
            return None

        # 处理扩展长度
        offset = 2
        if payload_length == 126:
            payload_length = int.from_bytes(data[offset:offset+2], byteorder='big')
            offset +=2
        elif payload_length == 127:
            payload_length = int.from_bytes(data[offset:offset+8], byteorder='big')
            offset +=8

        # 获取掩码
        masking_key = data[offset:offset+4]
        offset +=4

        # 循环接收所有载荷数据
        payload_data = bytearray()
        remaining = payload_length
        while remaining >0:
            chunk = connection.recv(min(remaining, 1024))
            if not chunk:
                return None
            payload_data.extend(chunk)
            remaining -= len(chunk)

        # 解码掩码
        decoded_bytes = bytearray()
        for i in range(payload_length):
            decoded_bytes.append(payload_data[i] ^ masking_key[i %4])

        if opcode ==0x1:
            return decoded_bytes.decode('utf-8')
        elif opcode ==0x2:
            return decoded_bytes  # 返回二进制数据
        elif opcode ==0x8:
            self.logger.info('Connection closed by client')
            return None
        elif opcode ==0x0:
            self.logger.warning('Continuation frames not implemented')
            return None
        else:
            self.logger.error(f'Unsupported frame type: {opcode}')
            return None
    except Exception as e:
        self.logger.error(f'Error receiving message: {e}')
        return None

（4）修复全局连接变量问题 #

当前代码中self.connection是全局变量，多客户端连接时会被覆盖，导致后续消息发送错误，应让每个handle_connection线程独立维护自身的连接对象。

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