Python多线程车载CAN总线数据记录仪定时存储漂移问题

看起来你的核心问题是定时存储的间隔随着运行时间逐渐变大，这在Linux非实时系统里确实容易出现，但你的实现方式有几个关键优化点，尤其是文件IO的处理逻辑，这才是导致漂移越来越严重的主要原因。

问题根源分析 #

先拆解你代码里的几个核心问题：

1. 文件IO耗时随数据量增长持续上升
   你每次存储都要把整个JSON文件加载到内存，append新数据后再完整写回。当文件里的记录越来越多（比如运行几小时后），json.load()和json.dump()的耗时会显著增加，每次循环的总时间直接超过100ms。而你的定时器只是在上一次的next_call基础上加间隔，这就导致延迟不断累加，最终间隔漂移到几分钟。

2. 定时器逻辑的小缺陷

   • once = bool 是错误的初始化方式，应该是 once = False，这个小错误可能导致第一次进入存储逻辑时的时间计算异常。
   • 当某次IO耗时超过100ms时，next_call - time.time()会变成负数，你用abs()处理后会sleep多余的时间，反而加剧了漂移。

3. 并发模型没解决核心问题
   你试过线程和进程，但问题的根源不在并发模型，而在于存储操作的耗时不可控且持续增加，换线程/进程无法抵消IO带来的延迟累加。

解决方案 #

1. 优化文件存储方式：避免全量读写JSON #

最有效的改进是改用JSON Lines格式（每行一个JSON对象），彻底避免每次加载整个文件。这种格式不仅写入效率高，后续上传Blob Storage时也可以直接按行处理，或者最后包装成数组。

修改后的存储逻辑示例：

def create_json_file():
    global FILE_INITIALIZED
    global FILE_NAME
    once = False  # 修正初始化错误
    next_run = time.time()  # 初始化绝对时间点
    
    try:
        while not stop_create.is_set():
            if not store_data:
                next_run = time.time() + INTERVAL_TIME
                time.sleep(max(0, next_run - time.time()))
                continue
            
            if not once:
                next_run = time.time()
                once = True
            
            # 计算UTC时间戳（保留你的原有逻辑）
            utc_offset_sec = time.altzone if time.localtime().tm_isdst else time.timezone
            utc_offset = datetime.timedelta(seconds=-utc_offset_sec)
            timestamp = datetime.datetime.now().replace(tzinfo=datetime.timezone(offset=utc_offset)).isoformat()
            
            tmp_json_dict = {
                'deviceId': DEVICE_ID,
                'timestamp': timestamp,
                **processed_data  # 合并字典更简洁
            }
            
            if not FILE_INITIALIZED:
                FILE_NAME = f'./storage/{DEVICE_ID}_{datetime.datetime.now().strftime("%Y%m%d_%H%M")}.json'
                # 创建文件并写入第一行
                with open(FILE_NAME, 'w') as outputfile:
                    json.dump(tmp_json_dict, outputfile)
                    outputfile.write('\n')  # 换行作为分隔
                FILE_INITIALIZED = True
            else:
                # 追加模式写入单行JSON
                with open(FILE_NAME, 'a') as outputfile:
                    json.dump(tmp_json_dict, outputfile)
                    outputfile.write('\n')
            
            # 更新下一次运行时间，处理超时情况
            next_run += INTERVAL_TIME
            sleep_time = next_run - time.time()
            if sleep_time > 0:
                time.sleep(sleep_time)
            else:
                # 如果本次执行超时，直接重置下一次时间，避免延迟累加
                logging.warning(f"Storage task delayed by {-sleep_time:.2f}s")
                next_run = time.time() + INTERVAL_TIME
                
    except Exception as err:
        logging.exception(err)
        if not stop_create.is_set():
            global thread_exception_file
            thread_exception_file = sys.exc_info()

2. 进一步优化：批量缓存数据 #

如果100ms的间隔还是太频繁，可以考虑缓存N条数据后再批量写入，比如每10条（1秒）写入一次，进一步减少IO次数：

# 新增缓存列表
data_cache = []
CACHE_SIZE = 10  # 缓存10条后写入

# 在存储逻辑里替换原写入代码：
data_cache.append(tmp_json_dict)
if len(data_cache) >= CACHE_SIZE or not FILE_INITIALIZED:
    # 写入缓存的数据
    with open(FILE_NAME, 'a' if FILE_INITIALIZED else 'w') as f:
        for item in data_cache:
            json.dump(item, f)
            f.write('\n')
    data_cache.clear()
    FILE_INITIALIZED = True

3. 可选：改用SQLite存储 #

如果JSON文件的性能还是不够，可以考虑用SQLite数据库存储数据，它的写入性能更稳定，尤其是数据量大的时候，而且查询和导出也更方便。比如用sqlite3模块创建表，每次插入一条记录，最后导出为JSON上传。

关键注意事项 #

• 绝对时间调度：始终基于next_run += INTERVAL_TIME计算下一次执行时间，而不是每次用当前时间加间隔，这样可以抵消单次小延迟；如果某次延迟过大，手动重置next_run为当前时间+间隔，避免延迟累加。
• 分离采集与存储线程：把CAN数据采集和存储分成两个独立线程，用队列传递数据，这样存储的IO耗时不会影响数据采集的实时性。

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