You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

跨类获取函数返回值问题:事件触发后值无法动态更新

解决B类获取A类事件触发更新值的问题

你的问题其实出在两个核心点上:实例不共享和阻塞式循环,我来一步步给你拆解并解决:

问题根源分析

  1. 实例不统一:你在B类里直接新建了A = A(),但如果logic()方法是在其他地方调用的另一个A实例,那B类的这个A实例根本没被修改过,自然一直返回0。
  2. 阻塞主线程: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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.05.11 08:57:06