无显式分隔符时如何正确拆分Socket缓冲JSON日志数据

我通过Socket接收由无固定长度、无显式分隔符的JSON日志组成的数据，部分日志存在嵌套JSON结构，无法直接用}作为分隔符拆分。此前采用Python的partition()方法，以b"'}"作为分隔符从缓冲中提取完整JSON对象，该方法仅在日志最后字段为字符串时有效，当最后字段为整数时无法正确拆分，求可行解决方案。

现有Buffer类代码：

class Buffer:

    def __init__(self,sock):
        self.sock = sock
        self.buffer = b''

    def get_line(self):
        while b"'}" not in self.buffer:
            log_b = self.sock.recv(8192)
            if not log_b:
                return None
            self.buffer += log_b

        log_p, sep, self.buffer = self.buffer.partition(b"'}")
        log_p = log_p + sep
        return log_p

解决方案 #

核心思路是利用JSON解析器的增量解析能力，通过识别完整的JSON语法结构来拆分日志，而非依赖固定分隔符。这种方法能自动处理嵌套大括号、不同类型结尾的日志，彻底解决拆分异常问题。

改进后的Buffer类实现 #

由于日志采用Python字典风格的单引号格式（非标准JSON），我们先将其转换为标准JSON格式，再使用json.JSONDecoder.raw_decode()方法实现增量解析——该方法会返回第一个完整JSON对象的结束位置，据此可准确拆分缓冲数据：

import json

class Buffer:
    def __init__(self, sock):
        self.sock = sock
        self.buffer = b''
        self.decoder = json.JSONDecoder()

    def get_line(self):
        while True:
            try:
                # 将缓冲字节转为字符串
                data_str = self.buffer.decode('utf-8')
                # 转换为标准JSON格式（替换单引号为双引号）
                json_compatible_str = data_str.replace("'", '"')
                # 增量解析第一个完整JSON对象，获取结束位置
                _, end_pos = self.decoder.raw_decode(json_compatible_str)
                # 提取完整日志的字节数据
                full_log = self.buffer[:end_pos]
                # 更新缓冲，保留未解析的剩余数据
                self.buffer = self.buffer[end_pos:]
                return full_log
            except (json.JSONDecodeError, UnicodeDecodeError):
                # 缓冲数据不完整，继续接收
                chunk = self.sock.recv(8192)
                if not chunk:
                    # 连接关闭，返回None
                    return None
                self.buffer += chunk

方案优势 #

• 兼容嵌套结构：解析器会识别JSON的层级结构，不会误将嵌套的}当作日志结束符
• 适配任意结尾类型：无论日志最后字段是字符串、整数还是其他JSON类型，都能准确识别完整对象
• 鲁棒性强：通过增量解析逐步缓冲数据，确保不会遗漏或截断日志

特殊情况处理 #

如果日志字符串内部包含单引号（如'It\'s a test'），直接替换单引号会破坏字符串结构，此时可改用正则表达式仅替换字符串外部的单引号：

import re

def convert_to_json(s):
    # 匹配不在转义单引号和成对单引号内的单引号
    return re.sub(r"(?<!\\)'(?!(([^']*'){2})*[^']*$)", '"', s)

将上述函数替换原代码中的data_str.replace("'", '"')即可。

内容的提问来源于stack exchange，提问作者m.yagmur