Python中如何实现延迟任务的增加和删除

在Python里搞延迟任务，用asyncio或者threading.Timer都行。asyncio适合异步应用，Timer适合简单场景。下面给你两个能直接跑的方案。

方案一：用asyncio（推荐给异步程序） 这个方案用asyncio.create_task、asyncio.sleep和asyncio.CancelledError来管理任务。

import asyncio

class AsyncDelayManager:
    def __init__(self):
        self.tasks = {}  # 存任务ID和对应的task对象
        self.next_id = 0

    async def _exec_task(self, task_id, delay, func, *args, **kwargs):
        try:
            await asyncio.sleep(delay)
            await func(*args, **kwargs) if asyncio.iscoroutinefunction(func) else func(*args, **kwargs)
        except asyncio.CancelledError:
            print(f"任务 {task_id} 被取消")
        finally:
            self.tasks.pop(task_id, None)

    def add_task(self, delay, func, *args, **kwargs):
        """增加延迟任务，返回任务ID"""
        task_id = self.next_id
        self.next_id += 1
        task = asyncio.create_task(self._exec_task(task_id, delay, func, *args, **kwargs))
        self.tasks[task_id] = task
        return task_id

    def cancel_task(self, task_id):
        """取消任务"""
        if task_id in self.tasks:
            self.tasks[task_id].cancel()
            return True
        return False

# 使用示例
async def my_task(name):
    print(f"执行任务: {name}")

async def main():
    manager = AsyncDelayManager()
    
    # 增加3秒后执行的任务
    task_id = manager.add_task(3, my_task, "hello")
    print(f"已创建任务 ID: {task_id}")
    
    # 取消任务（把这行注释掉就能看到任务执行）
    # manager.cancel_task(task_id)
    
    # 等足够时间让任务执行或取消
    await asyncio.sleep(5)

# 运行
asyncio.run(main())

方案二：用threading.Timer（适合同步程序） 这个更简单直接，但注意它是在独立线程里跑的。

import threading
from typing import Callable, Any

class TimerDelayManager:
    def __init__(self):
        self.timers = {}  # 存任务ID和Timer对象
        self.next_id = 0

    def add_task(self, delay: float, func: Callable, *args, **kwargs) -> int:
        """增加延迟任务，返回任务ID"""
        def task_wrapper():
            try:
                func(*args, **kwargs)
            finally:
                self.timers.pop(task_id, None)
        
        task_id = self.next_id
        self.next_id += 1
        
        timer = threading.Timer(delay, task_wrapper)
        timer.start()
        self.timers[task_id] = timer
        return task_id

    def cancel_task(self, task_id: int) -> bool:
        """取消任务"""
        if task_id in self.timers:
            self.timers[task_id].cancel()
            self.timers.pop(task_id, None)
            return True
        return False

# 使用示例
def my_sync_task(name):
    print(f"执行同步任务: {name}")

if __name__ == "__main__":
    manager = TimerDelayManager()
    
    # 增加5秒后执行的任务
    task_id = manager.add_task(5, my_sync_task, "world")
    print(f"已创建任务 ID: {task_id}")
    
    # 取消任务（把这行注释掉就能看到任务执行）
    # manager.cancel_task(task_id)
    
    # 主线程等一会儿
    threading.Event().wait(7)

选哪个？

• 如果你的程序已经是asyncio的，用方案一。
• 如果是普通同步程序，用方案二。

一句话建议：异步程序用asyncio方案，同步程序用Timer方案。