8

在我的项目中，我尝试启动一个 REST API（使用 FastAPI 构建并使用 Hypercorn 运行），另外我还希望在启动时启动一个 RabbitMQ 使用者（使用 aio_pika）：

Aio Pika 提供了强大的连接，可在失败时自动重新连接。如果我使用hypercorn app:app消费者运行下面的代码并且其余接口正确启动，但是从 aio_pika 重新连接不再起作用。如何在两个不同的进程（或线程？）中归档生产稳定的 RabbitMQ Consumer 和 RestAPI。我的 python 版本是 3.7，请注意我实际上是 Java 和 Go 开发人员，以防我的方法不是 Python 方式:-)

@app.on_event("startup")
def startup():
   loop = asyncio.new_event_loop()
    asyncio.ensure_future(main(loop))


@app.get("/")
def read_root():
   return {"Hello": "World"}


async def main(loop):
connection = await aio_pika.connect_robust(
    "amqp://guest:guest@127.0.0.1/", loop=loop
)

async with connection:
    queue_name = "test_queue"

    # Creating channel
    channel = await connection.channel()  # type: aio_pika.Channel

    # Declaring queue
    queue = await channel.declare_queue(
        queue_name,
        auto_delete=True
    )  # type: aio_pika.Queue

    async with queue.iterator() as queue_iter:
        # Cancel consuming after __aexit__
        async for message in queue_iter:
            async with message.process():
                print(message.body)

                if queue.name in message.body.decode():
                    break
4

1 回答 1

3

在@pgjones 的帮助下，我设法将消费开始更改为：

@app.on_event("startup")
def startup():
    loop = asyncio.get_event_loop()
    asyncio.ensure_future(main(loop))

并启动jobwithasyncio.ensure_future并将当前事件循环作为参数传递，从而解决了问题。

如果有人有不同/更好的方法会很有趣谢谢！

于 2019-12-09T10:33:19.893 回答