Py-Cord定时任务触发后无限执行问题求助

问题描述 #

使用Py-Cord开发Discord机器人，需求是让机器人每日特定时间执行一次指定函数，但到达设定时间后，函数开始无限重复执行。机器人部署在云服务器上，推测问题与重连机制有关，但未找到具体原因。

相关代码 #

import discord
import datetime
from discord.ext import commands, tasks
import requests

utc = datetime.timezone(datetime.timedelta(hours=9))
send_time = datetime.time(hour=1, minute=0,tzinfo=utc)

class dailyNotice(commands.Cog):
    def __init__(self, bot):
        self.bot = bot
        self.my_task.start()

    def cog_unload(self):
        self.my_task.cancel()

    @tasks.loop(time=send_time)
    async def my_task(self):
         another_function()

    @my_task.before_loop
    async def before_printer(self):
        print('waiting...')
        await self.bot.wait_until_ready()

@bot.event
async def on_ready():
    await bot.change_presence(status=discord.Status.online)
    
    loaded_cogs = bot.cogs.keys()
    if "dailyNotice" in loaded_cogs:
        print('dailyNotice already running')
    elif "dailyNotice" not in loaded_cogs:
        print('dailyNotice')
        bot.add_cog(dailyNotice(bot))

补充说明 #

曾添加以下代码试图防止重连时重复加载Cog，但终端未输出任何相关打印信息：

loaded_cogs = bot.cogs.keys()
if "dailyNotice" in loaded_cogs:
    print('dailyNotice already running')
elif "dailyNotice" not in loaded_cogs:
    print('dailyNotice')
    bot.add_cog(dailyNotice(bot))

问题原因 #

1. on_ready事件多次触发：Discord机器人在重连、断线恢复时会多次触发on_ready事件，每次触发都会执行添加Cog的逻辑，导致多个dailyNotice实例被创建，每个实例都会启动一个my_task任务。到设定时间后，所有任务实例同时执行函数，表现为无限重复执行。
2. Cog存在性判断失效：重连过程中，机器人的Cog列表可能被重置，导致bot.cogs中找不到已加载的dailyNotice，进而重复添加。

修复方案 #

方案1：提前加载Cog（推荐） #

将Cog的加载逻辑移到on_ready事件外，确保只加载一次：

# 在Bot初始化完成后直接添加Cog，无需在on_ready中处理
bot.add_cog(dailyNotice(bot))

@bot.event
async def on_ready():
    await bot.change_presence(status=discord.Status.online)
    print("机器人已上线")

方案2：标记初始化状态，避免重复执行 #

用全局变量标记是否已完成Cog加载，确保on_ready仅在第一次触发时执行加载逻辑：

initialized = False

@bot.event
async def on_ready():
    global initialized
    if initialized:
        return
    initialized = True
    
    await bot.change_presence(status=discord.Status.online)
    print('dailyNotice')
    bot.add_cog(dailyNotice(bot))

方案3：在Cog内部检查任务状态 #

在Cog的初始化方法中，检查任务是否已运行，避免重复启动：

class dailyNotice(commands.Cog):
    def __init__(self, bot):
        self.bot = bot
        # 仅当任务未运行时启动
        if not self.my_task.is_running():
            self.my_task.start()

    def cog_unload(self):
        self.my_task.cancel()

    @tasks.loop(time=send_time)
    async def my_task(self):
         # 若another_function是同步函数，建议用异步方式调用避免阻塞事件循环
         # await asyncio.to_thread(another_function)
         another_function()

    @my_task.before_loop
    async def before_printer(self):
        print('waiting...')
        await self.bot.wait_until_ready()

额外提示 #

• 若another_function是同步函数，直接在异步任务中调用可能阻塞机器人事件循环，建议使用await asyncio.to_thread(another_function)来异步执行同步代码。
• 确认服务器时间与代码中设置的UTC+9时区一致，避免任务触发时间异常。

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