Python Asyncio全局变量更新未被检测，求原因及解决方案

1. 循环导入导致的模块变量副本问题 #

你的代码存在循环导入：main.py导入foo，foo.py又导入main。Python在这种场景下会生成两个独立的模块实例：

• 直接运行main.py时，它会被识别为__main__模块，循环里判断的RUNNING是__main__.RUNNING。
• 而foo.py导入的main是标准模块名的main，对应的是另一个模块对象，stop()修改的是main.RUNNING。

这两个RUNNING是完全独立的变量，所以foo里能读到修改后的False，但主函数的循环始终读取的是__main__模块里未被修改的True。

2. 同步阻塞卡住事件循环 #

主函数里用time.sleep(1)是同步阻塞调用，会彻底卡死asyncio的事件循环。哪怕后续改用asyncio.Event，事件循环也无法及时处理状态更新——因为sleep期间整个事件循环都处于停滞状态。

方案1：消除循环导入，传递控制对象 #

把需要共享的状态或控制逻辑直接传递给Foo类，彻底避免循环导入：

修改后代码： #

main.py

import asyncio
import time
import foo

RUNNING = True

async def main():
    # 把stop函数和RUNNING的读取逻辑传给Foo
    f = foo.Foo(stop_func=stop, get_running=lambda: RUNNING)
    i = 0

    while RUNNING:
        print("running")
        # 替换同步sleep为异步的asyncio.sleep，避免阻塞事件循环
        await asyncio.sleep(1)
        
        f.do_work(i)
        i += 1

def stop():
    print("stop call received")
    global RUNNING
    RUNNING = False

if __name__ == "__main__":
    print("starting")
    asyncio.run(main())

foo.py

class Foo:
    def __init__(self, stop_func, get_running):
        self.stop_func = stop_func
        self.get_running = get_running

    def do_work(self, i):
        if i == 3:
            self.stop_func()
            print(f"当前RUNNING值: {self.get_running()}")

方案2：单独的状态管理模块 #

创建一个独立的状态模块，所有模块都从这里读写共享状态，避免循环导入的变量副本问题：

新增state.py #

RUNNING = True

def set_running(value):
    global RUNNING
    RUNNING = value

def get_running():
    return RUNNING

修改main.py #

import asyncio
import time
import foo
import state

async def main():
    f = foo.Foo()
    i = 0

    while state.get_running():
        print("running")
        await asyncio.sleep(1)
        
        f.do_work(i)
        i += 1

def stop():
    print("stop call received")
    state.set_running(False)

if __name__ == "__main__":
    print("starting")
    asyncio.run(main())

修改foo.py #

import state

class Foo:
    def do_work(self, i):
        if i == 3:
            import main
            main.stop()
            print(f"当前RUNNING值: {state.get_running()}")

方案3：改用asyncio.Event（推荐，符合异步编程规范） #

用asyncio原生的事件机制管理停止信号，同时消除循环导入，替换同步阻塞的sleep：

修改main.py #

import asyncio
import foo

async def main():
    stop_event = asyncio.Event()
    f = foo.Foo(stop_event)
    i = 0

    while not stop_event.is_set():
        print("running")
        await asyncio.sleep(1)
        
        f.do_work(i)
        i += 1

if __name__ == "__main__":
    print("starting")
    asyncio.run(main())

修改foo.py #

class Foo:
    def __init__(self, stop_event):
        self.stop_event = stop_event

    def do_work(self, i):
        if i == 3:
            print("stop call received")
            self.stop_event.set()
            print(f"停止事件已触发: {self.stop_event.is_set()}")

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