Python asyncio - 使用Task的循环退出已被销毁，但它正在等待处理

Posted 2021-04-28

这是我的python程序的相关代码：

import discord
import asyncio

class Bot(discord.Client):
    def __init__(self):
        super().__init__()

    @asyncio.coroutine
    def my_background_task(self):
        yield from self.wait_until_ready()
        while not self.is_closed:
            yield from asyncio.sleep(3600*24) # <- This is line 76 where it fails
            doSomething()

bot = Bot()
loop = asyncio.get_event_loop()
try:
    loop.create_task(bot.my_background_task())
    loop.run_until_complete(bot.login('username', 'password'))
    loop.run_until_complete(bot.connect())
except Exception:
    loop.run_until_complete(bot.close())
finally:
    loop.close()

该程序偶尔退出（自己，但不应该），除了以外没有其他错误或警告

Task was destroyed but it is pending!
task: <Task pending coro=<my_background_task() running at bin/discordBot.py:76> wait_for=<Future pending cb=[Task._wakeup()]>>

如何确保程序不会随机退出？我在Xubuntu 15.10上有Python 3.4.3+。

答案

这是因为不和谐客户端模块每分钟左右需要控制一次。

这意味着任何窃取控制超过一定时间的函数都会导致discord的客户端进入无效状态（稍后可能会在客户端的下一次方法调用时将其表现为异常）。

要确保discord模块客户端可以ping不和服务器，您应该使用真正的多线程解决方案。

一种解决方案是将所有繁重的处理卸载到一个单独的进程（一个单独的线程不会，因为Python有一个全局解释器锁），并使用discord bot作为一个薄层，其职责是填充工作队列。

相关阅读：https://discordpy.readthedocs.io/en/latest/faq.html#what-does-blocking-mean

示例解决方案......这超出了问题的范围，但我已经编写了大部分代码。如果我有更多时间，我会写一个更短的解决方案:)

2部分，不和谐交互和处理服务器：

这是不和谐的倾听者。

import discord
import re
import asyncio
import traceback

import websockets
import json

# Call a function on other server
async def call(methodName, *args, **kwargs):
    async with websockets.connect('ws://localhost:9001/meow') as websocket:
        payload = json.dumps( {"method":methodName, "args":args, "kwargs": kwargs})
        await websocket.send(payload)
        #...
        resp = await websocket.recv()
        #...
        return resp

client = discord.Client()
tok = open("token.dat").read()

@client.event
async def on_ready():
    print('Logged in as')
    print(client.user.name)
    print(client.user.id)
    print('------')

@client.event
async def on_error(event, *args, **kwargs):
    print("Error?")

@client.event
async def on_message(message):
    try:
        if message.author.id == client.user.id:
            return
        m = re.match("(w+) for (d+).*?", message.content)
        if m:
            g = m.groups(1)
            methodName = g[0]
            someNumber = int(g[1])
            response = await call(methodName, someNumber)
            if response:
                await client.send_message(message.channel, response[0:2000])
    except Exception as e:
        print (e)
        print (traceback.format_exc())

client.run(tok)

这是用于处理大量请求的工作服务器。您可以使此部分同步或异步。

我选择使用一些名为websocket的魔法将数据从一个python进程发送到另一个python进程。但你可以使用你想要的任何东西。例如，您可以使一个脚本将文件写入目录，另一个脚本可以读取文件并处理它们。

import tornado
import tornado.websocket
import tornado.httpserver
import json
import asyncio
import inspect
import time

class Handler:
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def consume(self, text):
        return "You said {0} and I say hiya".format(text)

    async def sweeps(self, len):
        await asyncio.sleep(len)
        return "Slept for {0} seconds asynchronously!".format(len)

    def sleeps(self, len):
        time.sleep(len)
        return "Slept for {0} seconds synchronously!".format(len)


class MyService(Handler, tornado.websocket.WebSocketHandler):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def stop(self):
        Handler.server.stop()

    def open(self):
        print("WebSocket opened")

    def on_message(self, message):
        print (message)
        j = json.loads(message)
        methodName = j["method"]
        args = j.get("args", ())

        method = getattr(self, methodName)
        if inspect.iscoroutinefunction(method):
            loop = asyncio.get_event_loop()
            task = loop.create_task(method(*args))
            task.add_done_callback( lambda res: self.write_message(res.result()))
            future = asyncio.ensure_future(task)

        elif method:
            resp = method(*args)
            self.write_message(resp)

    def on_close(self):
        print("WebSocket closed")

application = tornado.web.Application([
    (r'/meow', MyService),
])

if __name__ == "__main__":
    from tornado.platform.asyncio import AsyncIOMainLoop
    AsyncIOMainLoop().install()

    http_server = tornado.httpserver.HTTPServer(application)
    Handler.server = http_server
    http_server.listen(9001)

    asyncio.get_event_loop().run_forever()

现在，如果您在单独的python脚本中运行这两个进程，并告诉您的机器人“睡眠100”，它将愉快地睡眠100秒！ asyncio stuff用作make-shift工作队列，您可以通过将侦听器作为单独的python脚本运行来正确地将侦听器与后端处理分开。

现在，无论您的功能在“服务器”部分运行多长时间，都不会阻止客户端部分ping不和谐服务器。

图片无法上传，但是......无论如何，这是告诉机器人睡觉和回复的方法......注意睡眠是同步的。 http://i.imgur.com/N4ZPPbB.png

另一答案

我不认为问题发生在asyncio.sleep。无论如何你不应该压制你得到的异常：

bot = Bot()
loop = asyncio.get_event_loop()
try:
    # ...
except Exception as e:
    loop.run_until_complete(bot.close())
    raise e  # <--- reraise exception you got while execution to see it (or log it here)
finally:
    # ...

另一答案

您必须在退出时手动停止任务：

import discord
import asyncio

class Bot(discord.Client):
    def __init__(self):
        super().__init__()

    @asyncio.coroutine
    def my_background_task(self):
        yield from self.wait_until_ready()
        while not self.is_closed:
            yield from asyncio.sleep(3600*24) # <- This is line 76 where it fails
            doSomething()

bot = Bot()
loop = asyncio.get_event_loop()
try:
    task = loop.create_task(bot.my_background_task())
    loop.run_until_complete(bot.login('username', 'password'))
    loop.run_until_complete(bot.connect())
except Exception:
    loop.run_until_complete(bot.close())
finally:
    task.cancel()
    try:
        loop.run_until_complete(task)
    except Exception:
        pass 
    loop.close()