带有计时器和无限循环的Python异步

Question

我在这里找到了我计划的项目的两部分代码。 第一个有两个计时器......

class Timer:
    def __init__(self, interval, first_immediately, timer_name, context, callback):
        self._interval = interval
        self._first_immediately = first_immediately
        self._name = timer_name
        self._context = context
        self._callback = callback
        self._is_first_call = True
        self._ok = True
        self._task = asyncio.ensure_future(self._job())
        print(timer_name + " init done")

    async def _job(self):
        try:
            while self._ok:
                if not self._is_first_call or not self._first_immediately:
                    await asyncio.sleep(self._interval)
                await self._callback(self._name, self._context, self)
                self._is_first_call = False
        except Exception as ex:
            print(ex)

    def cancel(self):
        self._ok = False
        self._task.cancel()

async def timer1(timer_name, context, timer):
    context['count'] += 1
    print('callback: ' + timer_name + ", count: " + str(context['count']))


async def timer2(timer_name, context, timer):
    context['count'] += 1
    print('callback: ' + timer_name + ", count: " + str(context['count']))

t1 = Timer(interval=1, first_immediately=True, timer_name="Timer Channel", context={'count': 0}, callback=timer1)
t2 = Timer(interval=3, first_immediately=False, timer_name="Timer Scroll", context={'count': 0}, callback=timer2)

try:
    loop = asyncio.get_event_loop()
    loop.run_forever()
except KeyboardInterrupt:
    t1.cancel()
    t2.cancel()
    print("clean up done")

第二个带有无限循环来检查事件......

while True:
for key, mask in selector.select():
    device = key.fileobj

    for event in device.read():
        ecode = event.code
        if event.type == evdev.ecodes.EV_KEY:
            match event.code:
                case 114:
                    print("Vol down")
                case 115:
                    print("Vol up")

一切都按其应有的方式工作，但我不知道如何设置它，以便一切独立运行而不干扰其他进程。

致以诚挚的问候

Answer 1

有几件事需要改变。

首先，你从不安排任何事情。 创建一个入口点，例如

async def main()

，您可以在其中开始逻辑（实例化计时器等），然后执行

asyncio.run(main())

。这将创建一个事件循环并安排您的

main

函数。

import asyncio

# entrypoint for everything that needs to run async
async def main():

    try:
        t1 = Timer(interval=1, first_immediately=True, timer_name="Timer Channel", context={'count': 0}, callback=timer1)
        t2 = Timer(interval=3, first_immediately=False, timer_name="Timer Scroll", context={'count': 0}, callback=timer2)

        # some sync/async logic or loop...
        
    except KeyboardInterrupt:
        t1.cancel()
        t2.cancel()
        print("clean up done")

# Best practice to check for '__main__'
if __name__ == '__main__':
    # start the eventloop en schedule your "entrypoint"
    asyncio.run(main())

接下来，您将拥有非异步的异步函数。

timer1

和

timer2

不等待任何内容，因此不需要异步。

# note: I've removed the async keyword
def timer1(timer_name, context, timer):
    context['count'] += 1
    print('callback: ' + timer_name + ", count: " + str(context['count']))

由于这些函数不是异步的，我们不需要等待它们。

旁注：虽然

asyncio.ensure_future

没有错，但文档鼓励使用

asyncio.create_task

代替。请参阅文档。

async def _job(self):
    try:
        while self._ok:
            if not self._is_first_call or not self._first_immediately:
                await asyncio.sleep(self._interval)

            # await self._callback(self._name, self._context, self)
            self._callback(self._name, self._context, self)

            self._is_first_call = False
    except Exception as ex:
        print(ex)

def __init__(self, interval, first_immediately, timer_name, context, callback):
    self._interval = interval
    self._first_immediately = first_immediately
    self._name = timer_name
    self._context = context
    self._callback = callback
    self._is_first_call = True
    self._ok = True
    # self._task = asyncio.ensure_future(self._job())
    self._task = asyncio.create_task(self._job())
    print(timer_name + " init done")

如果您有任何其他问题，或者如果我误解了什么，请随时告诉我！

诗。 https://superfastpython.com/python-asyncio/ 是有关异步的信息的重要来源。