discord.py如何持久化存储命令冷却 解决重启后冷却失效问题

原代码失效核心原因 #

• 部分场景下start方法执行时cog尚未完成加载，导致cog内命令的冷却数据未被恢复
• 直接序列化CooldownMapping对象兼容性差，discord.py版本更新后内部结构变更会导致反序列化后状态异常
• 仅在正常调用logout时才持久化数据，进程意外退出时会丢失上次保存后所有的冷却记录

修复后实现逻辑 #

不直接序列化整个CooldownMapping对象，而是提取内部存储的用户/频道/服务器对应冷却过期时间戳进行持久化，兼容性更强，同时增加定时自动保存逻辑避免数据丢失。

import discord
from discord.ext import commands, tasks
import os
import pickle
from collections import defaultdict

cooldown_info_path = "cd.pkl"

class Bot(commands.Bot):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # 加载已保存的冷却数据，key是命令名，value是{桶id: 过期时间戳}
        self.saved_cooldowns = defaultdict(dict)
        if os.path.exists(cooldown_info_path):
            with open(cooldown_info_path, 'rb') as f:
                self.saved_cooldowns = pickle.load(f)
        # 启动定时保存任务，每5分钟存一次，避免意外退出丢数据
        self.auto_save_cooldowns.start()

    async def setup_hook(self):
        # 所有cog加载完成后再恢复冷却数据，避免漏了cog里的命令
        for cmd in self.commands:
            if cmd.name not in self.saved_cooldowns:
                continue
            # 把保存的过期时间写回CooldownMapping的缓存里
            for bucket_id, expire_at in self.saved_cooldowns[cmd.name].items():
                cmd._buckets._cache[bucket_id] = expire_at

    @tasks.loop(minutes=5)
    async def auto_save_cooldowns(self):
        self._save_cooldowns()

    def _save_cooldowns(self):
        # 只提取需要的过期时间，不序列化整个CooldownMapping
        cooldown_data = defaultdict(dict)
        for cmd in self.commands:
            if hasattr(cmd, '_buckets') and hasattr(cmd._buckets, '_cache'):
                cooldown_data[cmd.name] = cmd._buckets._cache.copy()
        with open(cooldown_info_path, 'wb') as f:
            pickle.dump(cooldown_data, f)

    async def logout(self):
        # 退出前手动保存一次
        self._save_cooldowns()
        self.auto_save_cooldowns.cancel()
        return await super().logout()

client = Bot(command_prefix=">", intents=discord.Intents.default()) # 注意intents要根据自己的需求配置

@client.event
async def on_ready():
    print("Ready!")

@client.command()
@commands.cooldown(1, 3600, commands.BucketType.user)
async def hello(ctx):
    await ctx.send("HEY")

class ACog(commands.Cog):
    def __init__(self, client):
        self.bot = client

    @commands.command()
    @commands.cooldown(1, 600, commands.BucketType.user)
    async def ping(self, ctx):
        msg = "Pong {0.author.mention}".format(ctx)
        await ctx.send(msg)

client.add_cog(ACog(client))
# 替换为你自己的机器人token
client.run("your_bot_token")

注意事项 #

• 如果使用discord.py 2.0及以上版本，必须正确配置机器人Intents，否则命令不会触发
• 可根据自己的需求调整auto_save_cooldowns的保存频率，越高频数据丢失风险越小，性能开销也越高
• 如果修改了命令的冷却时间规则，建议删除旧的cd.pkl文件，避免旧数据影响新冷却规则生效

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