[已解决]异步函数调用如何不阻塞后续进程？

白two

我有一个异步函数，在里面再定义了一个异步函数

1. async def run():
   
2.     async def call_async_sleep():
   
3.             await asyncio.sleep(19)
   
4.             try:
   
5.                 await driver.close()
   
6.             except:
   
7.                 pass
   
8. 
   
9.     call_async_sleep()
   
10.     ......

复制代码

我需要只要一开始，里面的那个函数就会执行，但是不会阻碍后面函数的执行，然后不管最后结果如何(因为后续操作容易出bug，但是想做一个统一的处理方式)，都会执行这个函数内的操作

但它报错：

1. RuntimeWarning: coroutine 'run.<locals>.call_async_sleep' was never awaited

复制代码

加上 await 的话就会先等这个函数执行完，再执行后面的操作，但我需要的是它不会阻碍后面函数的运行
才接触异步，有哪个大佬能解决一下吗？

最佳答案

月排行榜 / 总排行榜

kogawananari

2021-11-14 23:46:03

coroutine 同一时间只能执行一个   future是注册了就可以托管运行的
无论是coroutine还是future都得await才行 不能直接执行  
除此之外还有一些API可以直接并发运行coroutine不需要自己去注册为future
其中最好用的应该是run_coroutine_threadsafe  他可以往你的进程里发送coroutine

我给出一个数据库查询的例子

1. """数据库连接测试"""
   
2. from threading import Thread
   
3. import asyncio
   
4. from sqlalchemy import func, select
   
5. from db.models import Dictdata
   
6. from db import async_session_local
   
7. 
   
8. 
   
9. async def test_select():
   
10.     async with async_session_local() as dbs:
    
11.         total_orm = select(func.count(Dictdata.id)).where(Dictdata.is_active == 1)
    
12.         total: int = (await dbs.execute(total_orm)).scalar()
    
13.         print(total)
    
14. 
    
15. 
    
16. def start_loop(loop):
    
17.     asyncio.set_event_loop(loop)
    
18.     loop.run_forever()
    
19. 
    
20. 
    
21. async def end_loop(timeout=15):
    
22.     await asyncio.sleep(timeout)
    
23.     loop = asyncio.get_event_loop()
    
24.     loop.stop()
    
25. 
    
26. new_loop = asyncio.new_event_loop()
    
27. t = Thread(target=start_loop, args=(new_loop,))
    
28. t.start()
    
29. asyncio.run_coroutine_threadsafe(test_select(), new_loop)
    
30. asyncio.run_coroutine_threadsafe(end_loop(3), new_loop)
    
31. t.join()

复制代码

test_select 就是主函数  其他的你都不需要改 在里面写就行

跳转到最佳答案楼层

kogawananari

coroutine 同一时间只能执行一个   future是注册了就可以托管运行的
无论是coroutine还是future都得await才行 不能直接执行  
除此之外还有一些API可以直接并发运行coroutine不需要自己去注册为future
其中最好用的应该是run_coroutine_threadsafe  他可以往你的进程里发送coroutine

我给出一个数据库查询的例子

1. """数据库连接测试"""
   
2. from threading import Thread
   
3. import asyncio
   
4. from sqlalchemy import func, select
   
5. from db.models import Dictdata
   
6. from db import async_session_local
   
7. 
   
8. 
   
9. async def test_select():
   
10.     async with async_session_local() as dbs:
    
11.         total_orm = select(func.count(Dictdata.id)).where(Dictdata.is_active == 1)
    
12.         total: int = (await dbs.execute(total_orm)).scalar()
    
13.         print(total)
    
14. 
    
15. 
    
16. def start_loop(loop):
    
17.     asyncio.set_event_loop(loop)
    
18.     loop.run_forever()
    
19. 
    
20. 
    
21. async def end_loop(timeout=15):
    
22.     await asyncio.sleep(timeout)
    
23.     loop = asyncio.get_event_loop()
    
24.     loop.stop()
    
25. 
    
26. new_loop = asyncio.new_event_loop()
    
27. t = Thread(target=start_loop, args=(new_loop,))
    
28. t.start()
    
29. asyncio.run_coroutine_threadsafe(test_select(), new_loop)
    
30. asyncio.run_coroutine_threadsafe(end_loop(3), new_loop)
    
31. t.join()

复制代码

test_select 就是主函数  其他的你都不需要改 在里面写就行

z5560636

kogawananari 发表于 2021-11-15 00:40
coroutine 同一时间只能执行一个   future是注册了就可以托管运行的
无论是coroutine还是future都得await ...

  我看他们直接是用task 来创建队列的，asyncio包里面有启动异步的，不需要用线程来启动。

tomok

关注感兴趣的帖。

kogawananari

z5560636 发表于 2021-11-15 09:58
我看他们直接是用task 来创建队列的，asyncio包里面有启动异步的，不需要用线程来启动。

task 是 future的子类  有什么问题吗 直接await task也是异步的

白two

kogawananari 发表于 2021-11-14 23:46
coroutine 同一时间只能执行一个   future是注册了就可以托管运行的
无论是coroutine还是future都得await ...

拜谢大佬，我再研究研究