如何优化SCADA传感器API多调用效率？Python提速方案咨询

从SCADA系统通过API采集传感器数据时，受限于API单次仅能获取单个传感器数据，且每次调用耗时3-5秒，成千上万个传感器的串行采集效率极低。现有串行采集脚本如下：

for ninv in sensor_array:       
        base_url = 'https://XXXXXXXXXXXXXXXXX'
        params = {
            'deviceid': ninv,
            'valueids': inv_data,
            'from': date_from,
            'to': date_to,
            'resolution': '15min',
    
        headers = {'Authorization':'Bearer '+token,}
        response = session.get(base_url, headers=headers, params=params)
        df_aux = pd.DataFrame(response.json())
        df_datasensors = pd.concat([df_datasensors, df_aux], axis=0)

核心优化：并发请求 #

网络IO是这类任务的核心瓶颈，使用并发（多进程/多线程/异步）可大幅压缩总耗时。以下是你关注的多进程方案具体实现：

多进程实现步骤 #

1. 封装单传感器采集逻辑：将单个传感器的请求、解析逻辑封装为独立函数，注意每个进程需单独创建HTTP会话（session无法跨进程共享）。
2. 进程池批量执行：通过concurrent.futures.ProcessPoolExecutor管理进程池，控制并发数，批量提交采集任务。
3. 一次性合并结果：收集所有进程返回的DataFrame后统一合并，避免串行pd.concat的性能损耗。

优化后代码示例 #

import pandas as pd
import requests
from concurrent.futures import ProcessPoolExecutor

# 可根据实际情况调整为参数传递
base_url = 'https://XXXXXXXXXXXXXXXXX'
inv_data = "你的valueids参数"
date_from = "起始日期"
date_to = "结束日期"
token = "你的授权令牌"

def fetch_sensor_data(ninv):
    # 每个进程独立创建会话
    session = requests.Session()
    headers = {'Authorization':'Bearer '+token}
    params = {
        'deviceid': ninv,
        'valueids': inv_data,
        'from': date_from,
        'to': date_to,
        'resolution': '15min'
    }
    try:
        response = session.get(base_url, headers=headers, params=params)
        response.raise_for_status()  # 主动抛出HTTP错误
        return pd.DataFrame(response.json())
    except Exception as e:
        print(f"采集传感器{ninv}失败: {str(e)}")
        return pd.DataFrame()  # 返回空DataFrame避免合并报错
    finally:
        session.close()

if __name__ == '__main__':
    sensor_array = ["传感器ID1", "传感器ID2", ...]  # 你的传感器ID列表
    # 并发数需根据API限流调整，建议先从10-15开始测试
    max_workers = 15
    
    with ProcessPoolExecutor(max_workers=max_workers) as executor:
        # 批量提交任务，获取结果迭代器
        results = executor.map(fetch_sensor_data, sensor_array)
    
    # 合并所有采集结果
    df_datasensors = pd.concat(results, axis=0, ignore_index=True)

额外优化建议 #

• 并发数控制：不要盲目调高max_workers，避免触发API限流或给服务器造成过大压力，建议先小规模测试后再调整。
• 错误重试：在采集函数中加入重试逻辑（如使用tenacity库），处理临时网络波动或API返回的5xx错误。
• API批量支持确认：再次检查SCADA系统API是否支持批量传入多个deviceid，若支持则效率远高于并发请求。
• 内存优化：若传感器数量极大，可分批次采集并直接写入文件（如CSV），避免一次性加载所有数据占用过多内存。

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