本文详解如何使用 python 的 multiprocessing 模块，让树莓派同时持续读取分贝传感器数据并响应阈值触发 10 秒音频录制，避免阻塞、确保实时性与线程安全。
本文详解如何使用 python 的 multiprocessing 模块，让树莓派同时持续读取分贝传感器数据并响应阈值触发 10 秒音频录制，避免阻塞、确保实时性与线程安全。
在嵌入式音频监控场景（如树莓派噪声监测系统）中，常需「后台持续采样 + 前端条件触发录音」的并发行为。单线程顺序执行会导致录音期间中断传感器读取，丢失关键事件；而 threading 在 CPython 中受 GIL 限制，对 I/O 密集型任务虽可用，但涉及硬件设备（如 USB 麦克风）时，multiprocessing 更可靠——它为每个任务分配独立进程与内存空间，真正并行运行。
核心设计思路是：一个进程持续读取传感器，检测到超阈值即“发信号”；另一进程等待信号后启动录音，期间屏蔽重复触发，录音结束自动恢复监听。这里推荐使用 multiprocessing.Lock 作为轻量级同步原语——初始加锁表示“未就绪录音”，传感器进程释放锁即“触发录音”，录音进程通过 acquire() 阻塞等待，完美解耦逻辑。
以下是可直接运行的结构化实现（含模拟测试逻辑）：
import multiprocessing
import time
import random
# 【模拟传感器读取】实际项目中替换为 GPIO/ADC 读取代码
def read_from_loudness_meter():
time.sleep(0.1) # 模拟传感器响应延迟
return random.randint(0, 100)
# 【模拟录音操作】实际项目中替换为 pyaudio 或 sounddevice 录制
def record_10_seconds():
print("\n✅ 录音开始（10秒）")
time.sleep(10)
print("⏹️ 录音结束\n")
# 【传感器进程】持续采样，超阈值则释放锁（触发录音）
def reading_values_from_loudness_meter(recording_lock, threshold):
print(f"? 启动分贝监测（阈值: {threshold}）...")
while True:
loudness = read_from_loudness_meter()
if loudness > threshold:
print(f"⚠️ 检测到强噪声: {loudness}dB → 触发录音")
try:
recording_lock.release() # 发送触发信号
except ValueError:
pass # 锁已释放，忽略
# 【录音进程】等待锁释放后执行录音，完成后立即重新上锁防重复触发
def recording_audio(recording_lock):
while True:
recording_lock.acquire() # 阻塞等待触发信号
record_10_seconds()
# 关键：非阻塞重锁，丢弃录音期间产生的新峰值（防连续录音）
# 若需“录音中再触发则立即开始下一段”，删除下一行
recording_lock.acquire(block=False)
if __name__ == "__main__":
# 主进程初始化
threshold = 95 # 根据实际传感器校准调整
recording_lock = multiprocessing.Lock()
recording_lock.acquire() # 初始锁定：禁止立即录音
# 启动录音进程（后台常驻）
recorder = multiprocessing.Process(
target=recording_audio,
args=(recording_lock,)
)
recorder.start()
try:
print("⚡ 系统就绪。按 Ctrl+C 停止。")
# 主进程直接运行传感器逻辑（无需另启进程，更简洁）
reading_values_from_loudness_meter(recording_lock, threshold)
except KeyboardInterrupt:
print("\n? 用户中断，正在退出...")
finally:
recorder.terminate()
recorder.join()
print("✅ 清理完成。")
关键注意事项：
macOS UI 自动化技能包 —— 用 Python 控制你的 Mac
下载
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限
- ✅
if __name__ == "__main__":必须存在：Windows/macOS 下multiprocessing需此保护，否则会递归创建子进程导致崩溃；Linux 虽可省略，但强烈建议统一写法。 - ✅ 锁的初始状态必须为
acquire()：确保首次录音前处于等待状态，避免竞争。 - ✅ 录音进程应常驻，传感器逻辑在主进程运行：比双
Process更易调试，且避免p2.join()导致主进程卡死（原问题中循环start/join是根本错误——Process对象不可复用）。 - ⚠️ 实际硬件接入时：
- 分贝传感器若基于 ADC（如 MCP3008），需在
read_from_loudness_meter()中加入spidev或gpiozero读取； - 录音部分推荐用
pyaudio实现 10 秒 WAV 录制，并添加时间戳命名文件（如rec_20240520_142315.wav）； - 树莓派 USB 麦克风需提前配置
alsa默认设备，避免pyaudio找不到输入流。
- 分贝传感器若基于 ADC（如 MCP3008），需在
进阶优化方向：
- 使用
multiprocessing.Queue传递具体噪声强度值或时间戳，使录音文件名携带事件信息； - 引入
multiprocessing.Event替代Lock，语义更清晰（event.set()触发 /event.wait()等待）； - 添加进程健康检查：当录音进程意外退出时，主进程自动重启它。
通过本方案，你将获得一个稳定、低延迟、易于扩展的多进程音频事件记录系统——传感器永不中断，录音精准响应，真正实现嵌入式环境下的实时异步协同。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
