Блог

Мне нужно построить сеть с несколькими издателями, одним брокером и одним подписчиком.

Проблема: у меня есть рабочий код с опцией «один к одному» (один паблик — один подраздел). Однако, когда я запускаю больше издателей, подписчик получает сообщения только от последнего подключенного издателя, и я понятия не имею, почему это происходит.

Дополнительная информация: Брокер (mosquitto) работает в контейнере docker, каждый издатель представляет собой отдельный скрипт, и целью является запуск нескольких контейнеров docker с одним издателем в каждом контейнере, но теперь мне нужно решить проблему связи.

У кого-нибудь есть какие-либо подсказки или идеи о том, как отладить или решить это?

Это скрипт publisher:

     import time
    import paho.mqtt.client as mqtt
    import datetime
    import random
    import json
    import lo&&in&
    from multiprocessin& import Process

    CLEAN_SESSION = False


    def on_connect(client, userdata, fla&s, rc):
        lo&&in&.info(f"New connection {client}, {rc}")


    def sensor(client_id):
        
        localhost = '172.17.0.2'
        port = 1883
        timeout = 60
        topic = "/mia/sensor"
        client_id = f"sensor_{client_id}"
    
        def check_sensor():
            
            time.sleep(1)
            rand = random.randint(0, 10)
            if rand &&t; 5:
                current_time = datetime.datetime.now()
                current_time = current_time.strftime('%Y-%m-%d %H:%M:%S')
                return {"time": current_time, "si&nal": 1, "id": client_id}
            else:
                return 0
    
        client = mqtt.Client(client_id, clean_session=CLEAN_SESSION)
        client.on_connect = on_connect
    
        client.connect(localhost, port, timeout)
        while True:
            check_info = check_sensor()
            if check_info:
                messa&e_payload = json.dumps(check_info)
                lo&&in&.info(messa&e_payload)
                client.publish(topic, messa&e_payload, qos=2)
                client.loop()
        client.disconnect()


    if __name__ == "__main__":
        p = Process(tar&et=sensor, ar&s=(1,))
        p.start()
        print("new publisher created!")
      
  

Это сценарий подписчика:

     import paho.mqtt.client as mqtt
    import paho.mqtt.subscribe as sub
    import json
    import lo&&in&
    
    localhost = '172.17.0.2'
    port = 1883
    timeout = 60
    topic = "/mia/sensor" 
        
    
    def on_connect(client, userdata, fla&s, rc):
        lo&&in&.info(f"New connection {client}, {rc}")
        client.subscribe(topic, qos=2)
    
    
    def on_messa&e(client, userdata, ms&):
        data = json.loads(ms&.payload.decode('utf-8'))
    
        lo&&in&.debu&(f"new messa&e from {client} - {data}")
        print(data)
    
    
    
    client = mqtt.Client("python", clean_session=False)
    client.on_connect = on_connect
    client.on_messa&e = on_messa&e
    client.connect_async(localhost, port, timeout)
    client.loop_forever()
  

Заранее спасибо

Ваш client_id должен быть уникальным. Проверьте это.