您能解释一下，为什么RabbitMQ丢失通过paho / python使用MQTT发送的消息吗？ MQTT生产者 AMQP生产者 MQTT消费者 AMQP消费者

问题描述

我在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条消息。

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

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

99979：dev / testtopic b'99979：2020-10-05T12：00：00.682216'

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

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

我做错什么了吗？您能告诉我消息丢失的地方吗？或给我提示如何调试此问题？结果与使用TLS或关闭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,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()

解决方法

一些建议：

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

_{注意： RabbitMQ团队监视rabbitmq-users mailing list，并且有时仅在StackOverflow上回答问题。}

amqp mqtt mqtt pika python