在工作线程/循环中运行异步函数

如何解决在工作线程/循环中运行异步函数

我正在尝试编写代码,以便能够从大部分同步代码中使用 asyncpg(以避免重复)。 由于一些非常奇怪的原因,协程 Database.test() 将在我的工作事件循环/线程中执行并返回。未来似乎工作正常。但是使用 asyncpg 连接到数据库只会挂起。知道原因吗?

另外,也许我应该改用 asyncio.run()。

from threading import Thread
import asyncio
import asyncpg

class AsyncioWorkerThread(Thread):

    def __init__(self,*args,daemon=True,loop=None,**kwargs):
        super().__init__(*args,daemon=daemon,**kwargs)
        self.loop = loop or asyncio.new_event_loop()
        self.running = False

    def run(self):
        self.running = True
        self.loop.run_forever()

    def submit(self,coro):
        fut = asyncio.run_coroutine_threadsafe(coro,loop=self.loop)
        return fut.result()

    def stop(self):
        self.loop.call_soon_threadsafe(self.loop.stop)
        self.join()
        self.running = False

class Database:
    async def test(self):
        print('In test')
        await asyncio.sleep(5)

    async def connect(self):
        # Put in your db credentials here
        # pg_user = ''
        # pg _password = ''
        # pg_host = ""
        # pg_port = 20
        # pg_db
        connection_uri = f'postgres://{pg_user}:{pg_password}@{pg_host}:{pg_port}/{pg_db}'
        self.connection_pool = await asyncpg.create_pool(
            connection_uri,min_size=5,max_size=10)

if __name__ == "__main__":
    db = Database()
    worker = AsyncioWorkerThread()
    worker.start()
    worker.submit(db.test())  # Works future returns correctly
    worker.submit(db.connect())  # Hangs,thread never manages to acquire

版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。

相关推荐


Selenium Web驱动程序和Java。元素在(x,y)点处不可单击。其他元素将获得点击?
Python-如何使用点“。” 访问字典成员?
Java 字符串是不可变的。到底是什么意思?
Java中的“ final”关键字如何工作?(我仍然可以修改对象。)
“loop:”在Java代码中。这是什么,为什么要编译?
java.lang.ClassNotFoundException:sun.jdbc.odbc.JdbcOdbcDriver发生异常。为什么?
这是用Java进行XML解析的最佳库。
Java的PriorityQueue的内置迭代器不会以任何特定顺序遍历数据结构。为什么?
如何在Java中聆听按键时移动图像。
Java“Program to an interface”。这是什么意思?
Java在半透明框架/面板/组件上重新绘画。
Java“ Class.forName()”和“ Class.forName()。newInstance()”之间有什么区别?
在此环境中不提供编译器。也许是在JRE而不是JDK上运行?
Java用相同的方法在一个类中实现两个接口。哪种接口方法被覆盖?
Java 什么是Runtime.getRuntime()。totalMemory()和freeMemory()?
java.library.path中的java.lang.UnsatisfiedLinkError否*****。dll
JavaFX“位置是必需的。” 即使在同一包装中
Java 导入两个具有相同名称的类。怎么处理?
Java 是否应该在HttpServletResponse.getOutputStream()/。getWriter()上调用.close()?
Java RegEx元字符(。)和普通点?