Telegram Bot异步定时执行MySQL更新任务遇阻求助

核心方案：使用python-telegram-bot的job_queue #

不用再纠结schedule模块的阻塞问题，job_queue是库内置的定时任务工具，和机器人的事件循环深度集成，完全不会影响用户命令的响应。

步骤1：编写定时任务函数 #

函数必须接受context参数（job_queue会自动传入），在函数内完成数据库更新操作：

import mysql.connector
from telegram.ext import Updater, CommandHandler
from datetime import time

# 替换成你的实际数据库配置
DB_CONFIG = {
    'user': '你的数据库账号',
    'password': '你的数据库密码',
    'host': 'localhost',
    'database': '你的数据库名'
}

def add_daily_rolls(context):
    conn = None
    cursor = None
    try:
        # 建立数据库连接
        conn = mysql.connector.connect(**DB_CONFIG)
        cursor = conn.cursor()
        # 执行目标SQL语句
        sql = "UPDATE users SET rolls = rolls + 1 WHERE rolls < 12;"
        cursor.execute(sql)
        conn.commit()
        print(f"更新完成，共修改{cursor.rowcount}条记录")
    except mysql.connector.Error as err:
        print(f"数据库错误: {err}")
        if conn:
            conn.rollback()  # 出错时回滚保证数据一致性
    finally:
        # 强制关闭游标和连接，避免资源泄漏
        if cursor:
            cursor.close()
        if conn:
            conn.close()

步骤2：在机器人启动时注册定时任务 #

在主函数中，通过updater.job_queue添加每日执行的任务，指定你需要的时间点（比如每日8点、14点、20点）：

# 你的/rolls命令处理函数（示例，你可替换为自己的实现）
def get_rolls(update, context):
    user_id = update.effective_user.id
    conn = None
    cursor = None
    try:
        conn = mysql.connector.connect(**DB_CONFIG)
        cursor = conn.cursor()
        cursor.execute("SELECT rolls FROM users WHERE user_id = %s;", (user_id,))
        result = cursor.fetchone()
        if result:
            update.message.reply_text(f"你的rolls数量: {result[0]}")
        else:
            # 首次查询时初始化用户数据
            cursor.execute("INSERT INTO users (user_id, rolls) VALUES (%s, 0);", (user_id,))
            conn.commit()
            update.message.reply_text("已为你初始化rolls，当前数量: 0")
    except mysql.connector.Error as err:
        update.message.reply_text(f"查询失败: {err}")
    finally:
        if cursor:
            cursor.close()
        if conn:
            conn.close()

def main():
    # 替换成你的机器人Token
    updater = Updater("你的Telegram机器人Token")
    dp = updater.dispatcher

    # 注册/rolls命令处理器
    dp.add_handler(CommandHandler("rolls", get_rolls))

    # 获取job_queue实例
    job_queue = updater.job_queue

    # 设置每日执行的三个时间点
    daily_times = [
        time(hour=8, minute=0),
        time(hour=14, minute=0),
        time(hour=20, minute=0)
    ]

    # 为每个时间点添加定时任务
    for run_time in daily_times:
        job_queue.run_daily(add_daily_rolls, time=run_time)
        print(f"已添加定时任务：每日{run_time.hour}:{run_time.minute}执行rolls更新")

    # 启动机器人
    updater.start_polling()
    updater.idle()

if __name__ == "__main__":
    main()

关键说明 #

• 为什么放弃schedule：schedule的while True循环会阻塞机器人的事件循环，导致无法处理用户的命令请求；而job_queue是基于机器人的事件循环运行的，两者完全兼容。
• context参数：job_queue会自动给定时任务函数传入这个参数，即使你在函数里不用它，也必须保留这个参数，否则会报错。
• 数据库连接：每次任务执行时新建连接、执行后关闭，避免长时间持有连接导致失效；如果需要更高性能，可以考虑使用数据库连接池。

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