Quart生命周期中无法访问Discord Bot实例，集成discord.py、Quart、Celery遇事件循环问题

• 直接报错原因：Celery Worker默认运行同步任务，执行asyncio.get_running_loop()时当前线程不存在已经启动的事件循环，因此抛出RuntimeError。
• 更底层的进程隔离问题：你运行的Quart服务（含discord bot实例）和Celery Worker是完全独立的两个进程，内存空间不共享。你在Celery进程中拿到的app实例不会执行before_serving钩子，因此app.discord_bot属性本身就不存在，就算解决了事件循环的问题也拿不到Quart进程中正在运行的bot实例。

方案1：跨进程HTTP通信（最优） #

不需要在Celery中直接访问bot实例，通过Quart开放内部接口做中转，是最稳妥无冲突的方案。

1. 首先在Quart服务中新增内部接口，封装你需要调用的bot操作：

# 加在create_app函数内
from quart import request

@app.post("/internal/exec-bot-action")
async def exec_bot_action():
    params = await request.get_json()
    action = params.get("action")
    # 按你需要的操作类型分发，举个发消息的例子
    if action == "send_msg":
        channel = app.discord_bot.get_channel(params["channel_id"])
        await channel.send(params["content"])
    # 其他操作同理扩展
    return {"code": 0, "msg": "success"}

2. Celery任务中直接调用该内部接口即可：

# 你的Celery任务示例
import httpx

@celery.task
def send_discord_msg_task(channel_id, content):
    httpx.post("http://127.0.0.1:8000/internal/exec-bot-action", json={
        "action": "send_msg",
        "channel_id": channel_id,
        "content": content
    })

方案2：Celery进程内独立初始化bot（不推荐） #

如果一定要在Celery中直接操作bot，可以手动创建事件循环并在Celery进程内单独初始化bot实例，修改create_celery中的ContextTask代码如下：

class ContextTask(celery.Task):
    def __call__(self, *args, **kwargs):
        # 处理事件循环不存在的问题
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
        
        # Celery进程内单独初始化bot
        if not hasattr(app, "discord_bot"):
            async def init_bot():
                app.discord_bot = commands.Bot(command_prefix=">")
                app.discord_bot.add_cog(MainCog(app.discord_bot))
                await app.discord_bot.login(Config.DISCORD_BOT_TOKEN)
                loop.create_task(app.discord_bot.connect())
            loop.run_until_complete(init_bot())
        
        async def helper():
            async with app.app_context():
                return await self.run(*args, **kwargs)
        return loop.run_until_complete(helper())

注意：该方案会启动两个独立的discord bot实例，二者状态不共享，且可能触发discord的多实例登录限制，仅适合简单测试场景使用。

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