Python asyncio streams：无stop byte的JSON接收优雅方案求问

算力阿峰

2026-8-19

解决无终止符JSON字节流的优雅方案

基于JSON结构的括号计数法：JSON顶级结构只能是{}或[]，可以维护一个括号计数器——读取字节时，遇到{/[就加1，遇到}/]就减1，当计数器归零时，即可判定已获取完整的JSON包。这种方式比全量校验JSON轻量得多，能精准定位包边界。
推动服务端规范传输格式：如果有沟通渠道，优先建议服务端在每个JSON包末尾添加固定分隔符（比如换行符\n或自定义特殊字节序列），这是从根源解决问题最省心的方式，后续维护成本极低。
使用流式JSON解析库：多数编程语言都有原生或第三方的流式JSON解析工具，无需自己处理缓冲区和校验逻辑：
- Python可以用ijson，流式读取完整JSON对象
- Java用Jackson Streaming API，逐事件解析流数据
- Go的标准库encoding/json自带的Decoder，可直接从TCP流中读取完整JSON结构

举两个简单示例：
Python示例：

import ijson
import socket

# 假设已建立socket连接
stream = socket_conn.makefile('rb')
# 流式读取完整JSON对象
for json_obj in ijson.items(stream, ''):
    # 处理获取到的完整JSON
    print(json_obj)

Go示例：

package main

import (
    "encoding/json"
    "net"
)

func main() {
    conn, _ := net.Dial("tcp", "server_addr:port")
    defer conn.Close()

    decoder := json.NewDecoder(conn)
    var data interface{}
    // Decoder自动处理流中JSON的边界判断
    for decoder.Decode(&data) == nil {
        // 处理完整JSON数据
        println(data)
    }
}