python - 使用 threading.Timer 与基于异步的函数

Question

我正在使用 Python 开发一个 Discord 机器人，它对来自 YouTube 的音乐进行排队，我正在研究一些在播放器停止时自动排队歌曲的方法，虽然所有代码都运行良好，但唯一的问题是我不能够每 15 秒检查一次玩家是否正在玩

    async def cmd_autoqueue(self,message, player,channel,author, permissions, leftover_args):

     print("autoq ran")
     if started == True:

       if player.is_stopped:
         await self.cmd_autoqadd(player, channel, author, permissions,leftover_args,song_url=last_url)
       threading.Timer(15.0,await self.cmd_autoqueue(message, player,channel,author, permissions, leftover_args)).start()  

我确实意识到

threading.Timer(15.0,await self.cmd_autoqueue(message, player,channel,author, permissions, leftover_args)).start()

调用该函数，如果我想将其作为稍后调用的内容传递，我会使用 lambda:但是，异步 lambda？

也开始 bool 值由其他东西管理，所以它在那里是为了“如果”，在这个问题中

解决方案:

    async def cmd_autoqueue(self,message, player,channel,author, permissions, leftover_args):
    global started
    print("loop")
    if started == True:
       await asyncio.sleep(15) 
       if player.is_stopped:
         await self.cmd_autoqadd(player, channel, author, permissions,leftover_args,song_url=last_url)
       await self.cmd_autoqueue(message, player,channel,author, permissions, leftover_args)

Answer 1

您可以使用loop.call_later，它返回一个与异步兼容的处理程序来取消任务。

你的类(class): def init(自身): self._loop = asyncio.get_event_loop() self._check_handler = 无

def schedule_check(self):
    self._check_handler = loop.call_later(
        15 * 60,  # Every 15 minutes
        self._loop.create_task, self._periodic_check())

def stop_checking(self):
    self._check_handler.cancel()    

async def start_player(self):
    await do something()
    if not check_handler:
        schedule_check()

async def stop_player(self):
    await do something()
    self.stop_checking()

async def _periodic_check(self):
    await do_something()
    self._schedule_check()

要将一个函数(需要一些参数)作为参数传递给另一个不允许传递参数的函数，您可以使用 functools.partial 绑定(bind)参数。