从 Flask 发布到 RabbitMQ

Question

我有一个由 Gunicorn 运行的 Flask Web 应用程序。 Web 应用程序有几个端点，它接收 JSON 数据并对其进行处理，然后需要将处理后的数据发布到 RabbitMQ 交换以供其他微服务获取以进行进一步处理。 HTTP 端不需要响应。

我有一个名为 rabbit.py 的文件，内容如下：

import pika
import json
import os

credentials = pika.PlainCredentials(os.environ['AMQP_USER'], os.environ['AMQP_USER'])
parameters = pika.ConnectionParameters(os.environ['AMQP_HOST'],
                                       os.environ['AMQP_PORT'],
                                       '/',
                                       credentials)
connection = pika.BlockingConnection(parameters)
channel = connection.channel()

channel.exchange_declare(exchange=os.environ['AMQP_EXCHANGE'],
                         exchange_type='fanout',
                         durable=True)

def publish(data):
    channel.basic_publish(exchange=os.environ['AMQP_EXCHANGE'],
                        routing_key='',
                        body=json.dumps(data),
                        properties=pika.BasicProperties(delivery_mode = 2))

然后我将通过 Flask 端点函数执行 publish(data_object) 以获取发布到 RabbitMQ 的数据。

问题是与 RabbitMQ 的连接由于 ping 未处理而超时

rabbitmq_1     | 2020-04-01 08:56:45.911 [info] <0.806.0> accepting AMQP connection <0.806.0> (172.19.0.3:48856 -> 172.19.0.2:5672)
rabbitmq_1     | 2020-04-01 08:56:45.915 [info] <0.806.0> connection <0.806.0> (172.19.0.3:48856 -> 172.19.0.2:5672): user 'guest' authenticated and granted access to vhost '/'
rabbitmq_1     | 2020-04-01 09:02:45.920 [error] <0.806.0> closing AMQP connection <0.806.0> (172.19.0.3:48856 -> 172.19.0.2:5672):
rabbitmq_1     | missed heartbeats from client, timeout: 60s

ingress-api_1  | [2020-04-01 09:06:16,835] ERROR in app: Exception on /ttn/v2 [POST]
ingress-api_1  | Traceback (most recent call last):
ingress-api_1  |   File "/usr/local/lib/python3.8/site-packages/flask/app.py", line 2446, in wsgi_app
ingress-api_1  |     response = self.full_dispatch_request()
ingress-api_1  |   File "/usr/local/lib/python3.8/site-packages/flask/app.py", line 1951, in full_dispatch_request
ingress-api_1  |     rv = self.handle_user_exception(e)
ingress-api_1  |   File "/usr/local/lib/python3.8/site-packages/flask/app.py", line 1820, in handle_user_exception
ingress-api_1  |     reraise(exc_type, exc_value, tb)
ingress-api_1  |   File "/usr/local/lib/python3.8/site-packages/flask/_compat.py", line 39, in reraise
ingress-api_1  |     raise value
ingress-api_1  |   File "/usr/local/lib/python3.8/site-packages/flask/app.py", line 1949, in full_dispatch_request
ingress-api_1  |     rv = self.dispatch_request()
ingress-api_1  |   File "/usr/local/lib/python3.8/site-packages/flask/app.py", line 1935, in dispatch_request
ingress-api_1  |     return self.view_functions[rule.endpoint](**req.view_args)
ingress-api_1  |   File "/app/main.py", line 28, in ttn_v2_endpoint
ingress-api_1  |     ttn.parse_ttn_http_post_v2(request.get_json())
ingress-api_1  |   File "/app/ttn.py", line 72, in parse_ttn_http_post_v2
ingress-api_1  |     rabbit.publish(parsed_object)
ingress-api_1  |   File "/app/rabbit.py", line 18, in publish
ingress-api_1  |     channel.basic_publish(exchange=os.environ['AMQP_EXCHANGE'],
ingress-api_1  |   File "/usr/local/lib/python3.8/site-packages/pika/adapters/blocking_connection.py", line 2248, in basic_publish
ingress-api_1  |     self._flush_output()
ingress-api_1  |   File "/usr/local/lib/python3.8/site-packages/pika/adapters/blocking_connection.py", line 1336, in _flush_output
ingress-api_1  |     self._connection._flush_output(lambda: self.is_closed, *waiters)
ingress-api_1  |   File "/usr/local/lib/python3.8/site-packages/pika/adapters/blocking_connection.py", line 522, in _flush_output
ingress-api_1  |     raise self._closed_result.value.error
ingress-api_1  | pika.exceptions.StreamLostError: Stream connection lost: ConnectionResetError(104, 'Connection reset by peer')

有一些第三方扩展将 rabbitmq 引入烧瓶。然后有一些建议如何实现这一点。有些人使用芹菜。

我需要一种将数据发布到 Rabbit 的简单直接的方法。我不需要订阅回复。我可以只禁用 ping 保持活动，或者最好和最简单的解决方案是什么？

Answer 1

我使用 Flask 扩展使其工作： https://github.com/wdtinc/flask-pika

它没有相同的连接错误，所以它似乎按预期工作。