如何在创建数据库行10分钟后自动删除该行？

除了time.sleep+threading的组合，还有以下几种更可靠或更灵活的方案：

1. 基于过期字段的定时清理 #

这种方法依赖数据库字段记录过期时间，再通过定时任务定期清理过期数据，适合不需要精确到秒级延迟的场景。

• 修改表结构：添加expire_at字段存储记录的过期时间

  cursor.execute("""
  CREATE TABLE IF NOT EXISTS customers(
      user_id INT, 
      user_name TEXT, 
      dt_string TEXT, 
      expire_at TIMESTAMP
  )
  """)
  

• 插入记录时设置过期时间：计算当前时间+10分钟作为过期时间

  from datetime import datetime, timedelta
  
  expire_time = datetime.now() + timedelta(minutes=10)
  cursor.execute("""
  INSERT INTO customers(user_id, user_name, dt_string, expire_at)
  VALUES(?, ?, ?, ?)
  """, (user_id, user_name, dt_string, expire_time))
  connection.commit()
  

• 定期执行清理：编写清理脚本，用系统定时任务（如Linux cron、Windows任务计划）或应用内调度器定期运行

  def clean_expired_records():
      conn = sqlite3.connect("clients.db")
      cursor = conn.cursor()
      cursor.execute("DELETE FROM customers WHERE expire_at <= ?", (datetime.now(),))
      conn.commit()
      conn.close()
  

2. 使用Python调度库（APScheduler） #

APScheduler是专门的任务调度库，支持多种调度方式（延迟执行、定时执行等），比手动线程更稳定，还能支持任务持久化。

• 安装依赖：

  pip install apscheduler
  

• 示例代码：

  from apscheduler.schedulers.background import BackgroundScheduler
  from datetime import datetime, timedelta
  import sqlite3
  
  def delete_customer(user_id):
      conn = sqlite3.connect("clients.db")
      cursor = conn.cursor()
      cursor.execute("DELETE FROM customers WHERE user_id=?", (user_id,))
      conn.commit()
      conn.close()
  
  # 初始化后台调度器
  scheduler = BackgroundScheduler()
  # 添加10分钟后执行的删除任务
  scheduler.add_job(
      delete_customer,
      'date',
      run_date=datetime.now() + timedelta(minutes=10),
      args=[user_id]
  )
  # 启动调度器
  scheduler.start()
  

  注意：如果需要应用重启后不丢失任务，可以配置APScheduler使用SQLite作为任务存储。

3. 借助消息队列实现延迟任务 #

如果你的系统已经使用消息队列，可以利用其延迟消息能力来实现，任务独立于应用进程，可靠性更高。

以Redis的有序集合实现简单延迟队列为例：

• 生产者（添加延迟任务）：

  import redis
  import time
  
  r = redis.Redis(host='localhost', port=6379, db=0)
  delay_seconds = 600  # 10分钟
  task_key = f"delete_customer_{user_id}"
  # 将任务加入有序集合，分数为到期时间戳
  r.zadd("delay_tasks", {task_key: time.time() + delay_seconds})
  

• 消费者（执行到期任务）：

  def process_delay_tasks():
      while True:
          now = time.time()
          # 获取所有到期的任务
          tasks = r.zrangebyscore("delay_tasks", 0, now, limit=(0, 10))
          for task in tasks:
              task_str = task.decode()
              # 尝试删除任务，避免多个消费者重复执行
              if r.zrem("delay_tasks", task):
                  target_user_id = int(task_str.split("_")[-1])
                  # 执行删除操作
                  conn = sqlite3.connect("clients.db")
                  cursor = conn.cursor()
                  cursor.execute("DELETE FROM customers WHERE user_id=?", (target_user_id,))
                  conn.commit()
                  conn.close()
          time.sleep(5)  # 每5秒检查一次
  

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