如何将Python脚本生成的实时值不间断发送到另一脚本的线程函数中

采用**线程安全的queue.Queue**作为中间数据缓冲，完全匹配所有限制条件，无需修改核心业务逻辑，也不会出现循环导入问题。

第一步：修改第二个脚本（secondScript.py） #

仅新增队列属性和数据推送方法，原有读取Arduino的逻辑完全不变：

from threading import Thread
import queue

class Second:
    def __init__(self):
        # 线程安全队列，maxsize设为1仅保留最新结果，不需要留存历史数据可以直接用这个配置
        self.result_queue = queue.Queue(maxsize=1)
        
        self.dum = Thread(target=self.func1)
        self.dum.daemon = True # 注意原代码拼写错误，正确拼写为daemon
        self.dum.start()

        # 无需再传入args参数，func2自行从队列取最新数据
        self.tis = Thread(target=self.func2)
        self.tis.daemon = True
        self.tis.start()

    # 新增数据推送接口，供第一个脚本调用
    def push_result(self, lastResult):
        # 队列满时先丢弃旧数据再存入新值，全程非阻塞
        if self.result_queue.full():
            self.result_queue.get_nowait()
        self.result_queue.put_nowait(lastResult)

    def func1(self):
        while True:
            # 原有读取Arduino的逻辑保留不动
            pass
 
    def func2(self):
        while True:
            try:
                # 阻塞等待新数据，无数据时自动让出CPU资源
                lastResult = self.result_queue.get(timeout=1)
                print(lastResult)
            except queue.Empty:
                # 无数据时继续循环，不影响线程运行
                continue

第二步：修改第一个脚本，仅新增2行代码，核心逻辑完全不动 #

from secondScript import Second

# 全局仅初始化一次Second实例
second_ins = Second()

# 原有其他业务逻辑保留不动
"""
"""
"""
while True:
    lastResult = <a list> # 原有生成实时结果的逻辑完全不变
    # 仅新增这一行推送数据，全程非阻塞，不会影响摄像头读取速度
    second_ins.push_result(lastResult)

方案符合限制的说明 #

• 无循环导入问题：仅第一个脚本导入第二个脚本的Second类，第二个脚本不需要导入第一个脚本的任何内容
• 不会阻塞第一个脚本运行：数据推送用非阻塞方法实现，无sleep、无IO等待，完全不影响摄像头实时读取逻辑
• 第二个脚本双线程独立运行：读取Arduino的func1逻辑没有任何改动，func2从队列取数打印，两个线程互不干扰
• 第一个脚本修改量极小：仅新增实例化和数据推送两行代码，原有业务逻辑无任何调整

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