python——asyncio模块实现协程、异步编程(三)
【八】协程停止
future对象有几个状态:
Pending
Running
Done
Cancelled
创建future的时候,task为pending,事件循环调用执行的时候当然就是running,调用完毕自然就是done,如果需要停止事件循环,就需要先把task取消。可以使用asyncio.Task获取事件循环的task
例如:
[python] view plain copy
- import asyncio
- import time
- now = lambda: time.time()
- async def do_some_work(x):
- print('Waiting: ', x)
- await asyncio.sleep(x)
- return 'Done after {}s'.format(x)
- coroutine1 = do_some_work(1)
- coroutine2 = do_some_work(2)
- coroutine3 = do_some_work(4)
- tasks = [
- asyncio.ensure_future(coroutine1),
- asyncio.ensure_future(coroutine2),
- asyncio.ensure_future(coroutine3)
- ]
- start = now()
- loop = asyncio.get_event_loop()
- try:
- loop.run_until_complete(asyncio.wait(tasks))
- except KeyboardInterrupt as e:
- print(asyncio.Task.all_tasks())
- for task in asyncio.Task.all_tasks():
- print(task.cancel())
- loop.stop()
- loop.run_forever()
- finally:
- loop.close()
- print('TIME: ', now() - start)
启动事件循环之后,马上ctrl+c,会触发run_until_complete的执行异常 KeyBorardInterrupt。然后通过循环asyncio.Task取消future
True表示cannel成功,loop stop之后还需要再次开启事件循环,最后在close,不然会报错。
循环task,逐个cancel是一种方案,可是正如上面我们把task的列表封装在main函数中,main函数外进行事件循环的调用。这个时候,main相当于最外出的一个task,那么处理包装的main函数即可。
[python] view plain copy
- import asyncio
- import time
- now = lambda: time.time()
- async def do_some_work(x):
- print('Waiting: ', x)
- await asyncio.sleep(x)
- return 'Done after {}s'.format(x)
- async def main():
- coroutine1 = do_some_work(1)
- coroutine2 = do_some_work(2)
- coroutine3 = do_some_work(4)
- tasks = [
- asyncio.ensure_future(coroutine1),
- asyncio.ensure_future(coroutine2),
- asyncio.ensure_future(coroutine3)
- ]
- done, pending = await asyncio.wait(tasks)
- for task in done:
- print('Task ret: ', task.result())
- start = now()
- loop = asyncio.get_event_loop()
- task = asyncio.ensure_future(main())
- try:
- loop.run_until_complete(task)
- except KeyboardInterrupt as e:
- print(asyncio.Task.all_tasks())
- print(asyncio.gather(*asyncio.Task.all_tasks()).cancel())
- loop.stop()
- loop.run_forever()
- finally:
- loop.close()
【九】不同线程的事件循环
很多时候,我们的事件循环用于注册协程,而有的协程需要动态的添加到事件循环中。一个简单的方式就是使用多线程。当前线程创建一个事件循环,然后在新建一个线程,在新线程中启动事件循环。当前线程不会被block。
[python] view plain copy
- import asyncio
- import time
- now = lambda: time.time()
- from threading import Thread
- def start_loop(loop):
- asyncio.set_event_loop(loop)
- loop.run_forever()
- def more_work(x):
- print('More work {}'.format(x))
- time.sleep(x)
- print('Finished more work {}'.format(x))
- start = now()
- new_loop = asyncio.new_event_loop()
- t = Thread(target=start_loop, args=(new_loop,))
- t.start()
- print('TIME: {}'.format(time.time() - start))
- new_loop.call_soon_threadsafe(more_work, 6)
- new_loop.call_soon_threadsafe(more_work, 3)
启动上述代码之后,当前线程不会被block,新线程中会按照顺序执行call_soon_threadsafe方法注册的more_work方法,后者因为time.sleep操作是同步阻塞的,因此运行完毕more_work需要大致6 + 3
【十】新线程协程
新线程协程的话,可以在主线程中创建一个new_loop,然后在另外的子线程中开启一个无限事件循环。主线程通过run_coroutine_threadsafe新注册协程对象。这样就能在子线程中进行事件循环的并发操作,同时主线程又不会被block。一共执行的时间大概在6s左右。
[python] view plain copy
- import asyncio
- import time
- now = lambda: time.time()
- from threading import Thread
- def start_loop(loop):
- asyncio.set_event_loop(loop)
- loop.run_forever()
- async def do_some_work(x):
- print('Waiting {}'.format(x))
- await asyncio.sleep(x)
- print('Done after {}s'.format(x))
- def more_work(x):
- print('More work {}'.format(x))
- time.sleep(x)
- print('Finished more work {}'.format(x))
- start = now()
- new_loop = asyncio.new_event_loop()
- t = Thread(target=start_loop, args=(new_loop,))
- t.start()
- print('TIME: {}'.format(time.time() - start))
- asyncio.run_coroutine_threadsafe(do_some_work(6), new_loop)
- asyncio.run_coroutine_threadsafe(do_some_work(4), new_loop)