如何无需多进程从主脚本中断Python模块中的无限循环？

方案一：线程+线程安全标志位 #

利用Python的threading模块将采集循环放在子线程中运行，通过线程安全的Event对象控制循环启停，避免阻塞主脚本的执行流程。

import threading
import time

class SensorHandler:
    def __init__(self):
        self._stop_event = threading.Event()
        self._capture_thread = None

    def _capture_loop(self):
        # 替换为实际的传感器采集逻辑
        while not self._stop_event.is_set():
            print("正在采集传感器数据...")
            time.sleep(1)
        print("采集循环已终止")

    def start_capture(self):
        if not self._capture_thread or not self._capture_thread.is_alive():
            self._stop_event.clear()
            self._capture_thread = threading.Thread(target=self._capture_loop)
            self._capture_thread.daemon = True  # 设置为守护线程，随主进程退出自动终止
            self._capture_thread.start()

    def stop_capture(self):
        self._stop_event.set()
        if self._capture_thread:
            self._capture_thread.join(timeout=2)  # 等待线程优雅退出，超时则强制终止

主脚本调用示例：

handler = SensorHandler()
handler.start_capture()
# 主脚本执行其他业务逻辑
time.sleep(5)
# 停止传感器采集
handler.stop_capture()

方案二：生成器+主动迭代控制 #

将采集逻辑封装为生成器，由主脚本通过迭代生成器来驱动循环，同时提供停止标志位让生成器主动退出。这种方式无需额外线程，完全由主脚本控制采集节奏。

import time

class SensorHandler:
    def __init__(self):
        self._should_stop = False

    def measure(self):
        self._should_stop = False
        while not self._should_stop:
            # 模拟传感器数据采集，替换为实际逻辑
            sensor_data = f"当前传感器读数: {time.time()}"
            yield sensor_data
            time.sleep(1)

    def stop_capture(self):
        self._should_stop = True

主脚本调用示例：

handler = SensorHandler()
start_time = time.time()
# 迭代生成器获取采集数据
for data in handler.measure():
    print(data)
    # 主脚本可根据条件主动触发停止，或外部调用stop_capture
    if time.time() - start_time > 5:
        handler.stop_capture()
        break

场景说明：

• 方案一适合需要后台持续采集、主脚本同时处理其他任务的场景，线程后台运行不阻塞主流程。
• 方案二更轻量化，采集节奏完全由主脚本控制，适合需要同步处理采集数据的场景。

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