如何让discord.py音乐队列实现Guild隔离，避免跨服务器共享？

问题背景 #

当前音乐机器人使用全局的queuee和songs列表存储队列，导致不同服务器（Guild）的队列互相共享，需要实现服务器级别的队列隔离，且不想使用数据库方案，寻求替代方法。

当前有问题的核心代码：

queuee = []
songs = []

@client.hybrid_command(
    name="play",
    with_app_command=True,
    description="Play a song in a voice channel with a url or keyword",
)
@commands.is_owner()
async def play(ctx, *, query):
    await ctx.send(f"Searching for {query.capitalize()}..")
    voice_channel = ctx.author.voice.channel
    vc = await voice_channel.connect()
    loop = asyncio.get_event_loop()
    data = await loop.run_in_executor(
        None, lambda: ytdl.extract_info(query, download=False)
    )
    if "entries" in data:
        video_format = data["entries"][0]["formats"][0]
    elif "formats" in data:
        video_format = data["formats"][0]
    song = video_format["url"]
    player = discord.FFmpegPCMAudio(song, **ffmpeg_options)
    vc.play(player, after=do_after)
    queuee.append(song)
    songs.append(query.capitalize())

替代数据库的解决方案 #

1. 使用字典映射服务器ID到队列（最简洁的内存方案） #

把全局列表改成以**服务器ID（ctx.guild.id）**为键的字典，每个键对应该服务器的专属队列列表。这样不同服务器的队列完全隔离，数据存在内存中，重启机器人后会丢失，但适合不需要持久化的场景。

修改后的代码示例：

# 初始化字典，键是guild_id，值是对应服务器的音频URL队列
queues = {}
# 存储各服务器的歌曲名称列表
song_names = {}

@client.hybrid_command(
    name="play",
    with_app_command=True,
    description="Play a song in a voice channel with a url or keyword",
)
@commands.is_owner()
async def play(ctx, *, query):
    guild_id = ctx.guild.id
    # 若当前服务器无队列，初始化空列表
    if guild_id not in queues:
        queues[guild_id] = []
    if guild_id not in song_names:
        song_names[guild_id] = []
    
    await ctx.send(f"Searching for {query.capitalize()}..")
    voice_channel = ctx.author.voice.channel
    vc = await voice_channel.connect()
    loop = asyncio.get_event_loop()
    data = await loop.run_in_executor(
        None, lambda: ytdl.extract_info(query, download=False)
    )
    if "entries" in data:
        video_format = data["entries"][0]["formats"][0]
    elif "formats" in data:
        video_format = data["formats"][0]
    song = video_format["url"]
    player = discord.FFmpegPCMAudio(song, **ffmpeg_options)
    vc.play(player, after=do_after)
    
    # 将歌曲添加到当前服务器的专属队列
    queues[guild_id].append(song)
    song_names[guild_id].append(query.capitalize())

2. 封装成GuildPlayer类（更结构化的方案） #

如果后续需要扩展更多服务器专属功能（比如当前播放状态、音量控制等），可以创建类封装每个服务器的播放器和队列，用字典管理这些实例，代码结构更清晰。

示例代码：

class GuildPlayer:
    def __init__(self):
        self.queue = []
        self.song_names = []
        self.vc = None  # 存储当前服务器的语音连接实例
        self.current_player = None

# 存储各服务器的播放器实例，键是guild_id
guild_players = {}

@client.hybrid_command(
    name="play",
    with_app_command=True,
    description="Play a song in a voice channel with a url or keyword",
)
@commands.is_owner()
async def play(ctx, *, query):
    guild_id = ctx.guild.id
    # 初始化当前服务器的播放器实例
    if guild_id not in guild_players:
        guild_players[guild_id] = GuildPlayer()
    
    player = guild_players[guild_id]
    await ctx.send(f"Searching for {query.capitalize()}..")
    voice_channel = ctx.author.voice.channel
    player.vc = await voice_channel.connect()
    loop = asyncio.get_event_loop()
    data = await loop.run_in_executor(
        None, lambda: ytdl.extract_info(query, download=False)
    )
    if "entries" in data:
        video_format = data["entries"][0]["formats"][0]
    elif "formats" in data:
        video_format = data["formats"][0]
    song = video_format["url"]
    player.current_player = discord.FFmpegPCMAudio(song, **ffmpeg_options)
    player.vc.play(player.current_player, after=do_after)
    
    # 添加到当前服务器的队列
    player.queue.append(song)
    player.song_names.append(query.capitalize())

3. 使用discord.py的Guild对象属性（可选） #

可以给每个Guild对象动态添加自定义属性存储队列，但要注意discord.py的对象可能会被重新实例化，导致属性丢失，适合简单场景：

@client.hybrid_command(
    name="play",
    with_app_command=True,
    description="Play a song in a voice channel with a url or keyword",
)
@commands.is_owner()
async def play(ctx, *, query):
    guild = ctx.guild
    # 初始化队列属性
    if not hasattr(guild, 'queue'):
        setattr(guild, 'queue', [])
    if not hasattr(guild, 'song_names'):
        setattr(guild, 'song_names', [])
    
    await ctx.send(f"Searching for {query.capitalize()}..")
    voice_channel = ctx.author.voice.channel
    vc = await voice_channel.connect()
    loop = asyncio.get_event_loop()
    data = await loop.run_in_executor(
        None, lambda: ytdl.extract_info(query, download=False)
    )
    if "entries" in data:
        video_format = data["entries"][0]["formats"][0]
    elif "formats" in data:
        video_format = data["formats"][0]
    song = video_format["url"]
    player = discord.FFmpegPCMAudio(song, **ffmpeg_options)
    vc.play(player, after=do_after)
    
    guild.queue.append(song)
    guild.song_names.append(query.capitalize())

注意事项 #

• 以上方案均为内存存储，机器人重启后队列数据会丢失，如果需要持久化，仍需使用数据库（如SQLite、MongoDB）或本地文件（如定期保存JSON）。
• 回调函数do_after也要同步修改，确保从当前服务器的队列中取出下一首播放，避免跨服务器调用队列。

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