如何在再次按下按钮时取消异步协程？

Question

再次单击按钮时如何取消已经运行的任务？

我已经草拟了代码，但它肯定不起作用

class TasksManager:
    def __init__(self) -> None:
        self.state = False
        self.list = [self.task_1, self.task_2, self.task_3]
 
    async def task_1(self):
        for i in range(10):
            await asyncio.sleep(1)
            print("Task 1: ", i)
    
    async def task_2(self):
        for i in range(10):
            await asyncio.sleep(1)
            print("Task 2: ", i)
 
    async def task_3(self):
        for i in range(10):
            await asyncio.sleep(1)
            print("Task 3: ", i)
 
    async def click(self, e: ft.TapEvent):
        if not self.state:
            self.state = True
            list = [asyncio.create_task(task()) for task in self.list]
            done, pending = await asyncio.wait(list, return_when=asyncio.FIRST_COMPLETED)
            print(done, pending)
        else:
            # здесь нужно отменить корутины
            for task in pending:
                task.cancel()
 
def main(page: ft.Page):
    manager = TasksManager()
    button = ft.TextButton("Click", on_click=manager.click)
    page.add(button)
    page.update()
 
ft.app(main)

任务必须强行停止

Answer 1

虽然更多详细信息会有所帮助，但

click

中存在几个问题。不过，我对 Flet 不太熟悉。

1. 创建任务后，您需要将其保存。 Python 的文档中提到了这一点

   asyncio.create_task

   :

   重要：保存对此函数结果的引用，以避免任务在执行过程中消失。事件循环仅保留对任务的弱引用。未在其他地方引用的任务可能会随时被垃圾收集，甚至在完成之前。对于可靠的“一劳永逸”后台任务，请将它们收集在一个集合中：

2. 您正在调用

   asyncio.wait

   ，它将阻塞，直到任务完成或取消。这是有关 Flet 和用例的更多详细信息很重要的地方。

   click

   是在它自己的任务中启动的吗？如果没有，对

   click

   的两次调用将等待第一个调用完成，然后再执行第二个调用。

3. 您没有在类中保存任务，因此后续调用

   click

   无法取消先前执行中创建的任务。

---

尝试这样的事情：

class TaskManager:
    def __init__(self) -> None:
        self.state = False
        self.__tasks_to_create = [self.task_1, self.task_2, self.task_3]
        self.__tasks: list[asyncio.Task] = []
 
    ...
 
    async def click(self, e: ft.TapEvent):
        if not self.state:
            self.state = True
            self.__tasks = [asyncio.create_task(task()) for task in self.__tasks_to_create]

    async def cancel_tasks(self, e: ft.TapEvent):
        for task in self.__tasks:
            task.cancel()
 
def main(page: ft.Page):
    manager = TaskManager()
    button = ft.TextButton("Click", on_click=manager.click)
    cancel_button = ft.TextButton("Cancel", on_click=manager.cancel_tasks)
    page.add(button)
    page.add(cancel_button)
    page.update()
 
ft.app(main)