我该如何修复任务已被破坏但仍处于待处理状态？

Question

我有一个问题。因此，每次用户在我的 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> (`Bl4cklist🔥Support#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

我在 Debian 10 vServer 上使用 Python3.9 和 pycord2.0.0b5。

Answer 1

await

表达式会阻塞包含的协程，直到

await

ed等待返回。这阻碍了协程的进展。但是

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
    check_channel = await check_channel_task
    if False in (role_count, check_channel):
        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_message

被阻塞在

await

中，但重构方法可能会独立于

on_message

方法而进行。为了实现这一点，我从两个新的协程中创建了两个任务。

asyncio.create_tasks

安排要运行的任务，无需

await

。一旦

on_message

将控制权交回任务创建后的任何

await

上的事件循环，这些任务就可以运行。

我没有运行代码。这是最好的努力。您必须尝试通过移动任务周围的块来进行实验。并且还可以使用

await

运行它，以避免

将 Future 连接到不同的循环

错误。

Answer 2

https://github.com/python/cpython/pull/29163

：当没有强引用时任务将被垃圾收集，也许对你有用.

Answer 3

此代码演示了

不正确的

关闭。 这里，当任务

client.run

完成时，一旦

loop.run_until_complete

方法返回，循环就会关闭，但任务

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

是

client

的实例，即您的班级。我们不调用

discord.Client

，而是创建一个 coro 并将其传递给

client.run

。asyncio.run

Answer 4

（至少对我来说）：

尝试将代码部分移动到异步函数中

并调用该函数，特别是如果您将来可能再次需要它们或者它们需要花费大量时间 - 这在大多数情况下解决了我的错误。

IODKU 让您消除单独的

import asyncio

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

Answer 5

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))

您是否使用自动提交 = ON？交易？ （那里的选择可能会影响性能；请提供详细信息，以便我们提供建议。）