你能解释一下，为什么 RabbitMQ 会丢失通过 paho/python 使用 MQTT 发送的消息吗？答案

【问题标题】：Can you explain, why RabbitMQ is losing messages sent using MQTT via paho/python?你能解释一下，为什么 RabbitMQ 会丢失通过 paho/python 使用 MQTT 发送的消息吗？
【发布时间】：2021-01-20 17:54:39
【问题描述】：

我在一个 linux 节点上运行 RabbitMQ 3.7.28，单一安装，没有更多的集群节点。启用 MQTT 消息插件，TLS 和非 TLS 连接均成功。使用 python 3.8 和 pika 1.1.0，我通过 AMQP 向代理发送了 1,000,000 条消息。

在发送消息期间，我连接了两个消费者：一个使用 pika/AMQP，一个使用 paho-mqtt 1.5.1。两个消费者都收到了 1,000,000 条消息。

然后我尝试使用 paho-mqtt 发送消息，在这个脚本完成后，两个客户端都收到了 999,983 条消息。重复测试表明，丢弃的消息数量不同，但总是几十个。

为了弄清楚发生了什么，我将消息计数添加到消息中。结果消息显示，只有最后一条消息丢失。消费者将此消息显示为最后一个条目：

99979: 开发/测试主题 b'99979: 2020-10-05T12:00:00.682216'

（第一个99979是来自消费者的计数器，第二个是来自生产者的计数器）

为了让事情变得更好，我设置了 qos=1。现在可靠地在 20 条消息后消费者停止接收消息。在我打算发送的消息数量之后，生产者存在且没有错误。

我做错了吗？你能告诉我，消息在哪里丢失了吗？或者给我一个关于如何调试这个问题的提示？结果与使用 TLS 或将其关闭无关。

如果您有任何问题，请提出来！

谢谢。

供参考：这是我使用的（大部分）代码：

MQTT 生产者

import paho.mqtt.client as mqtt
from datetime import datetime

client = mqtt.Client()
client.username_pw_set(user, password)
client.connect(server, port)

print(datetime.utcnow().isoformat())
for i in range(1000000):
    client.publish("dev/testtopic", f'{i + 1}: {datetime.utcnow().isoformat()}', qos=0)
print(datetime.utcnow().isoformat())

client.disconnect()

AMQP 生产者

import pika
from datetime import datetime
from urllib.parse import quote


with pika.BlockingConnection(pika.URLParameters(f'amqp://{user}:{password}@{server}:{port}/{vhost}')) as connection:
    print(datetime.utcnow().isoformat())
    channel = connection.channel()
    routing_key = 'dev.testtopic'
    for i in range(1000000):
        channel.basic_publish(
            exchange='amq.topic', routing_key=routing_key, body=f'{i}: {datetime.utcnow().isoformat()}')
    print(datetime.utcnow().isoformat())

MQTT 消费者

import paho.mqtt.client as mqtt


def on_connect(client, userdata, flags, rc):
    print("Connected with result code " + str(rc))

    # Subscribing in on_connect() means that if we lose the connection and
    # reconnect then subscriptions will be renewed.
    client.subscribe("$SYS/#")
    client.subscribe("dev/testtopic")


def on_message(client, userdata, msg):
    global count
    count += 1
    print(f'{count}: {msg.topic} {str(msg.payload)}')


count = 0

client = mqtt.Client()
client.username_pw_set(user, password)
client.on_connect = on_connect
client.on_message = on_message

client.connect(server, host)
client.loop_forever()

AMQP 消费者

import pika

def callback(ch, method, properties, body):
    global count
    count += 1
    print(f'{count}: {method.routing_key} {body}')


with pika.BlockingConnection(pika.URLParameters(f'amqp://{user}:{password}@{server}:{port}/{vhost}')) as connection:
    channel = connection.channel()

    result = channel.queue_declare(queue='', exclusive=True)
    queue_name = result.method.queue

    channel.queue_bind(exchange='amq.topic', queue=queue_name, routing_key='dev.testtopic')
    print(' [*] Waiting for messages. To exit press CTRL+C')
    count = 0


    channel.basic_consume(
        queue=queue_name, on_message_callback=callback, auto_ack=True)

    channel.start_consuming()

【问题讨论】：

您是否尝试将“on_disconnect()”添加到您的 MQTT 使用者以查看它是否完全断开连接？如果问题仅在运行结束时发生，请尝试在 MQTT Producer 上注释掉“client.disconnect()”调用，看看是否有任何改变......可能是时间问题。
非常感谢。消费者不会断开连接。事实证明，这是一个时间和/或冲洗问题；通过对代码进行一些修改，我可以可靠地快速发送超过 5,000,000 条消息。我会弄清楚这是由服务器上的 paho 还是 MQTT 插件引起的，并相应地创建一个票证。

标签： python rabbitmq mqtt amqp pika

【解决方案1】：

一些建议：

您的 AMQP 发布者应该使用发布者确认。没有它们，您可能会丢失消息 - https://www.rabbitmq.com/confirms.html#publisher-confirms
您的 MQTT 客户端在所有消息发布之前退出。这不是 RabbitMQ 错误。您需要注册on_publish 回调并确保在程序退出之前发布已完成。执行此操作的一个老生常谈的方法是在您上次发布后等待一段时间（30 秒？）然后退出。

_{注意：RabbitMQ 团队会监控 rabbitmq-users mailing list，并且有时只回答 StackOverflow 上的问题。}

【讨论】：