EMQX下带唯一标识的动态MQTT主题发布耗时远高于静态主题如何优化

性能根因 #

• 主题动态创建开销：EMQX采用树状结构存储主题元数据，首次发布新主题时需要创建节点、更新路由表，开销远高于复用已存在的静态主题。
• 保留消息重复写入开销：测试代码中所有消息均设置了setRetained(true)，每个新的动态主题都会触发一次保留消息持久化写入操作，静态主题仅需写入一次保留消息，后续发布无额外IO开销，因此速度差异极大。
• 同步发布阻塞叠加：你使用的是Paho同步客户端的publish方法，QoS=1时每次调用都会阻塞等待broker返回PUBACK确认后才会继续执行，动态主题的处理耗时更长，100次调用的阻塞时间叠加后总耗时被大幅放大。

优化措施 #

1. 调整保留消息配置 #

如果业务不需要为每个ID对应的动态主题保存保留消息，直接将mqttMessage.setRetained(true)修改为false，该调整可降低90%以上的动态主题发布耗时。如果确实需要保留消息能力，可开启EMQX的保留消息内存缓存配置，降低磁盘IO开销。

2. 改用异步发布模式 #

将同步IMqttClient替换为异步IMqttAsyncClient，通过回调处理发布结果，避免单线程阻塞等待，大幅提升发布吞吐量，示例修改如下：

public class MqttPublish {
    static IMqttAsyncClient instance = null;
    public static IMqttAsyncClient getInstance() throws MqttException {
        try {
            if (instance == null) {
                instance = new MqttAsyncClient(mqttHostUrl, "SimpleTestMQTT");
            }
            if (!instance.isConnected()) {
                MqttConnectOptions options = new MqttConnectOptions();
                options.setUserName("test");
                options.setPassword("test".toCharArray());
                options.setAutomaticReconnect(true);
                options.setCleanSession(false);
                options.setConnectionTimeout(10);
                instance.connect(options).waitForCompletion();
            }
        } catch (Exception e) {
            System.out.println("Exception in mqtt: {}" + e.getMessage());
        }
        return instance;
    }
    public static void publishMessage() throws MqttException {
        IMqttAsyncClient iMqttClient = getInstance();
        MqttMessage mqttMessage = new MqttMessage("Hello".getBytes());
        mqttMessage.setQos(1);
        mqttMessage.setRetained(false); // 按需调整保留消息配置
        // 动态主题异步发布
        System.out.println("Publish Start for pattern 1");
        int i = 0;
        final long startTime = System.currentTimeMillis();
        do {
            String topic = "/config/" + i + "/outward";
            iMqttClient.publish(topic, mqttMessage, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    // 发布成功逻辑
                }
                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    // 发布失败重试逻辑
                }
            });
            i++;
        } while(i < 100);
        System.out.println("Total time pattern 1: " + (System.currentTimeMillis() - startTime));
    }
}

3. 调整EMQX服务端配置 #

如果动态ID范围可控，可提前在EMQX中预创建所有需要用到的主题，避免发布时的动态创建开销；如果ID范围不可控，可开启EMQX的主题元数据缓存功能，降低频繁创建销毁主题的性能损耗。

4. 业务逻辑优化 #

如果接收端可以适配，可将ID参数放到消息Payload中传输，统一使用同一个静态主题发布所有消息，接收端从Payload中解析ID处理业务，完全消除动态主题的额外开销。

内容的提问来源于stack exchange，提问作者Pranav Shah