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

szqfcxe2  于 2023-10-20  发布在  RabbitMQ
关注(0)|答案(1)|浏览(160)

我在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。两位消费者都收到了100万条消息。
然后我尝试使用paho-mqtt发送消息,在此脚本完成后,两个客户端都收到了999,983条消息。重复的测试表明,不同数量的消息被丢弃,但总是在十.
为了弄清楚发生了什么,我在消息中添加了消息计数。结果显示,只有最后一条消息丢失。消费者在最后一条记录中显示了此消息:
99979:开发/测试主题b '99979:2020-10-05T12:00:00.682216'
(the第一个99979是消费者的计数器,第二个是生产者的计数器)
为了让事情变得更好,我设置qos=1。现在,在20条消息之后,消费者就停止接收消息了。生产者在我打算发送的消息数之后存在,没有错误。
我做错什么了吗?你能告诉我,这些信息是在哪里丢失的吗?或者给予提示如何调试此问题?结果与使用TLS或关闭TLS无关。
如果你有问题,请问他们!
谢谢
供参考:以下是我使用的(大部分)代码:

MQTT Producer

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 Consumer

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()
snvhrwxg

snvhrwxg1#

一些建议:

  • 您的AMQP发布者应该使用发布者确认。没有它们,您可能会丢失消息-https://www.rabbitmq.com/confirms.html#publisher-confirms
  • 您的MQTT客户端在所有消息发布之前退出。这不是RabbitMQ的bug。您需要注册on_publish回调,并确保在程序退出之前发布已完成。一个黑客的方法来做到这一点将等待一段时间后,您的最后一次发布(30秒?),然后退出。

相关问题