Python中如何并行运行两个无限循环：语音识别与保活信号发送

方案一：使用线程（修正主线程抢占问题） #

你之前遇到的线程抢占问题，大概率是没将保活线程设为守护线程，或者语音识别循环未合理释放CPU时间。以下是修正后的实现：

import threading
import time

# 替换为你的实际语音识别逻辑
def speech_recognition_loop():
    while True:
        print("正在进行语音识别...")
        # 这里编写语音识别代码（如调用SpeechRecognition库的listen方法）
        time.sleep(1)  # 模拟识别耗时，避免占满CPU

# 保活信号发送逻辑
def keep_alive_loop():
    while True:
        print("发送保活信号...")
        # 这里编写发送保活信号的代码（如socket发送、API调用）
        time.sleep(3)  # 每3秒发送一次

if __name__ == "__main__":
    # 创建保活线程并设为守护线程（主线程结束时自动终止）
    keep_alive_thread = threading.Thread(target=keep_alive_loop, daemon=True)
    keep_alive_thread.start()

    # 主线程运行语音识别循环
    speech_recognition_loop()

说明：守护线程不会阻塞程序退出，若你的语音识别代码是阻塞式（如等待麦克风输入），它会自动释放GIL，线程可正常切换，不会出现抢占问题。

方案二：使用asyncio（解决阻塞问题） #

如果语音识别依赖同步阻塞的库，直接放入asyncio事件循环会卡住，需用asyncio.to_thread（Python3.9+）将阻塞代码移至线程池执行：

import asyncio
import time

# 模拟阻塞的语音识别函数
def blocking_speech_recognition():
    print("正在进行语音识别...")
    time.sleep(1)  # 模拟阻塞耗时
    return "识别结果"

async def speech_recognition_loop():
    while True:
        # 将阻塞函数放入线程池执行，不阻塞事件循环
        result = await asyncio.to_thread(blocking_speech_recognition)
        print(f"语音识别结果：{result}")

async def keep_alive_loop():
    while True:
        print("发送保活信号...")
        # 若保活代码也是阻塞式，同样用to_thread包装
        await asyncio.sleep(3)  # 异步休眠，不占用事件循环

async def main():
    # 并行运行两个异步任务
    await asyncio.gather(
        speech_recognition_loop(),
        keep_alive_loop()
    )

if __name__ == "__main__":
    asyncio.run(main())

说明：asyncio.to_thread会将同步代码放到后台线程执行，事件循环可同时处理保活任务，彻底解决阻塞问题。若语音识别库本身支持异步，直接写异步调用即可。

方案三：使用多进程（解决启动问题） #

多进程启动失败通常是Windows下未用if __name__ == "__main__"包裹启动代码，导致子进程重复执行主模块逻辑。以下是正确实现：

import multiprocessing
import time

def speech_recognition_loop():
    while True:
        print("正在进行语音识别...")
        time.sleep(1)

def keep_alive_loop():
    while True:
        print("发送保活信号...")
        time.sleep(3)

if __name__ == "__main__":
    # 创建两个独立进程
    speech_process = multiprocessing.Process(target=speech_recognition_loop)
    keep_alive_process = multiprocessing.Process(target=keep_alive_loop)

    # 启动进程
    speech_process.start()
    keep_alive_process.start()

    # 等待进程结束（让程序持续运行）
    speech_process.join()
    keep_alive_process.join()

说明：if __name__ == "__main__"是Windows下多进程的必要条件，每个进程拥有独立Python解释器，无GIL限制，适合CPU密集型场景（如本地大模型语音推理）。

选择建议 #

• 若语音识别是IO密集型（等待麦克风输入、网络请求），优先选线程/asyncio，开销更小；
• 若语音识别是CPU密集型（本地模型推理），选多进程可充分利用多核CPU。

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