Python 多进程实现噪声触发录音的并行控制教程-Python教程-PHP中文网

Python 多进程实现噪声触发录音的并行控制教程

本文介绍如何使用 Python multiprocessing 模块在树莓派上实现“响度检测”与“音频录制”的真正并行运行：当传感器持续监测环境响度时，一旦超阈值即触发 10 秒录音，且检测过程不中断、不阻塞。

本文介绍如何使用 python `multiprocessing` 模块在树莓派上实现“响度检测”与“音频录制”的真正并行运行：当传感器持续监测环境响度时，一旦超阈值即触发 10 秒录音，且检测过程不中断、不阻塞。

在嵌入式音频监控场景（如树莓派 loudness meter）中，常见需求是：响度传感器持续采样 → 超阈值瞬间启动录音 → 录音期间仍需继续采样 → 多次触发需支持排队或防重叠。这本质上是一个典型的生产者-消费者并发问题，而 threading 因 GIL 限制难以满足实时 I/O 并行需求，multiprocessing 才是更可靠的选择。

但初学者常陷入两个典型误区：

❌ 反复调用 process.start()（进程只能启动一次，重复调用会报错）；
❌ 在主进程中用 while True: p.start(); p.join() 串行重启子进程，彻底丧失并行性。

正确解法是：让两个进程长期存活，通过进程间同步原语（如 multiprocessing.Lock）协调状态。核心逻辑如下：

检测进程（Producer）：持续读取传感器值，一旦发现 loudness > threshold，就释放一个共享锁（lock.release()），通知录音进程“可以开始工作”；
录音进程（Consumer）：始终阻塞在 lock.acquire() 上等待信号；收到信号后立即执行 record_10_seconds()；录音结束后，主动重新加锁（acquire(block=False)），避免在录音过程中重复响应新触发——这一设计可防止“录音嵌套”或资源冲突。

以下是精简可运行的参考实现（已适配树莓派实际部署场景）：

macOS UI 自动化技能包 —— 用 Python 控制你的 Mac

让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限

下载

import multiprocessing
import time
import random

# 【模拟传感器读取】实际项目中替换为 GPIO/ADC 读取逻辑
def read_from_loudness_meter():
    time.sleep(0.1)  # 模拟硬件采样间隔
    return random.randint(0, 100)

# 【模拟录音】实际项目中替换为 pyaudio 或 arecord 命令调用
def record_10_seconds():
    print("\n▶️ 正在录制 10 秒音频...")
    time.sleep(10)
    print("⏹️ 录制完成\n")

def reading_values_from_loudness_meter(recording_lock, threshold):
    """持续检测响度，超阈值即释放锁以触发录音"""
    print("? 响度监测已启动（Ctrl+C 停止）...")
    while True:
        loudness = read_from_loudness_meter()
        if loudness > threshold:
            print(f"⚠️  检测到高响度 {loudness} > {threshold}，触发录音")
            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  # 根据实际传感器校准调整
    lock = multiprocessing.Lock()
    lock.acquire()  # 初始锁定：确保首次录音需显式触发

    # 启动录音进程（守护模式，后台持续运行）
    recorder = multiprocessing.Process(
        target=recording_audio,
        args=(lock,),
        name="AudioRecorder"
    )
    recorder.start()

    try:
        # 主进程运行检测逻辑（非阻塞）
        reading_values_from_loudness_meter(lock, THRESHOLD)
    except KeyboardInterrupt:
        print("\n? 监测已停止")
    finally:
        recorder.terminate()
        recorder.join()

✅ 关键注意事项：

必须包含 if __name__ == "__main__": —— Windows/macOS 下 fork 安全必需，树莓派（Linux）虽可省略，但强烈建议保留以保障跨平台兼容性；
recording_lock 是 multiprocessing.Lock 实例（非 threading.Lock），它是进程安全的共享同步对象；
recorder.terminate() 应在退出前显式调用，避免僵尸进程；
实际部署时，将 read_from_loudness_meter() 替换为 ADC（如 MCP3008）或数字传感器（如 MAX4466 配 GPIO 中断）读取逻辑；record_10_seconds() 可调用 subprocess.run(['arecord', '-d', '10', 'capture.wav']) 实现零依赖录音。