aysncio

Posted 2021-01-19 xinghun85

event_loop

一个线程只有一个event_loop

1、get_event_loop() 只能在主线程中创建event_loop，子线程的loop要在主线程通过new_event_loop创建，然后在子线程中使用set_event_loop来设置，如果 在子线程中直接用get_event_loop创建就会报错

RuntimeError: There is no current event loop in thread ‘Thread-1‘.

2、new_event_loop(),创建另一个event_loop，与get_event_loop不是同一个。

3、set_event_loop(loop)，为线程设置一个event_loop为loop，无返回值.

 

wait、 gather

gather(*task),以列表返回异步函数的返回结果

wait(task,timeout)可设置超时时间，分别返回的是完成的futures，和未完成的futures，如果要结果，再通过future.result()来获取。

 

run_coroutine_threadsafe、wrap_future、call_soon_threadsafe

loop.run_until_complete()，任务完成就停止循环

loop.run_forever() ，loop一直运行，直到loop.stop才会停止。loop.close()也不行。

loop.call_soon_threadsafe(),任务完成后调用函数

asyncio.run_coroutine_threadsafe(coroutine,loop)，以线程安全的方式运行协程，可实现跨线程调用协程。

asyncio.wrap_future(),创建个新的future与原future连接，只要其中一个完成，另一个也完成。

代码如下

 1 main_loop = asyncio.get_event_loop()       #主线程 创建主循环
 2 sub_loop = asyncio.new_event_loop()        #创建另一个新的循环，用于子线程
 3 task = [asyncio.wrap_future(asyncio.run_coroutine_threadsafe(do_some_work(i, randint(0,4)),sub_loop)) for i in range(4)]  
　　#创建任务，将异步程序用asyncio.run_coroutine_threadsafe()函数包装成future，再用asyncio.wrap_future包装成新的future，新future会连到主循环中，所以就可以在主循环中使用run_until_complete来等待子循环完成
 4 th = threading.Thread(target=sub_loop.run_forever)  #创建子线程，启动循环
 5 th.start()
 6 print("thread",threading.enumerate())
 7 main_loop.run_until_complete(asyncio.wait(task))  #主循环，直到任务完成结束
 8 sub_loop.call_soon_threadsafe(sub_loop.stop)      #子循环调用stop，停止异步
 9 th.join()　　　　　　　　　　　　　　　　　　　　　　　　　# 线程结束
10 print("End thread",threading.enumerate())

 

输出：

thread [<Thread(Thread-1, started 13448)>, <_MainThread(MainThread, started 9788)>]
0 Waiting 1
1 Waiting 3
2 Waiting 4
3 Waiting 0
3 Done after 0s
0 Done after 1s
1 Done after 3s
2 Done after 4s
End thread [<_MainThread(MainThread, started 9788)>]

 在子线程内开异步：

先上代码：

 1 async def do_some_work(i,x):
 2     print(‘{} Waiting {}‘.format(i,x))
 3     await asyncio.sleep(x)
 4     print(‘{} Done after {}s‘.format(i,x))
 5     return "haha"
 6 
 7 def start_sub_loop():
 8     sub_sub_loop = asyncio.new_event_loop()
 9     asyncio.set_event_loop(sub_sub_loop)
10     print("create start_sub_loop threads ", threading.enumerate())
11     start = now()
12     task = [asyncio.ensure_future(do_some_work(i, randint(0,4))) for i in range(4)]
13     d,p= sub_sub_loop.run_until_complete(asyncio.wait(task))
14     for i in d:
15         print(i.result())
16     for i in p:
17         print(i)
18 19     print("cost time:",time.time()-start)
20     print("over threads",threading.enumerate())
21 def other_thread():
22     print("create other_thread thread:", threading.enumerate())
23     t = threading.Thread(target=start_sub_loop)
24     t.start()
25     t.join()
26     print("after t.join() threads ",threading.enumerate())
27 28 29 th = threading.Thread(target=other_thread)
30 th.start()
31 th.join()
32 print("End thread",threading.enumerate())

输出：

 1 create other_thread thread: [<_MainThread(MainThread, started 15288)>, <Thread(Thread-1, started 14816)>]
 2 create start_sub_loop threads  [<_MainThread(MainThread, started 15288)>, <Thread(Thread-1, started 14816)>, <Thread(Thread-2, started 15252)>]
 3 0 Waiting 3
 4 1 Waiting 1
 5 2 Waiting 0
 6 3 Waiting 0
 7 2 Done after 0s
 8 3 Done after 0s
 9 1 Done after 1s
10 0 Done after 3s
11 haha
12 haha
13 haha
14 haha
15 cost time: 3.000324249267578
16 over threads [<_MainThread(MainThread, started 15288)>, <Thread(Thread-1, started 14816)>, <Thread(Thread-2, started 15252)>]
17 after t.join() threads  [<_MainThread(MainThread, started 15288)>, <Thread(Thread-1, started 14816)>]
18 End thread [<_MainThread(MainThread, started 15288)>]

上面代码共开了三个线程，只在最后一个线程开协程。其实和在主线程开协程一样，不同的是在主线程可以用get_event_loop，在子线程只能用new_event_loop,和set_event_loo。