Discord.py音乐机器人快速执行两次play命令返回None崩溃问题

我正在开发一个discord.py音乐机器人，整体运行正常，但碰到个烦人的bug：快速连续执行两次play命令（该命令通过yt_dlp查询YouTube视频）时，第一条命令会因为返回None崩溃，第二条却能正常运行。我对asyncio不太熟，怀疑问题出在asyncio.to_thread()上。

崩溃代码片段 #

@classmethod
async def parse_from(cls, data):
    thread = asyncio.to_thread(ytdl.extract_info, data, download=False)
    result = await thread
    if 'entries' in result: #<= 这里崩溃，因为result是None
        try:
            data = result['entries'][0]
        except IndexError:
            raise commands.UserInputError('未找到任何结果，请尝试使用不同的搜索词。')
    filename = data['url']
    return cls(discord.FFmpegPCMAudio(filename, **ffmpeg_opts), data=data)

报错回溯 #

argument of type 'NoneType' is not iterable
  File "D:\Bureau\CODE\Python\Bot-Discord\cogs\voice.py", line 51, in parse_from
    if 'entries' in data:
  File "D:\Bureau\CODE\Python\Bot-Discord\cogs\voice.py", line 158, in play
    song = await YT_Song.parse_from(song_query)
  File "D:\Bureau\CODE\Python\Bot-Discord\main.py", line 44, in on_message
    await bot.process_commands(message)
  File "D:\Bureau\CODE\Python\Bot-Discord\main.py", line 143, in <module>
    bot.run(config.private_token)
TypeError: argument of type 'NoneType' is not iterable

问题原因 #

yt-dlp的extract_info方法不是线程安全的，如果多个线程共享同一个ytdl实例并发调用该方法，会导致内部状态冲突，其中一个调用会返回None。asyncio.to_thread()每次调用都会启动新线程，当快速连续执行play命令时，两个线程同时操作同一个ytdl实例，就会触发这个问题。

修复方案 #

方案1：为每个请求创建独立的yt-dlp实例

避免全局复用ytdl实例，每次调用parse_from时新建实例，确保线程隔离：

@classmethod
async def parse_from(cls, data):
    # 每次调用都初始化新的yt-dlp实例
    ytdl_instance = yt_dlp.YoutubeDL(ytdl_opts)
    result = await asyncio.to_thread(ytdl_instance.extract_info, data, download=False)
    
    # 先检查result是否为None，避免后续报错
    if result is None:
        raise commands.UserInputError('未找到任何结果，请尝试使用不同的搜索词。')
        
    if 'entries' in result:
        try:
            data = result['entries'][0]
        except IndexError:
            raise commands.UserInputError('未找到任何结果，请尝试使用不同的搜索词。')
    filename = data['url']
    return cls(discord.FFmpegPCMAudio(filename, **ffmpeg_opts), data=data)

方案2：用锁控制并发调用（适合必须复用实例的场景）

如果需要复用ytdl实例来节省资源，可以用asyncio锁确保同一时间只有一个线程调用extract_info：

# 在YT_Song类中定义一个类级别的锁
ytdl_lock = asyncio.Lock()

@classmethod
async def parse_from(cls, data):
    # 使用锁确保串行调用extract_info
    async with cls.ytdl_lock:
        result = await asyncio.to_thread(ytdl.extract_info, data, download=False)
    
    if result is None:
        raise commands.UserInputError('未找到任何结果，请尝试使用不同的搜索词。')
        
    if 'entries' in result:
        try:
            data = result['entries'][0]
        except IndexError:
            raise commands.UserInputError('未找到任何结果，请尝试使用不同的搜索词。')
    filename = data['url']
    return cls(discord.FFmpegPCMAudio(filename, **ffmpeg_opts), data=data)

通用优化：添加None值检查

不管采用哪种方案，都要先对result做None判断，这能大幅提升代码健壮性，避免因返回值异常导致崩溃。

验证效果 #

修改后快速连续执行两次play命令，两次请求都会正常处理，不会再出现第一条命令崩溃的情况。

内容的提问来源于stack exchange，提问作者Daf'ium