Python异步方法创建新线程最优方案：解决Discord Bot无响应问题

我正在开发Discord Bot，需要在执行耗时几秒到一分钟的任务时，仍能正常响应其他指令。目前用asyncio.get_event_loop().create_task调用异步方法，但Bot在queue_system.py代码运行期间会失去响应。需要找到从异步方法创建新线程的最优实现方案，以下是精简后的代码示例：

class StableCog(commands.Cog, name='Stable Diffusion', description='Create images from natural language.'):
    def __init__(self, bot):
        self.text2image_model = Text2Image()
        self.bot = bot

    @commands.slash_command(description='Create an image.')
    async def dream(self, ctx, -- 大量参数 -- ):
        print(f'Request -- {ctx.author.name}#{ctx.author.discriminator} -- Prompt: {query}')
        asyncio.get_event_loop().create_task(src.bot.queue_system.dream_async(ctx, -- 大量参数 -- ))

queue_system.py中的代码：

async def dream_async(ctx, -- 大量参数 --):
    await ctx.interaction.response.send_message('Added to queue! You are # in queue')
    embed = discord.Embed()
    try:
        # 此处为耗时的图片生成等代码，已省略
        await ctx.channel.send(embed=embed, file=discord.File(fp=buffer, filename=f'{seed}.png'))

    except Exception as e:
        embed = discord.Embed(title='txt2img failed', description=f'{e}\n{traceback.print_exc()}', color=embed_color)
        await ctx.channel.send(embed=embed)

注：queue_system.py目前并非完整的队列系统，先解决异步运行dream方法的问题。

问题根源 #

你当前的dream_async是异步函数，但如果其中的耗时任务（比如图片生成）是同步阻塞的IO/CPU密集型操作，就算用create_task也会卡住整个事件循环——因为asyncio是单线程模型，同步阻塞代码会占满事件循环，导致其他指令无法被处理。

最优实现方案：用线程池异步执行同步任务 #

1. 拆分耗时同步代码 #

把dream_async里的耗时逻辑抽成单独的同步函数，隔离阻塞操作：

# 同步的图片生成函数，所有耗时操作都放在这里
def generate_image(prompt, seed, **kwargs):
    # 原有的模型加载、图片生成、buffer处理等耗时代码
    buffer = ...
    return buffer, seed

2. 在异步函数中调用线程执行同步任务 #

修改queue_system.py的dream_async，用asyncio.to_thread将同步耗时任务丢到单独线程执行，不阻塞事件循环：

import asyncio
import traceback
import discord

async def dream_async(ctx, query, seed, ...):
    await ctx.interaction.response.send_message('Added to queue! You are # in queue')
    embed = discord.Embed()
    try:
        # 用to_thread把同步任务放到线程池，不占用事件循环
        buffer, seed = await asyncio.to_thread(generate_image, query, seed, ...)
        await ctx.channel.send(embed=embed, file=discord.File(fp=buffer, filename=f'{seed}.png'))
    except Exception as e:
        embed = discord.Embed(title='txt2img failed', description=f'{e}\n{traceback.format_exc()}', color=0xFF0000)
        await ctx.channel.send(embed=embed)

3. 原dream命令的调用保持不变 #

你的dream命令里用create_task启动dream_async的逻辑是正确的，这样命令能立刻返回，不会卡住用户交互：

@commands.slash_command(description='Create an image.')
async def dream(self, ctx, ...):
    print(f'Request -- {ctx.author.name}#{ctx.author.discriminator} -- Prompt: {query}')
    asyncio.create_task(src.bot.queue_system.dream_async(ctx, query, seed, ...))
    # 无需await，create_task会将任务丢到事件循环后台执行

补充：CPU密集型任务的自定义线程池 #

如果图片生成是CPU密集型操作（比如本地Stable Diffusion推理），可以自定义线程池控制并发数，避免线程过多耗尽资源：

from concurrent.futures import ThreadPoolExecutor

# 根据CPU核心数设置线程数，比如4个线程
executor = ThreadPoolExecutor(max_workers=4)

async def dream_async(ctx, query, seed, ...):
    await ctx.interaction.response.send_message('Added to queue! You are # in queue')
    embed = discord.Embed()
    try:
        # 使用自定义线程池执行同步任务
        buffer, seed = await asyncio.get_event_loop().run_in_executor(executor, generate_image, query, seed, ...)
        await ctx.channel.send(embed=embed, file=discord.File(fp=buffer, filename=f'{seed}.png'))
    except Exception as e:
        embed = discord.Embed(title='txt2img failed', description=f'{e}\n{traceback.format_exc()}', color=0xFF0000)
        await ctx.channel.send(embed=embed)

关键注意事项 #

• 禁止在同步线程函数中调用Discord API（比如ctx.channel.send），所有Discord操作必须在异步代码中通过await执行。
• 后续实现队列系统时，需控制任务并发数，避免同时启动过多生成任务导致资源耗尽。

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