如何在Python中按1MB批次发送大型JSON对象？

没问题，我来帮你实现大型JSON按1MB批次发送的需求！先看看你的现有代码，再一步步给你解决思路和完整代码～

首先提个小优化：你的现有代码里如果value是Python字典（JSON对象），直接用data=value会把它编码成表单数据，而非JSON格式。正确的做法要么用json=value参数，要么先把对象序列化成JSON字符串再传给data，同时保持Content-Type头为application/json。

接下来分两种常见场景实现分批发送：

场景1：JSON是大型数组（最易处理的情况） #

如果你的value是包含大量元素的数组，我们可以按元素分组，每组序列化后的大小尽量接近1MB，然后分批发送每个合法的子数组。这样接收端可以直接解析每一批的JSON，不需要额外拼接。

代码实现： #

import json
import requests

URL_ATTRIBUTE = "你的目标接口地址"
CHUNK_SIZE = 1024 * 1024  # 1MB，单位是字节

# 注意：Python3用items()替代Python2的iteritems()
for key, value in attributeJs.items():
    if isinstance(value, list):
        current_chunk = []
        current_byte_size = 0
        
        for item in value:
            # 序列化单个元素，计算它的utf-8编码字节大小
            item_json = json.dumps(item)
            item_size = len(item_json.encode('utf-8'))
            
            # 若加入当前元素后超过1MB，就发送当前批次
            if current_byte_size + item_size > CHUNK_SIZE and current_chunk:
                chunk_json = json.dumps(current_chunk)
                headers = {'Content-Type': 'application/json'}
                
                try:
                    requests.packages.urllib3.disable_warnings()
                    response = requests.post(
                        url=URL_ATTRIBUTE,
                        headers=headers,
                        verify=False,
                        data=chunk_json,
                        timeout=(15, 20)
                    )
                    response.raise_for_status()  # 主动抛出HTTP错误
                    print(f"批次发送成功，状态码：{response.status_code}")
                    # 重置当前批次
                    current_chunk = []
                    current_byte_size = 0
                except requests.exceptions.RequestException as e:
                    print(f"发送批次失败：{str(e)}")
                    # 这里可添加重试逻辑或错误告警
            
            # 将当前元素加入批次
            current_chunk.append(item)
            current_byte_size += item_size
        
        # 发送最后一批剩余元素
        if current_chunk:
            chunk_json = json.dumps(current_chunk)
            headers = {'Content-Type': 'application/json'}
            
            try:
                requests.packages.urllib3.disable_warnings()
                response = requests.post(
                    url=URL_ATTRIBUTE,
                    headers=headers,
                    verify=False,
                    data=chunk_json,
                    timeout=(15, 20)
                )
                response.raise_for_status()
                print(f"最后一批发送成功，状态码：{response.status_code}")
            except requests.exceptions.RequestException as e:
                print(f"发送最后一批失败：{str(e)}")

场景2：JSON是单个超大对象 #

如果value是无法拆分的单个超大JSON对象，我们可以用HTTP的分块传输编码（Transfer-Encoding: chunked），把整个JSON拆成1MB的字节块，让requests自动分块发送。这种方式需要接收端支持分块传输（大部分现代服务器都支持）。

代码实现： #

import json
import requests

URL_ATTRIBUTE = "你的目标接口地址"
CHUNK_SIZE = 1024 * 1024  # 1MB，单位是字节

for key, value in attributeJs.items():
    # 处理单个大对象（非数组）
    if not isinstance(value, list):
        # 先序列化整个对象为JSON字符串，再转成字节
        full_json_str = json.dumps(value)
        full_json_bytes = full_json_str.encode('utf-8')
        
        # 拆分字节成1MB的块，生成迭代器
        chunk_iterator = (full_json_bytes[i:i+CHUNK_SIZE] for i in range(0, len(full_json_bytes), CHUNK_SIZE))
        
        headers = {
            'Content-Type': 'application/json',
            'Transfer-Encoding': 'chunked'  # 启用分块传输
        }
        
        try:
            requests.packages.urllib3.disable_warnings()
            response = requests.post(
                url=URL_ATTRIBUTE,
                headers=headers,
                verify=False,
                data=chunk_iterator,  # 传入迭代器实现分块发送
                timeout=(15, 20)
            )
            response.raise_for_status()
            print(f"大对象分块发送成功，状态码：{response.status_code}")
        except requests.exceptions.RequestException as e:
            print(f"发送大对象失败：{str(e)}")

关键注意点 #

• 编码问题：计算字节大小时必须用utf-8编码，因为JSON默认编码为utf-8，避免大小计算错误。
• 错误处理：代码中加入了response.raise_for_status()，会在HTTP状态码非2xx时抛出异常，方便排查问题。
• 兼容性：把iteritems()改成了items()以适配Python3；若仍用Python2，改回iteritems()即可。

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