python-asyncio - FastAPI如何将ZMQ添加到事件循环
问题描述
我很惊讶这并没有真正被详细询问,但由于某种原因,我无法在任何地方找到这个问题或解决方案。似乎很多人都遇到了一个问题,即您有一个 fastAPI 应用程序,该应用程序还需要与其他一些微服务进行通信(以比 http 消息更有效的方式)。我已经阅读了所有关于集成到 asyncio 中的 zmq 文档,但是到目前为止,我还没有找到任何关于如何使用 fastapi(甚至是 starlette)将 zmq 添加到事件循环中的信息。这是 zmq 网站上的代码示例:
import asyncio
import zmq
from zmq.asyncio import Context
ctx = Context.instance()
async def recv():
s = ctx.socket(zmq.SUB)
s.connect('tcp://127.0.0.1:5555')
s.subscribe(b'')
while True:
msg = await s.recv_multipart()
print('received', msg)
s.close()
这向我们展示了一个异步函数,这很棒,但是这又一次需要在事件循环中与 fastAPI 协程一起运行。这应该怎么做?除了后台任务之外,fastAPI 文档并没有真正为我们提供任何接口来运行单独的协程。我不确定后台任务中是否发生了任何其他魔法,但对于需要与另一个微服务通信的东西,我希望它具有类似于 fastAPI 协程的调度。此外,您无法在启动时启动后台任务,因此您必须进行一些欺骗性的调用才能使其运行(这很hacky ..但从技术上讲是可行的)。此外,如果我们可以只注册一个处理程序,那就更好了
@app.set("zmq_recv)
async def recv():
s = ctx.socket(zmq.SUB)
s.connect('tcp://127.0.0.1:5555')
s.subscribe(b'')....
这将基于某处的配置,允许 zmq 上下文中的所有消息自动转到此函数。这可能允许我们在 fastAPI 协程中运行 zmq,只需绑定另一个端口,并确保来自该端口的所有流量都流向这个特殊的 app.set 方法。像这样的东西我会很好......
ctx = Context.instance()
@app.on_event("startup")
async def startup_event():
s = ctx.socket(zmq.PULL)
s.bind('tcp://127.0.0.1:5555').setHandler("zmq_recv") # this setHandler is something magic that tells fastAPI to have all traffic on port 5555 to go to this handler function
@app.on_special_handler("zmq_recv")
async def zmq_recv(socket):
msg = await socket.recv_multipart()
print('received', msg)
理想情况下,这是我想要的,使用 fastAPI 协程......但是让 zmq 套接字调用对特定函数的响应。是否可以创建协程示例或这种模板化示例?如果不是,人们如何声称您可以有效地将 zmq 与 fastapi 一起使用?(我想后台任务可以解决问题..但它似乎真的很虚伪)
解决方案
我实际上正在寻找类似的东西,专门用于通过非 http 处理程序侦听来自消息队列的事件。我当前的解决方案是连接到事件循环并在那里添加监听器。它可能看起来像这样:(取自 fastapi -> issue中的这个问题)
loop = asyncio.get_event_loop()
loop.create_task(serve(app, config)) # run fastapi
loop.create_task(your_tcp_app()) # run your app
loop.run_forever() # start event loop
您也可以在初始化后调用应用程序正在使用的事件循环,这与您最初所做的更相似:
@app.on_event("startup")
async def startup_event():
loop = asyncio.get_event_loop() # should return the loop fastapi is already using
loop.create_task(your_tcp_app()) # run your app
我正在考虑这个解决方案,因为我害怕 python 中的线程管理,因为这将存在于生产服务中,我可能最终将两个进程完全分开以使事情变得更简单......我想要知道这是否对您有用,或者其他人是否对这种实现有任何想法:)
推荐阅读
- wpf - 使用 WPF 中的按钮调用 F# 函数(FsXaml 方法)
- node.js - 使用 Node.js 在 Hyperledger Fabric 中创建通道时出错
- c# - Microsoft.CrmSdk.XrmTooling.CoreAssembly 与 Targetframework .NET 4.5.2 在 CrmServiceClient.Execute 上导致 NullReference 异常
- python - 如何在 keras 模型中设置一个常数作为权重?
- .net - 没有找到要发布的结果。Azure DevOps 构建中的单元测试
- python - 在无限期间写入文件
- oracle - 使用阿拉伯语消息调用 web 服务
- delphi - 我如何知道字体是否具有给定 Unicode 字符的字形?
- c# - ASP.NET Core - 尝试使用 HealthChecks 时出错
- node.js - Docker Postgres 服务不允许我连接到它