Несмотря на то, что вы добавили дополнительный код, он не показывает много. Тем не менее, у меня есть опыт, когда подобное происходит со мной. Я делал тяжелое многопоточное приложение с MQTT, и его вполне удалось сохранить. Не совсем, но это так.
Причиной возникновения ошибки при уменьшении задержки является то, что у вас ОДИН клиент. Публикуя сообщение (я не уверен, потому что я не вижу ваш код), вы подключаетесь, отправляете сообщение и отключаете !. Поскольку вы используете этот процесс, вы, скорее всего, отправляете одно сообщение (все еще в процессе) и собираетесь опубликовать новое в новой теме. Однако первый поток завершается и отключает клиента. Новый поток пытается опубликовать, но вы не можете, потому что предыдущий поток отключил вас.
Решение:
1) Не отключать клиента при публикации
2) Рискованно, и вам нужно больше кода: Для каждой публикации создайте нового клиента, но убедитесь, что обрабатываете это правильно. Это означает: создавать клиента, публиковать и отключать снова и снова, но убедитесь, что вы правильно закрыли соединения и удалили клиентов, чтобы не хранить мертвые клиенты
3) решение 2) - попробуйте сделать функцию, которая будет делать все - создать клиента, подключиться и опубликовать и умирает после его окончания. Если вы используете такую функцию, я думаю, вам не придется заботиться о проблемах, возникающих в решении 2
Обновление:
Если ваша проблема в чем-то другом, я все же думаю, что это не из-за самих потоков, а из-за того, что несколько потоков пытаются управлять чем-то, что должно управляться только одним потоком - например, клиентским объектом
Обновление: код шаблона
Имейте в виду, что это мой старый код, и я больше его не использую, потому что моим приложениям нужно определенное отношение к потоку и т. Д., Поэтому я переписываю его для каждого приложения отдельно. Но этот работает как прелесть для непоточных приложений и, возможно, для многопоточных. Он может публиковать только с qos = 0
import paho.mqtt.client as mqtt
import json
# Define Variables
MQTT_BROKER = ""
MQTT_PORT = 1883
MQTT_KEEPALIVE_INTERVAL = 5
MQTT_TOPIC = ""
class pub:
def __init__(self,MQTT_BROKER,MQTT_PORT,MQTT_KEEPALIVE_INTERVAL,MQTT_TOPIC,transport = ''):
self.MQTT_TOPIC = MQTT_TOPIC
self.MQTT_BROKER =MQTT_BROKER
self.MQTT_PORT = MQTT_PORT
self.MQTT_KEEPALIVE_INTERVAL = MQTT_KEEPALIVE_INTERVAL
# Initiate MQTT Client
if transport == 'websockets':
self.mqttc = mqtt.Client(transport='websockets')
else:
self.mqttc = mqtt.Client()
# Register Event Handlers
self.mqttc.on_publish = self.on_publish
self.mqttc.on_connect = self.on_connect
self.connect()
# Define on_connect event Handler
def on_connect(self,mosq, obj, rc):
print("mqtt.thingstud.io")
# Define on_publish event Handler
def on_publish(self,client, userdata, mid):
print("Message Published...")
def publish(self,MQTT_MSG):
MQTT_MSG = json.dumps(MQTT_MSG)
# Publish message to MQTT Topic
self.mqttc.publish(self.MQTT_TOPIC,MQTT_MSG)
# Disconnect from MQTT_Broker
def connect(self):
self.mqttc.connect(self.MQTT_BROKER, self.MQTT_PORT, self.MQTT_KEEPALIVE_INTERVAL)
def disconnect(self):
self.mqttc.disconnect()
p = pub(MQTT_BROKER,MQTT_PORT,MQTT_KEEPALIVE_INTERVAL,MQTT_TOPIC)
p.publish('some messages')
p.publish('more messages')
Обратите внимание, что при создании объекта я подключаюсь автоматически, но я не отключаюсь. Это то, что вы должны сделать вручную
Я предлагаю вам попытаться создать столько объектов pub , сколько у вас есть датчиков, и публиковать их вместе с ними.