Python Paho MQTT 订阅者卡住 - influxdb 写答案

【问题标题】：Python Paho MQTT subscriber getting stuck - influxdb writePython Paho MQTT 订阅者卡住 - influxdb 写
【发布时间】：2021-03-11 05:28:31
【问题描述】：

我使用 paho mqtt 模块发布到一个主题，并从另一个程序订阅了它。我正在发布 10000 条消息，发布者能够在大约 2 秒内发送这些消息。在订阅者中，我收到消息并将值写入 influxdb。在大约 2000 条记录之后，MQTT 订阅者正在暂停并等待 time.sleep() 完成。

import paho.mqtt.client as mqtt #import the client1
import time
from datetime import datetime
from influxdb_client import InfluxDBClient, Point, Dialect, WriteOptions
from influxdb_client.client.write_api import SYNCHRONOUS

org = "my-ord"
bucket = "Bucket1"
token = "my-token"
client = InfluxDBClient(url="http://localhost:8086", token=token, org=org)

write_api = client.write_api(write_options=SYNCHRONOUS)
query_api = client.query_api()
#Function to write the record to influx
def update_db(point):
     
    write_api.write(bucket=bucket, record=point)
    print("Point written")

msg_count = 0   
#On message callback function
def on_message(client, userdata, message):
    global msg_count
    msg_count+=1
    print("message received " ,str(message.payload))

    _point1 = Point("mqtt2").tag("message","message").field("datapt",str(message.payload))

    update_db(_point1)

    print(msg_count)
#This is the Subscriber
ip = "localhost" 
client = mqtt.Client("P2")
client.on_message=on_message
client.connect(ip) 
client.loop_start()
client.subscribe("influx")

time.sleep(180)

client.loop_stop()
print(msg_count)

发布者在一秒钟内发布 10000 条消息。如果没有 influx write 命令，代码会一直运行到最后。当我包含写入内容时，订阅者会在大约 2000 条消息后停止。我应该改变什么才能让它工作？

【问题讨论】：

看起来您使用的是 QOS=0（默认），这意味着无法保证交付；一些代理对飞行消息的数量施加了限制（例如 mosquitto 默认为 20），并且可能会丢弃其他消息（即使在 mosquitto 的 QOS 级别较高的情况下，max_queued_messages 默认为 1000，您可能会遇到）。检查您的代理设置和代理日志；我猜它正在丢弃消息。
你的意思是代码在on_message()回调中的第一个和第二个print()之间停止了吗？

标签： python mqtt paho influxdb-python

【解决方案1】：

流入写入将是一个阻塞调用，您不应该在 MQTT 客户端回调中进行阻塞调用，因为它们会阻止客户端处理新的传入消息。

如果由于传入消息而需要阻塞 IO，则应将该工作交给单独的线程。

【讨论】：