如何使用 FastAPI 作为 RabbitMQ (RPC) 的使用者

Question

这里的示例展示了如何使用远程过程调用（RPC）在 python 中创建客户端和服务器。

但我无法想象 FastAPI 服务如何成为一个服务器，使用 pika for RabbitMQ 来消费来自 RCP 客户端的请求。

任何 Web 服务都将通过显式调用它们来请求，但是，我无法想象如何将 RabbitMQ 消费者集成到 Web 服务中。

另一方面，对于客户端来说，这很容易做到，通过显式调用 Web 服务，您可以发布对队列的请求，请参阅此示例

有什么帮助吗？或者一个好的开始？

Answer 1

您可以使用aio_pikaRPC模式并执行以下操作：

服务1（消耗）

循环消费：

# app/__init__.py

from fastapi import FastAPI
from app.rpc import consume

app = FastAPI()

...

@app.on_event('startup')
def startup():
    loop = asyncio.get_event_loop()
    # use the same loop to consume
    asyncio.ensure_future(consume(loop))

...

创建连接、通道并注册要从另一个服务调用的远程方法：

# app/rpc.py

from aio_pika import connect_robust
from aio_pika.patterns import RPC

from app.config import config

__all__ = [
    'consume'
]


def remote_method():
    # DO SOMETHING
    # Move this method along with others to another place e.g. app/rpc_methods
    # I put it here for simplicity
    return 'It works!'

async def consume(loop):
    connection = await connect_robust(config.AMQP_URI, loop=loop)
    channel = await connection.channel()
    rpc = await RPC.create(channel)

    # Register your remote method
    await rpc.register('remote_method', remote_method, auto_delete=True)
    return connection

这就是您需要使用和响应的全部内容，现在让我们看看调用此远程方法的第二个服务。

服务2（调用远程方法）

让我们首先创建 RPC 中间件来轻松管理和访问 RPC 对象，以便从 API 函数调用我们的远程方法：

# app/utils/rpc_middleware.py

import asyncio

from fastapi import Request, Response

from aio_pika import connect_robust
from aio_pika.patterns import RPC

from app.config import config

__all__ = [
    'get_rpc',
    'rpc_middleware'
]


async def rpc_middleware(request: Request, call_next):
    response = Response("Internal server error", status_code=500)
    try:
        # You can also pass a loop as an argument. Keep it here now for simplicity
        loop = asyncio.get_event_loop()
        connection = await connect_robust(config.AMQP_URI, loop=loop)
        channel = await connection.channel()
        request.state.rpc = await RPC.create(channel)
        response = await call_next(request)
    finally:

        # UPD: just thought that we probably want to keep queue and don't
        # recreate it for each request so we can remove this line and move
        # connection, channel and rpc initialisation out from middleware 
        # and do it once on app start

        # Also based of this: https://github.com/encode/starlette/issues/1029
        # it's better to create ASGI middleware instead of HTTP
        await request.state.rpc.close()
    return response


# Dependency to use rpc inside routes functions
def get_rpc(request: Request):
    rpc = request.state.rpc
    return rpc

应用RPC中间件：

# app/__init__.py

from app.utils import rpc_middleware

...

app.middleware('http')(rpc_middleware)

...

通过 API 函数中的依赖项使用 RPC 对象：

# app/api/whatever.py

from aio_pika.patterns import RPC

from app.utils import get_rpc

...

@router.get('/rpc')
async def rpc_test(rpc: RPC = Depends(get_rpc)):
    response = await rpc.proxy.remote_method()
    ...

添加一些日志记录来跟踪这两个服务中发生的情况。此外，您还可以将两个服务中的 RPC 逻辑合并为一个，以便能够在同一服务中使用和调用远程方法。

希望它有助于获得基本概念。