如何基于配置文件动态定义多进程循环实现传感器不同频率并行采样

测试代码运行异常的原因 #

你写的测试代码只有第一个循环运行，核心问题是构造Process对象时，错误地直接执行了目标函数，而不是把函数作为参数传给Process：
当你写Process(target=loop_a(5,"foo"))时，loop_a(5,"foo")会立刻执行进入死循环，代码根本走不到第二个Process的创建步骤，自然只有foo的输出。

修正后的测试代码如下：

from multiprocessing import Process
import time

def loop_a(sleeptime, string):
  while 1:
    print(string)
    time.sleep(sleeptime)

# 正确写法：target传函数本身，参数通过args元组传递
Process(target=loop_a, args=(5, "foo")).start()
Process(target=loop_a, args=(2, "bar")).start()

动态生成传感器采样循环的实现 #

你不需要提前为每个传感器定义独立的循环函数，只需要写一个通用的采样逻辑，遍历传感器配置列表批量创建进程即可：

1. 首先给你的传感器配置JSON新增sample_delay字段，配置每个传感器的采样间隔（单位：秒）
2. 完整实现代码如下：

import json
import time
from multiprocessing import Process

# 通用传感器采样循环
def sensor_worker(model, address, sample_delay):
    while True:
        H, T = read_sensor(model, address)
        send_data(H, T)
        time.sleep(sample_delay)

if __name__ == "__main__":
    # 读取传感器配置
    with open(inputfilelocation, 'r') as f:
        sensorlist = json.load(f)
    
    process_list = []
    # 动态为每个传感器创建独立进程
    for sensor in sensorlist:
        p = Process(
            target=sensor_worker,
            args=(
                sensor['model'],
                sensor['address'],
                sensor['sample_delay']
            )
        )
        p.start()
        process_list.append(p)
    
    # 阻塞主程序，防止主程序退出导致子进程被终止
    for p in process_list:
        p.join()

补充说明 #

• 如果传感器数量非常多（几十上百个），多进程的资源开销会比较大，可以换成threading.Thread替换Process，IO密集型场景下线程足够用，API和Process完全一致，仅需要修改导入语句和类名即可。
• 如果需要支持运行中动态新增/删除传感器，可以额外加一个管理进程维护传感器列表，按需启停worker，不需要重启整个程序。

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