Python开发Discord机器人时无法从线程函数触发异步函数问题

根因分析 #

1. RuntimeError: no running event loop报错和协程未等待警告的核心原因：asyncio的事件循环是线程本地的，你在tasker的独立线程中直接调用asyncio.create_task，当前线程没有正在运行的事件循环，协程无法被调度执行。
2. 不开启独立线程时Helper卡顿的原因：requests是同步IO库，执行请求时会阻塞整个asyncio事件循环，导致Discord机器人的所有异步任务都无法运行。

最小修改方案 #

不需要改动原有架构，只要调整3个核心点即可：

1. 直接使用Discord Client自带的loop属性，该属性绑定了主线程正在运行的事件循环
2. 跨线程提交协程任务时，用线程安全的asyncio.run_coroutine_threadsafe方法替代asyncio.create_task
3. 把tasker的工作线程设为守护线程，避免程序退出时线程卡住

完整修改后代码 #

import requests
import asyncio
import discord
from discord.ext import commands, tasks
from datetime import datetime, timedelta
import threading
import dateutil.parser # 原代码漏了该依赖导入，补上

discordbot = None
taskervar = None

class Helper(discord.Client):
    async def on_ready(self):
        global discordbot, taskervar
        servername = 'ServerName'
        discordbot = self
        self.servidores = dict()
        self.canais = dict()
        for i in range(len(self.guilds)):
            self.servidores[self.guilds[i].name] = {}
            self.servidores[self.guilds[i].name]['guild']=self.guilds[i]
            servidor = self.guilds[i]
            for k in range(len(servidor.channels)):
                canal = servidor.channels[k]
                self.canais[str(canal.name)] = canal
            if 'bottalk' not in self.canais.keys():
                newchan = await self.servidores[self.guilds[i].name]['guild'].create_text_channel('bottalk')
                self.canais[str(newchan.name)] = newchan
            self.servidores[self.guilds[i].name]['canais'] = self.canais
        self.bottalk = self.get_channel(self.servidores[servername]['canais']['bottalk'].id)
        await self.msg("Bot online: " + converteHora(datetime.now(),True))
        print(f'{self.user} has connected to Discord!')
        taskervar.startprocess()
    async def msg(self, msg):
        await self.bottalk.send(msg)
    async def on_message(self, message):
        if message.author == self.user:
            return
        else:
            print(message)

class tasker:
    def __init__(self):
        global discordbot, taskervar
        print('Tasker start')
        taskervar = self
        self.waiter = threading.Event()
        self.lastupdate = datetime.now()
        self.nextupdate = datetime.now()
        # 加daemon=True设为守护线程
        self.thread = threading.Thread(target=self.requests, daemon=True)
    def startprocess(self):
        if not self.thread.is_alive():
            self.waiter = threading.Event()
            self.interval = 60*5
            # 加daemon=True设为守护线程
            self.thread = threading.Thread(target=self.requests, daemon=True)
            self.thread.start()
    def requests(self):
        while not self.waiter.is_set():
            getInfo()
            self.lastupdate = datetime.now()
            self.nextupdate = datetime.now()+timedelta(seconds=self.interval)
            self.waiter.wait(self.interval)
    def stopprocess(self):
        self.waiter.set()

class getInfo:
    def __init__(self):
        global discordbot, taskervar
        self.requests()
    async def discordmsg(self,msg):
        await discordbot.msg(msg)
    def requests(self):
        jsondata = {"TestStatus": 1}
        if  jsondata['TestStatus'] == 1:
            print('here')
            # 替换为run_coroutine_threadsafe跨线程提交协程
            asyncio.run_coroutine_threadsafe(self.discordmsg("SOMETHING WENT WRONG"), loop=discordbot.loop)
            taskervar.stopprocess()
            return
        elif jsondata['TestStatus'] == 2:
            print('test')
            hora = converteHora(datetime.now(),True)
            # 替换为run_coroutine_threadsafe跨线程提交协程
            asyncio.run_coroutine_threadsafe(self.discordmsg(str("Everything is fine but not now: " + hora )), loop=discordbot.loop)
            print('test2')

            
def converteHora(dateUTC, current=False):
    if current:
        response = (dateUTC.strftime("%d/%m/%Y, %H:%M:%S"))
    else:
        response = (dateutil.parser.isoparse(dateUTC)-timedelta(hours=3)).strftime("%d/%m/%Y, %H:%M:%S")
    return response

async def main():
    TOKEN = 'TOKEN GOES HERE'
    tasker()
    # 修正gather的写法，不需要提前await start
    await asyncio.gather(
        Helper().start(TOKEN)
    )


if __name__ == '__main__':
    asyncio.run(main())

可选优化 #

如果后续不想维护独立线程，可以把requests替换为异步HTTP库aiohttp，直接把定时请求做成Discord的后台异步任务，完全不需要线程通信的逻辑。

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