python - 你能解释一下,为什么 RabbitMQ 会丢失通过 paho/python 使用 MQTT 发送的消息吗?
问题描述
我在 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()
解决方案
一些建议:
- 您的 AMQP 发布者应该使用发布者确认。没有它们,您可能会丢失消息 - https://www.rabbitmq.com/confirms.html#publisher-confirms
- 在发布所有消息之前,您的 MQTT 客户端正在退出。这不是 RabbitMQ 错误。您需要注册
on_publish
回调并确保在程序退出之前发布已完成。一个 hacky 方法是在您上次发布后等待一段时间(30 秒?)然后退出。
注意: RabbitMQ 团队会监控rabbitmq-users
邮件列表,有时只会在 StackOverflow 上回答问题。
推荐阅读
- spring - 如何在 SpringBoot 中拥有同一个类的两个实现(测试和实际)?
- java - Room DB - DISTINCT 用于组合 2 个字段 (GPS)
- python - 模块“cv2.cv2”没有属性“sfm”
- c# - 控制台应用程序中的 C# 多个入口点
- python - 在 Cython 中使用数组时计算速度的巨大损失
- c# - Blazor 项目中的 C++ DLL
- gradle - Gradle:让依赖的版本依赖于传递依赖的版本
- ios - 正则表达式以@开头,以空格结尾
- java - JUnit 5 模拟 YearMonth.now 但为 YearMonth.from 调用真实方法
- envoyproxy - Envoy 如何支持随机采样但稳定的跟踪?