Python多线程跨通信：定时循环函数的触发与停止方案咨询

针对你的需求，我推荐用**Python多线程+线程事件（threading.Event）**的方案，它能完美实现两个函数独立循环、同时执行，并且安全控制Function2的启停。下面是详细的思路和代码示例：

核心思路 #

1. 并发执行：用两个线程分别运行Function1和Function2，让它们同时独立循环。
2. 启动触发：Function1每60秒循环一次，第一次运行时自动启动Function2的线程（也可以根据需求调整为每次循环都检查并重启）。
3. 安全停止：用threading.Event作为停止信号，在“特定时刻”触发这个信号，让Function2优雅退出循环，避免强制终止线程导致的资源泄漏。

完整代码示例 #

import threading
import time

# Function2：每5秒循环一次，直到收到停止信号
def function2(stop_event):
    while not stop_event.is_set():
        # 替换成你实际要执行的任务逻辑
        print("Function2 正在运行...")
        time.sleep(5)
    print("Function2 已停止")

# Function1：每60秒循环一次，负责触发Function2启动
def function1(stop_event, func2_thread):
    while True:
        # 替换成你实际要执行的任务逻辑
        print("Function1 正在运行...")
        
        # 检查Function2线程是否存活，未存活则启动（确保只启动一次，避免重复创建）
        if not func2_thread.is_alive():
            func2_thread.start()
        
        # 等待60秒进入下一次循环
        time.sleep(60)
        
        # （可选）在这里添加自动停止Function2的条件，比如满足某个业务逻辑时
        # if some_business_condition:
        #     stop_event.set()
        #     break

if __name__ == "__main__":
    # 创建停止事件，用于控制Function2的启停
    stop_event = threading.Event()
    
    # 创建Function2的线程，设置为守护线程（主线程退出时自动终止）
    func2_thread = threading.Thread(
        target=function2, 
        args=(stop_event,), 
        daemon=True
    )
    
    # 创建Function1的线程
    func1_thread = threading.Thread(
        target=function1, 
        args=(stop_event, func2_thread), 
        daemon=True
    )
    
    # 启动Function1线程
    func1_thread.start()
    
    # 模拟“特定时刻停止Function2”：等待用户输入回车后停止
    input("按下回车键停止Function2...\n")
    
    # 触发停止信号
    stop_event.set()
    
    # 等待线程优雅结束
    func2_thread.join()
    func1_thread.join()
    
    print("所有函数已停止")

方案优势 #

• 线程安全：用Event传递停止信号，避免强制终止线程的风险，保证资源能正常释放。
• 独立性：两个线程完全独立运行，互不阻塞，满足“同时执行”的需求。
• 灵活性：可以轻松调整循环间隔，或者修改Function1触发Function2的逻辑（比如每次循环都重启Function2）。
• 可扩展性：如果后续需要添加更多定时任务，只需要新增线程即可。

备选方案：异步IO（asyncio） #

如果你的函数以IO操作为主（比如网络请求、文件读写），可以用asyncio实现更高效的并发：

import asyncio

async def function2(stop_event):
    while not stop_event.is_set():
        print("Function2 正在运行...")
        await asyncio.sleep(5)
    print("Function2 已停止")

async def function1(stop_event):
    # 启动Function2异步任务
    func2_task = asyncio.create_task(function2(stop_event))
    
    while True:
        print("Function1 正在运行...")
        await asyncio.sleep(60)
        
        # （可选）自动停止条件
        # if some_condition:
        #     stop_event.set()
        #     await func2_task
        #     break

async def main():
    stop_event = asyncio.Event()
    func1_task = asyncio.create_task(function1(stop_event))
    
    # 模拟特定时刻停止
    await asyncio.to_thread(input, "按下回车键停止Function2...\n")
    
    stop_event.set()
    await func1_task
    print("所有函数已停止")

asyncio.run(main())

注意事项 #

• 如果两个函数需要共享资源（比如全局变量、文件句柄），记得用threading.Lock（线程方案）或异步锁来保护，避免竞争条件。
• 守护线程设置daemon=True是为了让主线程退出时自动终止子线程，但如果需要优雅停止，还是要调用join()等待线程完成收尾工作。

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