Paho的多线程MQTT客户端无法使用Paho Python库接收消息

如何解决Paho的多线程MQTT客户端无法使用Paho Python库接收消息

我需要构建一个程序，该程序可以创建与MQTT任务和主机一样多的客户端：

• 两个主机：mqtt-1.hostname.com和mqtt-hostname.com
• 任务：my / task / 1，my / task / 2等

我遵循了这个著名的教程：http://www.steves-internet-guide.com/multiple-client-connections-python-mqtt/

有效方法：

• 创建线程
• 为每个线程创建客户端
• 每个客户端都订阅每个线程

什么不起作用：

• 收到消息

我修改了本教程的代码，目的是：我不想永远运行代码，只是在那一刻获取所有数据然后退出。

代码：

def Connect(client,broker,port,keepalive,client_data,run_forever=False):
    """Attempts connection set delay to >1 to keep trying
    but at longer intervals. If runforever flag is true then
    it will keep trying to connect or reconnect indefinetly otherwise
    gives up after 3 failed attempts"""
    connflag=False
    delay=5
    print("connecting ",client)
    badcount=0 # counter for bad connection attempts
    while not connflag:
        print("connecting to broker "+str(broker)+":"+str(port))
        print("Attempts ",str(badcount))
        time.sleep(delay)
        try:
            client.username_pw_set(user_id,password)
            client.connect(broker,keepalive)
            connflag = True
        except:
            client.badconnection_flag=True
            print("connection failed "+str(badcount))
            badcount +=1
            if badcount>=3 and not run_forever: 
                return -1
                raise SystemExit #give up
    print("Connect() done.")
    return 0


def mqtt_client_loop(client,keepalive=60,loop_function=None,\
             loop_delay=1,run_forever=False):
    #runs a loop that will auto reconnect and subscribe to topics
    #pass topics as a list of tuples. You can pass a function to be
    #called at set intervals determined by the loop_delay
    client.run_flag=True
    client.broker=broker
    print("running loop ")
    client.reconnect_delay_set(min_delay=1,max_delay=12)
        
    if not client.connected_flag:
        print("Connecting to ",broker)
        if Connect(client,run_forever) !=-1:
            pass
        else:#connect fails
            client.run_flag=False #break
            print("quitting loop for  broker ",broker)
    if client.connected_flag and loop_function: #function to call
        loop_function(client,loop_delay) #call function

    time.sleep(1)
    print("disconnecting from "+broker+" "+client_data["sub_topic"])
    if client.connected_flag:
        client.disconnect()
        client.connected_flag=False


def on_mqtt_message(client,userdata,message):
    print("on_mqtt_message_callback")
    print("message received ",str(message.payload.decode("utf-8")))
    print("message topic=",message.topic)
    print("message qos=",message.qos)
    print("message retain flag=",message.retain)
    // do stuff
    client.publish(message.topic,payload=None,qos=0,retain=True) # remove message


def on_mqtt_connect(client,flags,rc):
    print("on_mqtt_connect " + str(rc))
    if rc == 0:
        client.connected_flag=True #set flag
        client.loop_start()
        for c in clients:
            if client == c["client"]:
                print("subscribe to "+c["sub_topic"]+" ("+c["broker"]+")")
                client.subscribe(c["sub_topic"],qos=1)
                print("done!")
        print("connected OK")
    else:
        print("Bad connection Returned code=",rc)
        client.loop_stop()


def on_mqtt_disconnect(client,rc):
    client.connected_flag=False #set flagclient_loop
    client.loop_stop()
    print("client disconnected ok")


def on_mqtt_publish(client,mid):
   time.sleep(1)
   print("In on_pub callback mid= ",mid)


def pub(client,loop_delay):
    #print("in publish")
    pass


def on_mqtt_subscribe(client,mid,granted_qos):
    print("Subscribed to topic " + str(mid) + " with QoS: " + str(granted_qos))


def on_mqtt_log(mqttc,obj,level,string):
    print(string)