如何在while True循环中每隔X分钟写入CSV且不阻塞实时数据

你遇到的核心问题是要在不阻塞实时数据流程的前提下，定时执行CSV写入操作，以下是几种实用的实现方案：

方案1：记录时间差，循环内触发写入 #

不需要额外依赖或线程，通过记录上次写入的时间，每次循环检查当前时间与上次写入时间的差值，达到5分钟时再执行写入，完全不影响实时数据的采集和推送。

修改后的完整代码示例：

import csv
import time
from datetime import datetime, timedelta

# 初始化变量
last_write_time = datetime.now()
file_exists = False
# 设置5分钟的写入间隔
WRITE_INTERVAL = timedelta(minutes=5)

while True:
    # 1. 实时采集传感器数据，给仪表盘提供数据
    get_temperature(args.factor)
    get_pressure()
    get_humidity()
    get_light()
    if not args.enviro:
        get_gas()
        get_particulates()
    
    current_data = collect_all_data()
    if DEBUG:
        logging.info('Sensor data: {}'.format(current_data))
    
    # 2. 检查是否到达写入时间
    current_time = datetime.now()
    if current_time - last_write_time >= WRITE_INTERVAL:
        with open('enviroplus.csv', 'a', newline='') as csvfile:
            writer = csv.DictWriter(
                csvfile,
                delimiter=',',
                quotechar='|',
                quoting=csv.QUOTE_MINIMAL,
                fieldnames=['Date/Time','temperature','humidity','pressure','oxidising','reducing','nh3','lux','proximity','pm1','pm25','pm10']
            )
            if not file_exists:
                writer.writeheader()
                file_exists = True
            # 写入当前数据（如果需要累积5分钟内的所有数据，可以维护一个列表存储）
            writer.writerow({
                'Date/Time': current_time.strftime('%Y-%m-%d %H:%M:%S'),
                **current_data
            })
        # 更新上次写入时间
        last_write_time = current_time
    
    # 可选：添加极短的sleep，避免循环占用过多CPU资源，不影响实时性
    time.sleep(0.1)

如果需要保存5分钟内的所有数据再批量写入，可以新增一个data_buffer = []列表，每次循环把current_data加入列表，到时间后用writer.writerows(data_buffer)写入，再清空列表即可。

方案2：多线程分离任务 #

把实时数据采集和CSV写入拆分成两个独立线程，主线程专注于实时数据处理，写入线程负责定时执行CSV操作，彻底避免互相干扰。

代码示例：

import csv
import threading
import time
from datetime import datetime, timedelta
from queue import Queue

# 用于传递数据的队列
data_queue = Queue()
file_exists = False

# CSV写入线程函数
def csv_writer():
    global file_exists
    WRITE_INTERVAL = 5 * 60  # 5分钟转为秒数
    while True:
        # 等待5分钟
        time.sleep(WRITE_INTERVAL)
        # 取出队列中所有累积的数据
        data_list = []
        while not data_queue.empty():
            data_list.append(data_queue.get())
        if not data_list:
            continue
        # 写入CSV
        with open('enviroplus.csv', 'a', newline='') as csvfile:
            writer = csv.DictWriter(
                csvfile,
                delimiter=',',
                quotechar='|',
                quoting=csv.QUOTE_MINIMAL,
                fieldnames=['Date/Time','temperature','humidity','pressure','oxidising','reducing','nh3','lux','proximity','pm1','pm25','pm10']
            )
            if not file_exists:
                writer.writeheader()
                file_exists = True
            writer.writerows(data_list)

# 启动写入线程（设置为守护线程，随主线程退出）
threading.Thread(target=csv_writer, daemon=True).start()

# 主线程：实时采集并推送数据
while True:
    get_temperature(args.factor)
    get_pressure()
    get_humidity()
    get_light()
    if not args.enviro:
        get_gas()
        get_particulates()
    
    current_data = collect_all_data()
    current_data['Date/Time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    # 将数据放入队列
    data_queue.put(current_data)
    
    if DEBUG:
        logging.info('Sensor data: {}'.format(current_data))
    
    time.sleep(0.1)

这个方案的优势是任务职责清晰，即使CSV写入偶尔出现延迟，也不会影响实时仪表盘的数据推送。

方案3：使用第三方定时库schedule #

如果不想手动处理时间逻辑，可以用schedule库简化定时任务的实现，代码更简洁易读。

首先安装依赖：

pip install schedule

代码示例：

import csv
import schedule
import time
from datetime import datetime
from threading import Thread

# 存储待写入的数据
data_buffer = []
file_exists = False

def write_csv_task():
    global file_exists
    if not data_buffer:
        return
    with open('enviroplus.csv', 'a', newline='') as csvfile:
        writer = csv.DictWriter(
            csvfile,
            delimiter=',',
            quotechar='|',
            quoting=csv.QUOTE_MINIMAL,
            fieldnames=['Date/Time','temperature','humidity','pressure','oxidising','reducing','nh3','lux','proximity','pm1','pm25','pm10']
        )
        if not file_exists:
            writer.writeheader()
            file_exists = True
        writer.writerows(data_buffer)
    # 清空缓冲区
    data_buffer.clear()

# 设置定时任务：每5分钟执行一次
schedule.every(5).minutes.do(write_csv_task)

# 启动定时任务线程
def run_scheduler():
    while True:
        schedule.run_pending()
        time.sleep(1)

Thread(target=run_scheduler, daemon=True).start()

# 主线程采集数据
while True:
    get_temperature(args.factor)
    get_pressure()
    get_humidity()
    get_light()
    if not args.enviro:
        get_gas()
        get_particulates()
    
    current_data = collect_all_data()
    current_data['Date/Time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    data_buffer.append(current_data)
    
    if DEBUG:
        logging.info('Sensor data: {}'.format(current_data))
    
    time.sleep(0.1)

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