主页 / 问题 / 1518250
Accepted
Justiks
Asked:2023-05-08 01:53:57 +0800 CST2023-05-08 01:53:57 +0800 CST

python 异步定时器不工作

  • 772
from asyncio import create_task, get_event_loop, sleep


class Timer:
    def __init__(self, color: bool):
        self.seconds: int = 900
        self.expires: bool = color
        self.color = lambda x: "белых" if color else "чёрных"
        get_event_loop().run_until_complete(self.start_timer())

    async def start_timer(self):
        """Запускает цикл событий"""
        
        await create_task(self.update_timer())

    async def update_timer(self):
        """Ведёт время отсчёта"""

        while self.seconds:
            if self.expires:
                self.seconds -= 1
                await sleep(1)

        assert self.seconds, f"Закончилось время у {self.color}. Противоположный игрок победил"

    def flip_the_timer(self):
        """Меняет положение таймера (активный/деактивный)"""

        self.expires = not self.expires

    def get_timeset(self) -> str:
        """:return время в виде таймера на электронных часах"""

        return f"{round(self.seconds / 60)}:{self.seconds % 60}"

上面的问题代码

一直有异步的困扰，但总是悄悄突破，这周却遇上了某种杀手case

计时器启动，我没有捕获到错误，但即使由于异步，代码也没有更进一步，但计时器愚蠢地等待，而第二个计时器应该竞争性地启动，但它没有出现

我创建了一个任务，并在我没有尝试解决问题时立即启动它。

对异步比较友好的各位程序员，请帮忙！

python

1 个回答

  1. Best Answer

    下面是一个使用 asyncio.loop 的例子：

    from asyncio import sleep, new_event_loop


async def task1():
    for i in range(15):
        print(f'task {i}')
        await sleep(1)


async def main():
    print('main1')
    await sleep(2)
    print('main2')
    await sleep(2)
    print('main3')
    await sleep(2)

loop = new_event_loop()

task_main = loop.create_task(main())
task_1 = loop.create_task(task1())

loop.run_until_complete(task_1)

    结果：

    main1
task 0
task 1
main2
task 2
task 3
main3
task 4
task 5
task 6
task 7
task 8
task 9
task 10
task 11
task 12
task 13
task 14

Process finished with exit code 0

    但就您而言，我认为线程更好：

    import time
from threading import Thread


class Timer:
    def __init__(self, color: bool):
        self.seconds: int = 900
        self.expires: bool = color
        self.color = lambda x: "белых" if color else "чёрных"
        self.thread = Thread(target=self.update_timer)
        self.thread.start()

    def update_timer(self):
        """Ведёт время отсчёта"""

        while self.seconds:
            if self.expires:
                self.seconds -= 1
                time.sleep(1)

        assert self.seconds, f"Закончилось время у {self.color}. Противоположный игрок победил"

    def flip_the_timer(self):
        """Меняет положение таймера (активный/деактивный)"""

        self.expires = not self.expires

    def get_timeset(self) -> str:
        """:return время в виде таймера на электронных часах"""

        return f"{round(self.seconds / 60)}:{self.seconds % 60}"


timer = Timer(True)
print('Do something....')
    • 0

相关问题