如何使用 pika 和 connexion 初始化 RabbitMQ 消费者

Question

我正在尝试设置一个 Python 微服务，它通过 RabbitMQ 接收消息，同时为 Kubernetes 健康检查提供一个 /health REST 端点。我正在使用 pika for the RabbitMQ consumer, and connexion 作为 REST 端点。

但是，当我在 main() 中启动 RabbitMQ 消费者时，connexion 应用程序不会启动。

python-app.py

#!/usr/bin/env python
import pika, sys, os, connexion
from flask import Flask, request, jsonify

app = connexion.FlaskApp(__name__, specification_dir='./')

def main():
    # Connection
    ...
    # Exchange and queues
    ...

    def callback(ch, method, properties, body):
        ...

    channel.basic_consume(queue='pg-python', on_message_callback=callback, auto_ack=True)

    print(' [*] Waiting for messages.')
    channel.start_consuming()
    app.run(port=8080, use_reloader=False)

@app.route('/api/v1/health', methods=['GET'])
def return_health():
    message = {'status':'Healthy! <3'}
    return jsonify(message)

if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print('Interrupted')
        try:
            sys.exit(0)
        except SystemExit:
            os._exit(0)

输出：

[*] Waiting for messages.

如何正确初始化这两个组件？我需要使用线程吗？

Answer 1

我已经通过在单独的线程中初始化 RabbitMQ 消费者解决了这个问题：

#!/usr/bin/env python
import pika, sys, os, threading
from flask import Flask, request, jsonify

app = Flask(__name__)

def start_rmq_connection():
    # Connection
    ...
    # Exchange and queues
    ...

    def callback(ch, method, properties, body):
        ...

    channel.basic_consume(queue='pg-python', on_message_callback=callback, auto_ack=True)
    print(' [*] Waiting for messages.')
    channel.start_consuming()

@app.route('/api/v1/health', methods=['GET'])
def return_health():
    message = {'status':'Healthy! <3'}
    return jsonify(message)

if __name__ == '__main__':
    try:
        thread_1 = threading.Thread(target=start_rmq_connection)
        thread_1.start()
        thread_1.join(0)
        app.run()
    except KeyboardInterrupt:
        print('Interrupted')
        try:
            sys.exit(0)
        except SystemExit:
            os._exit(0)

如何使用 pika 和 connexion 初始化 RabbitMQ 消费者

How to initialize RabbitMQ consumer using pika and connexion

python

pika

connexion