Python线程未并发执行问题：传感器多类型测量异常

核心问题修复：线程创建方式错误 #

你当前的线程代码直接调用了take_measurements函数，而非将其作为线程目标传递。这导致函数会在主线程中同步执行，完全没有利用多线程的并发能力。比如：

Thread(target=take_measurements('hourly', 60)).start()

这段代码会先执行take_measurements('hourly', 60)，等它完全跑完后才会创建线程（此时函数已执行完毕，线程无实际工作）。这就是你看到同一时间只有一个"线程"运行的原因——本质上都是主线程在依次执行这些函数。

正确写法是传递函数对象，通过args参数传递函数所需参数：

Thread(target=take_measurements, args=('hourly', 60)).start()

其他关键问题与修复 #

1. 线程安全：传感器访问需加锁
   多线程同时调用sensor.query()会引发竞争条件，导致数据读取异常或传感器响应混乱。需要用threading.Lock保护传感器访问：

import threading

sensor_lock = threading.Lock()

def take_measurements(measurement_type, num_iterations):
    # ... 其他代码 ...
    for i in range(num_iterations):
        with sensor_lock:
            pm2.5, pm10 = sensor.query()
        # ... 写入文件 ...

2. CSV写入参数错误
   csv.writer.writerow()仅接受一个可迭代参数（如列表或元组），你的代码会抛出参数数量错误，需修改为：

writer.writerow([pm2.5, pm10, curr_time()])

3. 文件覆盖与无效对象传递

• 用'w'模式打开文件会覆盖原有内容，建议改用'a'追加模式，或为每个测量任务生成带时间戳的文件名（避免数据丢失）。
• file.close()后文件对象失效，upload_data(file, measurement_type)无法正常读取内容，应传递文件名而非文件对象：

  def take_measurements(measurement_type, num_iterations):
      filename = f"{measurement_type}_{curr_time().replace(':', '-')}.csv"
      with open(filename, 'w', newline='') as file:
          writer = csv.writer(file)
          writer.writerow(['PM2.5', 'PM10', 'Timestamp'])  # 写入表头
          for i in range(num_iterations):
              with sensor_lock:
                  pm2.5, pm10 = sensor.query()
              writer.writerow([pm2.5, pm10, curr_time()])
              time.sleep(60)
      upload_data(filename, measurement_type)
  

使用with语句可自动关闭文件，无需手动调用file.close()。

4. 无限创建线程导致资源耗尽
   你的while True循环会持续创建新测量线程，短时间内就会耗尽系统资源。每个测量类型（小时/日/周/月）只需启动一次线程：

if __name__ == '__main__':
    sensor = SDS011("/dev/ttyUSB0")
    sensor.sleep(sleep=False)
    print("Preparing sensor...")
    time.sleep(15)
    print("Sensor is now running:")

    # 启动各测量线程，每个类型仅启动一次
    threads = []
    threads.append(Thread(target=take_measurements, args=('hourly', 60)))
    threads.append(Thread(target=take_measurements, args=('daily', 1440)))
    threads.append(Thread(target=take_measurements, args=('weekly', 10080)))
    threads.append(Thread(target=take_measurements, args=('monthly', 43800)))

    for t in threads:
        t.start()

    try:
        for t in threads:
            t.join()
    except KeyboardInterrupt:
        clean_exit()

完整修复后的示例代码 #

import csv
import time
import threading
from sds011 import SDS011  # 假设使用sds011官方库

sensor_lock = threading.Lock()

def curr_time():
    return time.strftime("%Y-%m-%d %H:%M:%S")

def upload_data(filename, measurement_type):
    # 实现数据上传逻辑，通过文件名读取内容
    with open(filename, 'r') as f:
        data = f.read()
    print(f"Uploading {measurement_type} data from {filename}")
    # ... 上传代码 ...

def take_measurements(measurement_type, num_iterations):
    filename = f"{measurement_type}_{curr_time().replace(':', '-')}.csv"
    with open(filename, 'w', newline='') as file:
        writer = csv.writer(file)
        writer.writerow(['PM2.5', 'PM10', 'Timestamp'])
        for i in range(num_iterations):
            with sensor_lock:
                pm2.5, pm10 = sensor.query()
            writer.writerow([pm2.5, pm10, curr_time()])
            time.sleep(60)
    upload_data(filename, measurement_type)

if __name__ == '__main__':
    sensor = SDS011("/dev/ttyUSB0")
    sensor.sleep(sleep=False)
    print("Preparing sensor...")
    time.sleep(15)
    print("Sensor is now running:")

    threads = [
        threading.Thread(target=take_measurements, args=('hourly', 60)),
        threading.Thread(target=take_measurements, args=('daily', 1440)),
        threading.Thread(target=take_measurements, args=('weekly', 10080)),
        threading.Thread(target=take_measurements, args=('monthly', 43800))
    ]

    for t in threads:
        t.start()

    try:
        for t in threads:
            t.join()
    except KeyboardInterrupt:
        sensor.sleep(sleep=True)
        print("Sensor stopped, exiting...")

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