在 Python 中实现实时获取数据并能够通过按键中止程序运行，可以通过结合信号处理机制与主循环中的数据采集逻辑来实现。程序可以持续从数据源（如传感器、网络接口或文件流）读取信息，并在检测到特定中断信号（如 `Ctrl+C`）时优雅地中止运行。 为了防止因误按 `Ctrl+C` 导致程序意外终止，可以在程序中注册自定义的信号处理函数，以替代默认的中断行为。该处理函数可以设置一个标志位，用于通知主循环退出。这种方式不仅提高了程序的稳定性，还能在程序退出时执行清理操作，例如保存状态、关闭文件句柄或释放资源[^2]。 以下是一个实现示例： ```python import time import signal import sys # 标志位，用于控制主循环是否继续运行 running = True def signal_handler(signal, frame): """信号处理函数，用于捕获 Ctrl+C 并设置退出标志""" global running print("正在中止程序...") running = False # 注册信号处理函数 signal.signal(signal.SIGINT, signal_handler) # 实时获取数据的模拟函数 def fetch_data(): return time.time() # 主循环，持续获取数据直到收到中断信号 print("程序正在运行，请按 Ctrl+C 中止") while running: data = fetch_data() print(f"当前数据: {data}") time.sleep(1) ``` 上述代码中，`signal.signal(signal.SIGINT, signal_handler)` 用于注册一个信号处理函数，该函数在检测到 `Ctrl+C` 时将全局变量 `running` 设置为 `False`，从而退出主循环。这种方式避免了程序直接崩溃，并能控制输出信息的格式，使程序更加健壮和用户友好[^2]。 在实际应用中，`fetch_data()` 函数可以替换为从传感器读取数值、从网络接口接收数据流，或者从数据库中查询最新记录等操作。此外，还可以在退出前加入清理逻辑，例如关闭串口连接、保存缓存数据等，以确保程序的完整性。 ### 多线程与异步数据采集 在某些情况下，数据采集任务可能需要在后台运行，以避免阻塞主循环或信号处理流程。此时，可以结合 `threading` 模块实现并发数据采集。通过将数据获取逻辑放入独立线程中运行，主循环可以专注于响应中断信号或处理其他任务。 以下是一个多线程版本的示例： ```python import threading import time import signal import sys running = True def signal_handler(signal, frame): global running print("正在中止程序...") running = False signal.signal(signal.SIGINT, signal_handler) def data_collector(): while running: data = time.time() print(f"采集到数据: {data}") time.sleep(1) collector_thread = threading.Thread(target=data_collector) collector_thread.start() collector_thread.join() print("程序已安全退出") ``` 该实现中，`data_collector` 函数在一个独立线程中运行，主循环仅负责等待线程结束。这种方式有助于将数据采集逻辑与主控逻辑分离，提高程序的可维护性和可扩展性。