Python Paho-Mqtt：订阅更多主题需要注意什么？答案

【问题标题】：Python Paho-Mqtt: What should I pay attention to when subscribing to more topics?Python Paho-Mqtt：订阅更多主题需要注意什么？
【发布时间】：2021-06-09 06:47:41
【问题描述】：

使用 mqtt 订阅客户端，我订阅了很多线程（超过 6000 个），但没有得到即时更改的结果。我落后了。 mqtt 是否可以在后台并行订阅太多线程？ loop_start 就够了吗？

订阅更多主题需要注意什么？

import logging
import paho.mqtt.client as mqtt
import requests
import zmq
import pandas as pd

PORT=1351

def set_publisher():
    context = zmq.Context()
    socket_server = context.socket(zmq.PUB)
    socket_server.bind(f"tcp://*:{PORT}")
    return socket_server 

# The callback for when the client receives a CONNACK response from the server.
def on_connect(client, userdata, flags, rc):
    #logging.warning(f"Connected with result code :: code : {rc}")
    print(f"Connected with result code :: code : {rc}")

    # Subscribing in on_connect() means that if we lose the connection and
    # reconnect then subscriptions will be renewed.
    client.subscribe(topics)

# The callback for when a PUBLISH message is received from the server.
def on_message(client, userdata, msg):
    msg = msg.payload
    # logging.info(f"message:: {msg}\n")
    print(f"message:: {msg}\n")
    if msg:
        publisher.send(f"{msg}")

def on_disconnect(client, userdata, rc):
    if rc != 0:
        # logging.warning(f"Unexpected disconnection :: code: {rc}")
        print(f"Unexpected disconnection :: code: {rc}")
    #todo: if rc is change hostname raise except


client = mqtt.Client(protocol=mqtt.MQTTv31, transport="tcp")
client.username_pw_set(******, password=******)

topics = [(f"topic{i}", 0) for i in 6000]

client.on_connect = on_connect
client.on_message = on_message
client.on_disconnect = on_disconnect

if client.connect(hostname= *****, port= **** , keepalive=300) != 0:
    # logging.info("Could not connect to MQTT Broker !")
    print("Could not connect to MQTT Broker !")
    
client.loop_forever(timeout=3000)

【问题讨论】：

标签： python-3.x multithreading mqtt publish-subscribe paho

【解决方案1】：

您正在描述一种计算能力不足以处理您的场景的情况（在客户端或代理或两者之间）。这是一种常见的情况，这就是性能测试的目的：您的设置是否根据您的要求处理您的场景？然后，容量规划将这个问题扩展到：......在未来。

【讨论】：