跨类获取函数返回值问题:事件触发后值无法动态更新
解决B类获取A类事件触发更新值的问题
你的问题其实出在两个核心点上:实例不共享和阻塞式循环,我来一步步给你拆解并解决:
问题根源分析
- 实例不统一:你在B类里直接新建了
A = A(),但如果logic()方法是在其他地方调用的另一个A实例,那B类的这个A实例根本没被修改过,自然一直返回0。 - 阻塞主线程:
logic()里的while循环如果在主线程运行,会彻底卡住程序,B类根本没机会执行获取值的操作。
解决方案1:共享实例+线程化循环(基础版)
先解决实例共享的问题,再把循环放到线程里避免阻塞,这样B类就能实时拿到更新后的值了。
修改后的代码
import threading import time class A: def __init__(self): self.ev_nearby = 0 self.is_running = False # 用来控制循环启停的标志 def changeValue(self): self.ev_nearby = 1 return self.ev_nearby def getChangedValue(self): return self.ev_nearby def logic(self): self.is_running = True # 模拟你的somethingHappens条件,实际替换成你的事件判断逻辑 while self.is_running: self.changeValue() print(f"A类已更新ev_nearby: {self.ev_nearby}") time.sleep(2) # 模拟事件间隔,按需调整 class B: def __init__(self, a_instance): # 接收外部传入的A实例,确保和运行logic的是同一个 self.a_instance = a_instance def fetch_a_value(self): value_from_a = self.a_instance.getChangedValue() print(f"B类获取到A的值: {value_from_a}") return value_from_a # 实际调用示例 if __name__ == "__main__": # 创建唯一的A实例 a_obj = A() # 把同一个A实例传给B b_obj = B(a_obj) # 把A的logic放到线程里运行,避免阻塞主线程 logic_thread = threading.Thread(target=a_obj.logic) logic_thread.daemon = True # 设置守护线程,程序退出时自动结束 logic_thread.start() # 模拟B类定时获取值的逻辑 while True: b_obj.fetch_a_value() time.sleep(1)
关键说明
- 实例共享:B类不再自己创建A实例,而是接收外部已经在运行
logic()的那个实例,这样才能拿到真正被修改后的值。 - 线程处理:把
logic()放到单独线程运行,主线程就能正常处理B类的取值操作,不会被循环卡死。
解决方案2:观察者模式(更高效的主动推送)
如果不想让B类一直轮询取值,可以用观察者模式,让A类在值改变时主动通知B类,适合实时性要求高的场景。
代码示例
import threading import time class A: def __init__(self): self.ev_nearby = 0 self.observers = [] # 存储所有需要通知的观察者(比如B类实例) def add_observer(self, observer): self.observers.append(observer) def _notify_observers(self): # 值改变时主动通知所有观察者 for observer in self.observers: observer.on_a_value_updated(self.ev_nearby) def changeValue(self): self.ev_nearby = 1 self._notify_observers() # 触发通知 return self.ev_nearby def logic(self): while True: self.changeValue() time.sleep(2) class B: def on_a_value_updated(self, new_value): # 收到A类的更新通知时自动处理 print(f"B类收到A类的更新值: {new_value}") # 使用示例 if __name__ == "__main__": a_obj = A() b_obj = B() # 把B类注册为A的观察者 a_obj.add_observer(b_obj) # 启动A的逻辑线程 logic_thread = threading.Thread(target=a_obj.logic) logic_thread.daemon = True logic_thread.start() # 保持主线程运行 while True: time.sleep(1)
优势
不需要B类主动轮询,A类值一变化就会立刻推送给B类,减少资源消耗,响应更及时。
内容的提问来源于stack exchange,提问作者KurgerBing
相关产品推荐
相关产品推荐

