如何通过subprocess/stdout加速JSON字符串加载为Python字典

更高效率的实现方式 #

针对你当前的测试场景，可通过以下手段进一步提升加载速度：

• 替换JSON解析库：标准库json在大体积JSON解析场景下性能偏低，可替换为ujson，pypy环境兼容度良好，解析速度可提升30%~50%。
• 优化stdout传输逻辑：原有print+readlines的方案存在文本编码、换行符匹配两层额外开销，可改为子进程先输出4字节的JSON长度前缀，再输出JSON二进制流，主进程按固定长度读取数据，避免额外开销。
• 改用内存文件系统存储临时文件：Linux环境下可将临时文件存储路径指定为/dev/shm，完全规避磁盘IO开销，速度比磁盘临时文件高40%以上。
• 替换序列化协议：如果业务允许不用JSON序列化，可改用msgpack等二进制序列化协议，序列化后体积比JSON小30%左右，解析速度也更快。

优化后stdout直传示例代码 #

子进程代码：

import json
import sys
import struct

JSON_SIZE = 250000
NUM_JSON = 20

d = [{'a': x, 'b': x, 'c': x, 'd': x, 'e': x, 'f': x} for x in range(JSON_SIZE)]
d_bytes = json.dumps(d).encode('utf-8')
len_bytes = struct.pack('I', len(d_bytes))

for _ in range(NUM_JSON):
    sys.stdout.buffer.write(len_bytes)
    sys.stdout.buffer.write(d_bytes)
sys.stdout.buffer.flush()

主进程代码：

import subprocess
import struct
import ujson

proc = subprocess.Popen(['pypy3', 'mysubprocess.py'],
                        stdout=subprocess.PIPE,
                        stderr=subprocess.PIPE)

while True:
    len_bytes = proc.stdout.read(4)
    if not len_bytes:
        break
    data_len = struct.unpack('I', len_bytes)[0]
    d_bytes = proc.stdout.read(data_len)
    d = ujson.loads(d_bytes)

该方案在相同测试环境下耗时可降低到3s以内。

无文件mmap共享内存实现方案 #

可以直接通过POSIX共享内存实现无文件的mmap内存共享，完全不需要写入磁盘，比原有临时文件mmap方案再快15%左右：

• 子进程通过shm_open创建共享内存对象，将JSON字节直接写入共享内存区，将共享内存名称和数据长度传递给主进程
• 主进程直接mmap对应的共享内存对象读取数据，使用完成后销毁共享内存即可

POSIX共享内存示例代码 #

需要先安装posix-ipc依赖：pypy3 -m pip install posix-ipc，仅支持Linux/macOS环境。
子进程代码：

import json
import os
import sys
from posix_ipc import SharedMemory, O_CREAT

JSON_SIZE = 250000
NUM_JSON = 20

d = [{'a': x, 'b': x, 'c': x, 'd': x, 'e': x, 'f': x} for x in range(JSON_SIZE)]
d_bytes = json.dumps(d).encode('utf-8')
data_len = len(d_bytes)

for i in range(NUM_JSON):
    shm_name = f"/shm_json_{os.getpid()}_{i}"
    shm = SharedMemory(shm_name, O_CREAT, size=data_len)
    os.write(shm.fd, d_bytes)
    shm.close_fd()
    print(f"{shm_name}\t{data_len}")
sys.stdout.flush()

主进程代码：

import subprocess
import mmap
import ujson
from posix_ipc import SharedMemory, unlink_shared_memory

proc = subprocess.Popen(['pypy3', 'mysubprocess.py'],
                        stdout=subprocess.PIPE,
                        stderr=subprocess.PIPE)

for line in proc.stdout.readlines():
    line = line.strip().decode('utf-8')
    shm_name, data_len = line.split('\t')
    data_len = int(data_len)
    shm = SharedMemory(shm_name)
    with mmap.mmap(shm.fd, length=data_len, access=mmap.ACCESS_READ) as mmap_obj:
        d = ujson.loads(mmap_obj.read())
    shm.close_fd()
    unlink_shared_memory(shm_name)

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