我该如何修复任务已被破坏但正在等待处理?

How can i fix Task was destroyed but it is pending?

我有问题。因此,每当用户在我的 discord 服务器上写聊天消息时,我都会运行一个任务 - 它称为 on_message。所以我的机器人在这个事件中有很多事情要做,我经常得到这种错误:

Task was destroyed but it is pending!
task: <Task pending name='pycord: on_message' coro=<Client._run_event() done, defined at /Bots/gift-bot/discord/client.py:374> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f68a7bdfc10>()]>>

所以我想如果我想解决这个问题,我需要加速我的代码。但遗憾的是,我不知道如何修复此错误。

编辑: 我整合了时间,这就是我打印的内容:

Task was destroyed but it is pending!
task: <Task pending name='pycord: on_message' coro=<Client._run_event() done, defined at /Bots/gift-bot/discord/client.py:374> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f01063f98e0>()]>>
2 if checks done - 7.867813110351562e-06
5 if checks done - 0.0061550140380859375
mysql checks done - 0.010785341262817383
task done - 0.13075661659240723
2 if checks done - 8.344650268554688e-06
5 if checks done - 0.011545896530151367
mysql checks done - 0.02138519287109375
task done - 0.11132025718688965
2 if checks done - 2.0503997802734375e-05
5 if checks done - 0.008122920989990234
mysql checks done - 0.012276411056518555
2 if checks done - 1.0728836059570312e-05
5 if checks done - 0.014346837997436523
mysql checks done - 0.040288448333740234
task done - 0.12520265579223633
2 if checks done - 1.0728836059570312e-05
5 if checks done - 0.0077972412109375
mysql checks done - 0.013320684432983398
task done - 0.1502058506011963
task done - 0.10663175582885742
2 if checks done - 9.775161743164062e-06
5 if checks done - 0.006486177444458008
mysql checks done - 0.011229515075683594
Task was destroyed but it is pending!
task: <Task pending name='pycord: on_message' coro=<Client._run_event() done, defined at /Bots/gift-bot/discord/client.py:374> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f010609a9d0>()]>>
2 if checks done - 6.67572021484375e-06
5 if checks done - 0.0049741268157958984
mysql checks done - 0.008575677871704102
task done - 0.10633635520935059

这是集成计时的代码:

    @commands.Cog.listener("on_message")
    async def on_message(self, message):
        start = time.time()

        # Check ob Nachricht gezählt werden kann


        if message.author.bot:
            return

        if message.type != discord.MessageType.default:
            return
            
        print(f"2 if checks done - {time.time() - start}")

        if isinstance(message.channel, discord.channel.DMChannel):
            return await message.reply(f'Hey {message.author.name}!\nLeider bin ich der falsche Ansprechpartner, falls du Hilfe suchst.. <:pepe_hands:705896495601287320>\nBetrete den https://discord.gg/deutschland Bl4cklist-Discord und sende unserem Support-Bot <@671421220566204446> (`Bl4cklistSupport#7717`) eine Private-Nachricht, damit sich unser Support-Team um dein Problem so schnell es geht kümmern kann. <:pepe_love:759741232443949107>')

        # ENTFERNEN AM 30. APRIL
        prefix_now = await get_prefix(message)
        if message.content.startswith(str(prefix_now)):
            try:
                await message.reply("› <a:alarm:769215249261789185> - **UMSTIEG AUF SLASH-COMMANDS:** Ab **jetzt** laufen alle Befehle dieses Bots auf `/` - um Leistung zu sparen und die Erfahrung zu verbessern. Nutze `/help` um eine Befehlsliste zu sehen.")
            except discord.Forbidden:
                pass
            return

        if self.client.user in message.mentions:

                response = choice([
                "Mit mir kann man die coolsten Gewinnspiele starten! <a:gift:843914342835421185>",
                'Wird Zeit jemanden den Tag zu versüßen! <:smile:774755282618286101>',
                "Wer nicht auf diesem Server ist, hat die Kontrolle über sein Leben verloren! <a:lach_blue2:803693710490861608>",
                "Wann startet endlich ein neues Gewinnspiel? <:whut:848347703217487912>",
                "Ich bin der BESTE Gewinnspiel-Bot - Wer was anderes sagt, lügt! <:wyldekatze:842157727169773608>"
                ])

                try:
                    await message.reply(f"{response} (Mein Präfix: `/`)", mention_author=False)
                except (discord.Forbidden, discord.HTTPException, discord.NotFound):
                    pass
                return
                
        print(f"5 if checks done - {time.time() - start}")


        # Cooldown


        #self.member_cooldown_list = [i for i in self.member_cooldown_list if i[1] + self.cooldown_val > int(time.time())]
        #member_index = next((i for i, v in enumerate(self.member_cooldown_list) if v[0] == message.author.id), None)
        #if member_index is not None:
        #    if self.member_cooldown_list[member_index][1] + self.cooldown_val > int(time.time()):
        #        return

        #self.member_cooldown_list.append((message.author.id, int(time.time())))


        # Rollen-Check (Bonus/Ignore)


        count = 1
        mydb = await getConnection()
        mycursor = await mydb.cursor()
        await mycursor.execute("SELECT ignore_role_id, bonus_role_id FROM guild_role_settings WHERE guild_id = %s", (message.author.guild.id,))
        in_database = await mycursor.fetchone()
        if in_database:
            if in_database[0] is not None:
                role_list = in_database[0].split(" ")
                for roleid in role_list:
                    try:
                        int(roleid)
                    except ValueError:
                        continue

                    role = message.author.guild.get_role(int(roleid))
                    if role is None:
                        continue

                    if role in message.author.roles:
                        await mycursor.close()
                        mydb.close()
                        return

            if in_database[1] is not None:
                role_list = in_database[1].split(" ")
                for roleid in role_list:
                    try:
                        int(roleid)
                    except ValueError:
                        continue

                    role = message.author.guild.get_role(int(roleid))
                    if role is None:
                        continue

                    if role in message.author.roles:
                        count += 1


        # Kanal-Check (Bonus/Ignore)


        await mycursor.execute("SELECT ignore_channel_id FROM guild_channel_settings WHERE guild_id = %s", (message.author.guild.id,))
        in_database1 = await mycursor.fetchone()
        if in_database1:
            if in_database1[0] is not None:
                channel_list = in_database1[0].split(" ")
                for channelid in channel_list:

                    try:
                        int(channelid)
                    except ValueError:
                        continue

                    if int(message.channel.id) == int(channelid):
                        await mycursor.close()
                        mydb.close()
                        return
                        
        print(f"mysql checks done - {time.time() - start}")


        # In Datenbank eintragen

        await mycursor.execute("SELECT * FROM guild_message_count WHERE guild_id = %s AND user_id = %s",
                               (message.author.guild.id, message.author.id))
        in_database2 = await mycursor.fetchone()
        if in_database2:
            await mycursor.execute(
                "UPDATE guild_message_count SET user_id = %s, message_count = message_count + %s WHERE guild_id = %s AND user_id = %s",
                (message.author.id, count, message.author.guild.id, message.author.id))
        else:
            await mycursor.execute(
                "INSERT INTO guild_message_count (user_id, message_count, guild_id) VALUES (%s, %s, %s)",
                (message.author.id, count, message.author.guild.id))

        await mydb.commit()
        await mycursor.close()
        mydb.close()
        
        print(f"task done - {time.time() - start}")

如果我尝试使用 asyncio.run(client.start('token')) 启动我的机器人,我会多次收到此错误:

Ignoring exception in on_guild_channel_delete
Traceback (most recent call last):
  File "/Bots/gift-bot/discord/client.py", line 382, in _run_event
    await coro(*args, **kwargs)
  File "/Bots/gift-bot/cogs/misc_events.py", line 738, in on_guild_channel_delete
    await self.client.wait_until_ready()
  File "/Bots/gift-bot/discord/client.py", line 978, in wait_until_ready
    await self._ready.wait()
  File "/usr/local/lib/python3.9/asyncio/locks.py", line 226, in wait
    await fut
RuntimeError: Task <Task pending name='pycord: on_guild_channel_delete' coro=<Client._run_event() running at /Bots/gift-bot/discord/client.py:382>> got Future <Future pending> attached to a different loop

我在带有 pycord2.0.0b5 的 Debian 10 vServer 上使用 Python3.9。

IODKU 让您消除单独的 SELECT:

await mycursor.execute("SELECT * FROM guild_message_count WHERE guild_id = %s AND user_id = %s", (message.author.guild.id, message.author.id))
        in_database2 = await mycursor.fetchone()
        if in_database2:
            await mycursor.execute("UPDATE guild_message_count SET user_id = %s, message_count = message_count + %s WHERE guild_id = %s AND user_id = %s", (message.author.id, count, message.author.guild.id, message.author.id))
        else:
            await mycursor.execute("INSERT INTO guild_message_count (user_id, message_count, guild_id) VALUES (%s, %s, %s)", (message.author.id, count, message.author.guild.id))

-->

INSERT INTO guild_message_count
        (user_id, message_count, guild_id)
        VALUES 
        (%s, %s, %s)
    ON DUPLICATE KEY UPDATE
        message_count = message_count + 1

您是否使用 autocommit = ON?交易? (那里的选择可能会影响性能;请提供详细信息,以便我们提供建议。)

此错误可能是由于事件循环关闭不当造成的。

此代码演示不正确 关闭。

此处,loop.run_until_complete 方法 returns 任务 t1 完成后,循环关闭,但任务 t2 尚未完成。这会引发错误。

import asyncio


async def delay(n):
    print(f"sleeping for {n} second(s)")
    await asyncio.sleep(n)
    print(f"done sleeping for {n} second(s)")


loop = asyncio.get_event_loop()
t1 = loop.create_task(delay(1))
t2 = loop.create_task(delay(2))
loop.run_until_complete(t1)
loop.close()

输出;

sleeping for 1 second(s)
sleeping for 2 second(s)
done sleeping for 1 second(s)
Task was destroyed but it is pending!
task: <Task pending name='Task-1' coro=<delay() running at aio.py:10> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7fa794c5b970>()]>

正确关机。

此处第二个 loop.run_until_complete 在循环关闭之前等待一组挂起的任务。此行为可确保正确关闭。

import asyncio


async def delay(n):
    print(f"sleeping for {n} second(s)")
    await asyncio.sleep(n)
    print(f"done sleeping for {n} second(s)")


loop = asyncio.get_event_loop()
t1 = loop.create_task(delay(1))
t2 = loop.create_task(delay(2))
loop.run_until_complete(t1)

pending = asyncio.all_tasks(loop=loop)
group = asyncio.gather(*pending)
loop.run_until_complete(group)

loop.close()

最简单的方法是使用 asyncio.run 处理循环的正确关闭。

import asyncio


async def delay(n):
    print(f"sleeping for {n} second(s)")
    await asyncio.sleep(n)
    print(f"done sleeping for {n} second(s)")


async def main():
    t1 = asyncio.create_task(delay(1))
    t2 = asyncio.create_task(delay(2))
    await t2


asyncio.run(main())

对于 pycord 的问题,现在尝试此解决方案,其中 clientdiscord.Client 的实例,即您的 class。我们没有调用 client.run,而是创建了一个 coro 并将其传递给 asyncio.run.

import asyncio

asyncio.run(client.start('token'))

我遇到了一个同样错误的bug,然后我发现了这个https://github.com/python/cpython/pull/29163: task would be garbage collected when there is no strong reference, maybe it is useful to you.

await 表达式阻塞包含协程,直到 awaited 可等待 returns。这阻碍了协程的进展。但是 await 在协程中是必要的,以便将控制权交还给事件循环,以便其他协程可以继续进行。

太多 await 可能会有问题,只会让进度变慢。
我通过将协程方法分解为子任务来重构 on_message 协程方法。

async def _check_channel(self, message, pool):
    async with pool.acquire() as conn:
        async with conn.cursor() as cursor:
            await cursor.execute(
                "SELECT ignore_channel_id FROM guild_channel_settings WHERE guild_id = %s",
                (message.author.guild.id,),
            )
            in_database = await cursor.fetchone()

    if in_database and in_database[0] is not None:
        channel_list = in_database[0].split(" ")
        for channelid in channel_list:

            try:
                channel_id_int = int(channelid)
            except ValueError:
                continue

            if int(message.channel.id) == channel_id_int:
                return False


async def _get_role_count(self, message, pool):
    async with pool.acquire() as conn:
        async with conn.cursor() as cursor:
            await cursor.execute(
                "SELECT ignore_role_id, bonus_role_id FROM guild_role_settings WHERE guild_id = %s",
                (message.author.guild.id,),
            )
            in_database = await cursor.fetchone()
    if in_database:
        first_item, second_item, *_ = in_database
        if first_item is not None:
            role_list = first_item.split(" ")
            for roleid in role_list:
                try:
                    roleid_int = int(roleid)
                except ValueError:
                    continue

                role = message.author.guild.get_role(roleid_int)
                if role is None:
                    continue
                if role in message.author.roles:
                    return False

        if second_item is not None:
            role_list = second_item.split(" ")
            count = 0
            for roleid in role_list:
                try:
                    roleid_int = int(roleid)
                except ValueError:
                    continue

                role = message.author.guild.get_role(roleid_int)
                if role is None:
                    continue
                if role in message.author.roles:
                    count += 1
            return count


@commands.Cog.listener("on_message")
async def on_message(self, message):
    if message.author.bot:
        return
    if message.type != discord.MessageType.default:
        return
    if isinstance(message.channel, discord.channel.DMChannel):
        return

    # Cooldown

    self.member_cooldown_list = [
        i
        for i in self.member_cooldown_list
        if i[1] + self.cooldown_val > int(time.time())
    ]
    member_index = next(
        (
            i
            for i, v in enumerate(self.member_cooldown_list)
            if v[0] == message.author.id
        ),
        None,
    )
    if member_index is not None:
        if self.member_cooldown_list[member_index][1] + self.cooldown_val > int(
            time.time()
        ):
            return

    self.member_cooldown_list.append((message.author.id, int(time.time())))

    loop = asyncio.get_running_loop()
    db_pool = await aiomysql.create_pool(
        minsize=3,
        host="<host>",
        port=3306,
        user="<user>",
        password="<password>",
        db="<db_name>",
        autocommit=False,
        loop=loop,
    )
    count = 1

    check_channel_task = asyncio.create_task(
        self._check_channel(self, message, db_pool)
    )
    role_count_task = asyncio.create_task(self._get_role_count(self, message, db_pool))

    # write to database

    mydb = await db_pool.acquire()
    mycursor = await mydb.cursor()
    await mycursor.execute(
        "SELECT * FROM guild_message_count WHERE guild_id = %s AND user_id = %s",
        (message.author.guild.id, message.author.id),
    )
    in_database = await mycursor.fetchone()

    role_count = await role_count_task
    if False in (role_count, await check_channel_task):
        await mycursor.close()
        db_pool.release(mydb)
        db_pool.close()
        await db_pool.wait_closed()
        return
    if role_count:
        count += role_count
    if in_database:
        await mycursor.execute(
            "INSERT INTO guild_message_count (user_id, message_count, guild_id) VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE message_count = message_count + 1",
            (message.author.id, count, message.author.guild.id),
        )

    await mydb.commit()
    await mycursor.close()
    db_pool.release(mydb)
    db_pool.close()
    await db_pool.wait_closed()

我使用 on_message 方法的一部分代码创建了两个私有异步方法,以实现并发处理。虽然 on_messageawait 中被阻塞,但重构方法可能独立于 on_message 方法进行。为了实现这一点,我从两个新协程中创建了两个任务。 asyncio.create_tasks 将任务安排为 运行,无需 await。这些任务可能 运行 一旦 on_message 将控制权交还给任务创建后任何 await 上的事件循环。

我没有 运行 密码。这是尽力而为。您必须尝试通过移动 await 周围的任务块来进行试验。并且 运行 它与 client.run 以避免 将 Future 附加到不同的循环 错误。