정기적 인 간격으로 비동기 iterator를 틱

Question

async for 함께 (주로) 정기적 인 간격으로 새 값을 반환해야하는 비동기 iterator를 구현하고 있습니다.

import asyncio 

class Clock(object): 
    def __init__(self, interval=1): 
     self.counter = 0 
     self.interval = interval 
     self.tick = asyncio.Event() 
     asyncio.ensure_future(self.tick_tock()) 

    async def tick_tock(self): 
     while True: 
      self.tick.clear() 
      await asyncio.sleep(self.interval) 
      self.counter = self.__next__() 
      self.tick.set() 

    def __next__(self): 
     self.counter += 1 
     return self.counter 

    def __aiter__(self): 
     return self 

    async def __anext__(self): 
     await self.tick.wait() 
     return self.counter 

이 asyncio.Event를 사용하는 것보다 더 나은 또는 청소기 방법이 있나요 :

우리는 n 초 ~ 모든 카운터를 증가하는 간단한 시계와 같은 반복자를 설명 할 수 있습니까? 이 반복자에서 두 개 이상의 코 루틴이 async for입니다.

Answer 1

내 의견으로는 귀하의 접근 방식이 좋습니다. 파이썬 3.6 이후, 당신은 또한 asynchronous generators을 사용할 수 있습니다 :

async def clock(start=0, step=1, interval=1.): 
    for i in count(start, step): 
     yield i 
     await asyncio.sleep(interval) 

그러나, 여러 코 루틴 사이에 공유 할 수 없습니다. 작업에서 시계를 실행하고 비동기 반복 인터페이스를 통해 데이터를 사용할 수있게해야합니다. 이는 본질적으로 코드에서 수행 한 작업입니다. 여기에 possible implementation이 있습니다.

Answer 2

파이썬 3.6 이상을 사용하는 경우 더 읽기 쉬운 asynchronous generators을 사용할 수 있습니다.

async def Clock(interval=1): 
    counter = 0 
    while True: 
     await asyncio.sleep(interval) 
     counter += 1 
     yield counter 


async def main(): 
    async for i in Clock(1): 
     print(i) 
     if i == 4: 
      break 


if __name__ == '__main__': 
    loop = asyncio.get_event_loop() 
    try: 
     loop.run_until_complete(main()) 
    finally: 
     loop.run_until_complete(loop.shutdown_asyncgens()) 
     loop.close()