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

白two

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

async def run():
    async def call_async_sleep():
            await asyncio.sleep(19)
            try:
                await driver.close()
            except:
                pass

    call_async_sleep()
    ......

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

但它报错：

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

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

"""数据库连接测试"""
from threading import Thread
import asyncio
from sqlalchemy import func, select
from db.models import Dictdata
from db import async_session_local


async def test_select():
    async with async_session_local() as dbs:
        total_orm = select(func.count(Dictdata.id)).where(Dictdata.is_active == 1)
        total: int = (await dbs.execute(total_orm)).scalar()
        print(total)


def start_loop(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()


async def end_loop(timeout=15):
    await asyncio.sleep(timeout)
    loop = asyncio.get_event_loop()
    loop.stop()

new_loop = asyncio.new_event_loop()
t = Thread(target=start_loop, args=(new_loop,))
t.start()
asyncio.run_coroutine_threadsafe(test_select(), new_loop)
asyncio.run_coroutine_threadsafe(end_loop(3), new_loop)
t.join()

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

跳转到最佳答案楼层

kogawananari

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

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

"""数据库连接测试"""
from threading import Thread
import asyncio
from sqlalchemy import func, select
from db.models import Dictdata
from db import async_session_local


async def test_select():
    async with async_session_local() as dbs:
        total_orm = select(func.count(Dictdata.id)).where(Dictdata.is_active == 1)
        total: int = (await dbs.execute(total_orm)).scalar()
        print(total)


def start_loop(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()


async def end_loop(timeout=15):
    await asyncio.sleep(timeout)
    loop = asyncio.get_event_loop()
    loop.stop()

new_loop = asyncio.new_event_loop()
t = Thread(target=start_loop, args=(new_loop,))
t.start()
asyncio.run_coroutine_threadsafe(test_select(), new_loop)
asyncio.run_coroutine_threadsafe(end_loop(3), new_loop)
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 ...

拜谢大佬，我再研究研究