如何为递归asyncio函数统一变量作用域并实现任务阻塞控制

一、避免global，共享变量x的最佳实践 #

不用global关键字实现多函数共享状态，最常用的两种方式是使用可变容器（如字典）或自定义状态类，核心是传递可变对象的引用，让所有函数操作同一个实例的属性/值：

方式1：用字典封装共享状态 #

import asyncio

async def main():
    print('hello from main function')
    # 用字典封装共享的x
    state = {'x': 0}
    print('current x:', state['x'])
    
    # 用create_task替代ensure_future，是Python 3.7+推荐的异步任务创建方式
    asyncio.create_task(what_is_x(state))
    await first_async(state)
    print('final x:', state['x'])
    

async def first_async(state):
    print('hello from first function')
    state['x'] += 1
    print('current x:', state['x'])

    await asyncio.sleep(2)
    await second_async(state)
    

async def second_async(state):
    print('hello from second function')
    state['x'] += 1
    print('current x:', state['x'])

    if state['x'] > 5:
        print('exit second function...')
        return
    await asyncio.sleep(2)
    await first_async(state)


async def what_is_x(state):
    while True:
        print('what is x ?', state['x'])
        await asyncio.sleep(3)
    

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

方式2：自定义状态类（更适合复杂场景） #

如果后续需要扩展更多共享变量，用类封装会更清晰：

import asyncio

class SharedState:
    def __init__(self):
        self.x = 0

async def main():
    print('hello from main function')
    state = SharedState()
    print('current x:', state.x)
    
    asyncio.create_task(what_is_x(state))
    await first_async(state)
    print('final x:', state.x)
    

async def first_async(state):
    print('hello from first function')
    state.x += 1
    print('current x:', state.x)

    await asyncio.sleep(2)
    await second_async(state)
    

async def second_async(state):
    print('hello from second function')
    state.x += 1
    print('current x:', state.x)

    if state.x > 5:
        print('exit second function...')
        return
    await asyncio.sleep(2)
    await first_async(state)


async def what_is_x(state):
    while True:
        print('what is x ?', state.x)
        await asyncio.sleep(3)
    

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

这两种方式运行后，what_is_x()会打印出实时更新的x值，而非固定的0。

二、让what_is_x()阻塞另外两个异步函数的执行 #

要实现what_is_x()与first_async/second_async的互斥执行（即同一时间只有一方能运行），可以用asyncio.Lock（异步锁）。锁会保证同一时间只有一个协程持有锁，其他协程会阻塞直到锁被释放。

修改后的代码（基于自定义状态类+锁）：

import asyncio

class SharedState:
    def __init__(self):
        self.x = 0
        self.exec_lock = asyncio.Lock()  # 定义异步锁

async def main():
    print('hello from main function')
    state = SharedState()
    print('current x:', state.x)
    
    asyncio.create_task(what_is_x(state))
    await first_async(state)
    print('final x:', state.x)
    

async def first_async(state):
    # 获取锁，执行期间其他协程无法获取该锁
    async with state.exec_lock:
        print('hello from first function')
        state.x += 1
        print('current x:', state.x)

        await asyncio.sleep(2)
    await second_async(state)
    

async def second_async(state):
    async with state.exec_lock:
        print('hello from second function')
        state.x += 1
        print('current x:', state.x)

        if state.x > 5:
            print('exit second function...')
            return
        await asyncio.sleep(2)
    await first_async(state)


async def what_is_x(state):
    while True:
        async with state.exec_lock:
            print('what is x ?', state.x)
            await asyncio.sleep(3)  # 持有锁期间，first/second会阻塞
    

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

逻辑说明： #

• 当what_is_x()获取锁后，会持有锁3秒（sleep(3)），这段时间first_async和second_async尝试获取锁时会被阻塞，直到what_is_x()释放锁。
• 反之，当first_async或second_async持有锁时，what_is_x()会等待直到锁被释放后再执行。

如果需要更灵活的阻塞逻辑（比如what_is_x()主动触发/解除阻塞），可以用asyncio.Event：定义一个事件，first_async/second_async在执行前等待事件被设置，what_is_x()控制事件的设置/清除，但这种场景下互斥锁更符合“阻塞执行”的需求。

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