使用烧瓶服务器时用于rabbitMQ的pika崩溃答案

【问题标题】：pika for rabbitMQ crashing while using flask server使用烧瓶服务器时用于rabbitMQ的pika崩溃
【发布时间】：2019-05-27 10:34:54
【问题描述】：

所以我们有一个单线程烧瓶服务器运行，我们在其中接收来自 python 应用程序客户端的请求。在这个烧瓶服务器中，我们使用 rabbitMQ 和 pika 库来将消息分发给其他客户端。发生的事情是在 get 函数中程序因错误而崩溃：

pika.exceptions.ConnectionClosed: (505, 'UNEXPECTED_FRAME - 预期 60 类的内容标题，而不是非内容标题框架'）

我在堆栈溢出和其他方面搜索了很多关于此的主题，但它们都解决了多线程的问题，但事实并非如此。除非在 app.run(threaded=yes) 中调用，否则 Flask 只能使用一个线程。

当在短时间内（例如每秒 5 条）发送多条消息时，程序通常会崩溃，同样重要的是要注意每秒都会收到带有对该函数的请求的消息：

@app.route('/api/users/getMessages', methods=['POST'])  
def get_Messages():  
    data = json.loads(request.data)
    token = data['token']

    payload = jwt.decode(token, 'SECRET', algorithms=['HS256'])
    istid = payload['istid']
    print('istid: '+istid)

    messages = []

    queue = channel.queue_declare(queue=istid)
    for i in range(queue.method.message_count):
        method_frame, header_frame, body = channel.basic_get(queue=istid, no_ack=True)
        if method_frame:
            #print(method_frame, header_frame, body)
            messages.append(body)
        else:
            print('No message returned')

    res = {'messages':messages, 'error':0}
    return jsonify(res)

在此代码中，它通常会在该行中崩溃：

queue = channel.queue_declare(queue=istid)

但我们也尝试将代码更改为使用 while 而不是 for 当正文为 None 并且它在行中崩溃时结束的地方：
method_frame, header_frame, body = channel.basic_get(queue=istid, no_ack=True) 在这种情况下。同样重要的是，崩溃是随机的，它可以工作几次，然后在发送消息时在获取请求后随机崩溃。如果有人知道与此相关的任何事情，我们将不胜感激。

另一个注意事项，我们曾考虑使用带有回调的 basic_consume 而不是 basic_get，但我们没有找到可行的方法，因为我们必须将消息发回并让多个用户向同一个函数发出请求。

编辑#1： 在 rabbitMQ 文档rabbitmq 中，如果您搜索函数“def basic_get”，您会注意到有一些 TODO cmets 以及对此的引用

由于实现细节，不能再次调用直到回调被执行。

所以我怀疑这可能是正在发生的事情，但即使是这样我也不知道如何解决。

【问题讨论】：

您在哪里创建连接和通道对象？
您当前的代码不是线程安全的。您需要检查 pika 以获取线程安全示例，或在 get_Messages 中创建连接/通道。
@LukeBakken 我在一开始就在函数和通道之外创建连接（换句话说，只有在我们运行应用程序的时候）。我们尝试在需要获取/发送消息的任何时候打开和关闭频道，它仍然崩溃。然后，我们每次必须获取/发送消息时也尝试打开/关闭连接。它奏效了，但我们认为它太耗费资源并且是一种糟糕的解决方法。
@eandersson 你是什么意思？我们没有使用线程。据我所知，代码只在一个线程中运行，因为我们没有使用线程，也没有告诉烧瓶在多个线程上运行。我是否做出了错误的假设？
“然后，每次我们必须获取/发送消息时，我们都尝试打开/关闭连接。它起作用了......” - 这证明多个线程正在访问您的连接和通道当前有问题的代码。如果您打开 Pika 的调试输出，我想您可能会在日志消息中看到线程 ID，或者您可以使用自定义日志格式化程序添加该信息。

标签： python python-3.x flask rabbitmq pika

【解决方案1】：

对于任何对该解决方案感兴趣的人，就像在其他 cmets 中一样，该程序不是线程安全的，因为从 1.0 版开始，flask 使用 threaded = True 作为默认值。
解决方案是：
1) 使用 app.run(threaded = False)
运行烧瓶 2) 每当使用 pika 访问通道/连接时，通过实施锁来使程序线程安全。

【讨论】：