如何在Spring Boot中使用MQTT

<dependency>  
    <groupId>org.springframework.integration</groupId>  
    <artifactId>spring-integration-mqtt</artifactId>  
    <version>5.3.2.RELEASE</version>  
</dependency>

public class MqttFactory {  

    private static MqttProperties configuration;  
    
    private static MqttClient client;  
 
    /**
    *   获取客户端实例
    *   单例模式, 存在则返回, 不存在则初始化
    */
    public static MqttClient getInstance() {    
        if (client == null) {      
            init();    
        }    
        return client;  
    }  
    
    /**
    *   初始化客户端
    */
    public static void init() {    
        try {      
            client = new MqttClient(configuration.getAddress(), "client-" + System.currentTimeMillis());      
            // MQTT配置对象
            MqttConnectOptions options = new MqttConnectOptions();      
            // 设置自动重连, 其它具体参数可以查看MqttConnectOptions
            options.setAutomaticReconnect(true);      
            if (!client.isConnected()) {        
            client.connect(options);      
            }    
        } catch (MqttException e) {      
            LOGGER.error(String.format("MQTT: 连接消息服务器[%s]失败", configuration.getAddress()));    
        }  
    }
    
}

public class MqttUtil {  

    /**
    *   发送消息
    *   @param topic 主题
    *   @param data 消息内容
    */
    public static void send(String topic, Object data) {    
        // 获取客户端实例
        MqttClient client = MqttFactory.getInstance();    
        ObjectMapper mapper = new ObjectMapper();    
        try {
            // 转换消息为json字符串
            String json = mapper.writeValueAsString(data);      
            client.publish(topic, new MqttMessage(json.getBytes(StandardCharsets.UTF_8)));    
        } catch (JsonProcessingException e) {      
            LOGGER.error(String.format("MQTT: 主题[%s]发送消息转换json失败", topic));    
        } catch (MqttException e) {      
            LOGGER.error(String.format("MQTT: 主题[%s]发送消息失败", topic));    
        }  
    }
    
    /** 
    * 订阅主题 
    * @param topic 主题 
    * @param listener 消息监听处理器 
    */
    public static void subscribe(String topic, IMqttMessageListener listener) {  
        MqttClient client = MqttFactory.getInstance();  
        try {    
            client.subscribe(topic, listener);  
        } catch (MqttException e) {    
            LOGGER.error(String.format("MQTT: 订阅主题[%s]失败", topic));  
        }
    }
    
}

public class MessageListener implements IMqttMessageListener {  

    /** 
    * 处理消息
    * @param topic 主题 
    * @param mqttMessage 消息 
    */
    @Override  
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {   
        LOGGER.info(String.format("MQTT: 订阅主题[%s]发来消息[%s]", topic, new String(mqttMessage.getPayload())));  
    }
    
    public static void main(String[] args) {  
        //订阅主题test01, 使用MessageListener来处理它的消息
        MqttUtil.subscribe("test01", new MessageListener());
    }

}