Python中如何高效处理字节流？
语言：python
问题：通过 socket 收到 b'\x7d\x76\x76\x76\x76\x76\x76\x76\x[XOR]\x7d'
我该如何去处理这串 bytes。
1 ）在 recv 中找到 0x7e 2 ）校验 3 ）进一步处理
python 不像 c 那么好处理字节流。 求大神赐教
Python中如何高效处理字节流？
12 回复
import io
import struct
from typing import Union, Optional
class ByteStreamProcessor:
"""高效字节流处理工具类"""
def __init__(self, data: Union[bytes, bytearray, io.BytesIO]):
"""
初始化字节流处理器
Args:
data: 字节数据，可以是bytes、bytearray或BytesIO对象
"""
if isinstance(data, (bytes, bytearray)):
self._stream = io.BytesIO(data)
elif isinstance(data, io.BytesIO):
self._stream = data
else:
raise TypeError("数据必须是bytes、bytearray或BytesIO类型")
def read_int(self, byteorder: str = 'little', signed: bool = False) -> Optional[int]:
"""读取整数"""
try:
size = {1: 'B', 2: 'H', 4: 'I', 8: 'Q'}[self._stream.getbuffer().nbytes - self._stream.tell()]
fmt = f'{">" if byteorder == "big" else "<"}{size}'
data = self._stream.read(struct.calcsize(fmt))
return struct.unpack(fmt, data)[0] if data else None
except (KeyError, struct.error):
return None
def read_bytes(self, length: int) -> Optional[bytes]:
"""读取指定长度的字节"""
data = self._stream.read(length)
return data if data else None
def read_string(self, encoding: str = 'utf-8') -> Optional[str]:
"""读取字符串（以null结尾）"""
chars = []
while True:
char = self._stream.read(1)
if not char or char == b'\x00':
break
chars.append(char)
return b''.join(chars).decode(encoding) if chars else None
def write_int(self, value: int, size: int = 4, byteorder: str = 'little') -> None:
"""写入整数"""
fmt = f'{">" if byteorder == "big" else "<"}{{1: "B", 2: "H", 4: "I", 8: "Q"}[size]}'
self._stream.write(struct.pack(fmt, value))
def write_bytes(self, data: bytes) -> None:
"""写入字节数据"""
self._stream.write(data)
def write_string(self, text: str, encoding: str = 'utf-8', null_terminated: bool = True) -> None:
"""写入字符串"""
encoded = text.encode(encoding)
self._stream.write(encoded)
if null_terminated:
self._stream.write(b'\x00')
def seek(self, position: int) -> None:
"""移动流位置"""
self._stream.seek(position)
def tell(self) -> int:
"""获取当前流位置"""
return self._stream.tell()
def getvalue(self) -> bytes:
"""获取所有字节数据"""
return self._stream.getvalue()
@property
def remaining(self) -> int:
"""获取剩余字节数"""
current = self._stream.tell()
self._stream.seek(0, io.SEEK_END)
end = self._stream.tell()
self._stream.seek(current)
return end - current
# 使用示例
def process_network_packet(packet_data: bytes):
"""处理网络数据包示例"""
processor = ByteStreamProcessor(packet_data)
# 读取协议头
packet_id = processor.read_int(byteorder='big', size=2)
packet_length = processor.read_int(size=4)
# 读取有效载荷
if packet_length > 0:
payload = processor.read_bytes(packet_length)
if payload:
# 解析payload中的多个字段
payload_processor = ByteStreamProcessor(payload)
field1 = payload_processor.read_int(size=2)
field2 = payload_processor.read_string()
print(f"Packet ID: {packet_id}")
print(f"Field1: {field1}, Field2: {field2}")
return processor.getvalue()
# 创建测试数据
test_data = b'\x00\x01' + struct.pack('>I', 10) + b'Hello\x00World\x00'
result = process_network_packet(test_data)
# 构建数据包示例
def build_packet():
processor = ByteStreamProcessor(bytearray())
processor.write_int(1, size=2, byteorder='big')
processor.write_int(20, size=4)
processor.write_string("Hello")
processor.write_string("World")
return processor.getvalue()
built_packet = build_packet()
print(f"Built packet: {built_packet}")
这个方案用BytesIO做内存流，struct处理二进制数据，封装了常用操作。
没看懂你要做什么，排版看得头疼。。。。。如果收到的数据有固定的结构或者头部，用 struct 根据 field 来拆。如果没有固定的格式，依赖特殊的 magic number，那么无论是 bytes 还是 bytearray 都是可以 find 或者 index 的。
一句话来说，用 Python 处理数据绝对比用 C 简单啊
