避免使用相同的变量命名不同的 asyncio 任务，以防止无法阻止的任务

如何解决避免使用相同的变量命名不同的 asyncio 任务，以防止无法阻止的任务

我正在尝试创建一个实时编码环境，编码人员可以在其中创建和停止异步任务。

我用

启动了一个交互式python会话/asyncio REPL

python -m asyncio。

然后我允许用户按名称创建异步任务，例如：

import asyncio # executed with 'python -m asyncio' command

# simple async function
async def sleep(sleep_time):
    while True:
        print(f'sleeping for {sleep_time} seconds.')
        await asyncio.sleep(sleep_time)

x = asyncio.create_task(sleep(2))
y = asyncio.create_task(sleep(4))

这会创建两个异步运行的任务，x 和 y。

这些任务可以很容易地停止

x.cancel() 和 y.cancel()。

但是如果用户在执行 x = asyncio.create_task(sleep(3)) 之前执行了一个同名 x.cancel() 的新任务，那么两个“x”任务将同时运行。 x.cancel() 将停止 x 的最后一个定义（休眠 3 秒），但随后 x 的原始定义——休眠 2 秒——仍将运行。没有简单的方法可以停止这个仍在后台运行的 x 任务。 x.cancel() 返回 False 而不是 True。

我的目标是避免在用户不小心使用相同的变量名两次时无法阻止的任务在后台运行。有没有办法阻止用户将新定义分配给 x 变量？还是更好的策略？

asyncio 特定的或基本的 Python 解决方案都是可以接受的。

谢谢。

解决方法

if "x" in globals():
    x.cancel()
    del x

这样的东西可能有用吗？ if "x" in globals(): 检查 x 是否已定义，del x“取消定义”x。

或者可以使用 try：在 NameError 未定义的情况下执行取消并捕获 x 异常

try:
    x.cancel()
except NameError:
    ...

如果你只是想prevent the user from launching a new task with the same name as an existing task，同样的技术仍然可以使用，也许是类似的东西：

if "x" in globals():
    if not x.done():
       print(" Another task with the same name is running... ")
    else:
       x = asyncio.create_task(sleep(3))
else:
    x = asyncio.create_task(sleep(3))