目次
Outline
asyncio
import time
import asyncio
from logging import StreamHandler, Formatter, getLogger, INFO, DEBUG
class AsyncLoopNest():
def __init__(self):
self.loop = asyncio.get_event_loop()
self.running = True
sh = StreamHandler()
sh.setLevel(DEBUG)
sh.setFormatter(
Formatter("%(asctime)s [%(threadName)s,%(thread)d][%(processName)s,%(process)d][%(module)s,%(funcName)s,%(lineno)d] %(message)s"))
self.logger = getLogger()
self.logger.setLevel(DEBUG)
self.logger.addHandler(sh)
def parent_loop(self, sec: int):
self.logger.debug(f' --parent loop start: wait {sec}sec--')
for i in range(sec):
self.logger.debug(f' --parent call {i}')
self.loop.run_in_executor(None, self.chiled_loop, i)
self.logger.info(f' --parent loop stop--')
def chiled_loop(self, child_num: int):
self.logger.debug(f' --child loop start: No. {child_num}--')
i = 0
while self.running == True:
self.logger.debug(f' --child process {child_num},{i}')
time.sleep(1)
i = i + 1
self.logger(f' --child loop stop--')
if __name__ == '__main__':
tc = AsyncLoopNest()
try:
tc.logger.debug(f'===main start===')
tc.loop.run_in_executor(None, tc.parent_loop, 3)
tc.logger.debug(f'===main some process start===')
for i in range(10):
tc.logger.debug(f'main some process {i}')
time.sleep(0.3)
tc.logger.debug(f'===main some process end===')
tc.logger.debug(f'===main end===')
except KeyboardInterrupt:
tc.logger.debug('Ctrl+C')
finally:
tc.running = False
tc.loop.call_soon(tc.loop.stop)
tc.loop.close
tc.logger.debug('end')
この記事を書いた人 Wrote this article
