<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;  
 
    /**
    *   獲取客戶端實(shí)例
    *   單例模式, 存在則返回, 不存在則初始化
    */
    public static MqttClient getInstance() {    
        if (client == null) {      
            init();    
        }    
        return client;  
    }  
    
    /**
    *   初始化客戶端
    */
    public static void init() {    
        try {      
            client = new MqttClient(configuration.getAddress(), "client-" + System.currentTimeMillis());      
            // MQTT配置對(duì)象
            MqttConnectOptions options = new MqttConnectOptions();      
            // 設(shè)置自動(dòng)重連, 其它具體參數(shù)可以查看MqttConnectOptions
            options.setAutomaticReconnect(true);      
            if (!client.isConnected()) {        
            client.connect(options);      
            }    
        } catch (MqttException e) {      
            LOGGER.error(String.format("MQTT: 連接消息服務(wù)器[%s]失敗", configuration.getAddress()));    
        }  
    }
    
}

public class MqttUtil {  

    /**
    *   發(fā)送消息
    *   @param topic 主題
    *   @param data 消息內(nèi)容
    */
    public static void send(String topic, Object data) {    
        // 獲取客戶端實(shí)例
        MqttClient client = MqttFactory.getInstance();    
        ObjectMapper mapper = new ObjectMapper();    
        try {
            // 轉(zhuǎn)換消息為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]發(fā)送消息轉(zhuǎn)換json失敗", topic));    
        } catch (MqttException e) {      
            LOGGER.error(String.format("MQTT: 主題[%s]發(fā)送消息失敗", topic));    
        }  
    }
    
    /** 
    * 訂閱主題 
    * @param topic 主題 
    * @param listener 消息監(jiān)聽(tīng)處理器 
    */
    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]發(fā)來(lái)消息[%s]", topic, new String(mqttMessage.getPayload())));  
    }
    
    public static void main(String[] args) {  
        //訂閱主題test01, 使用MessageListener來(lái)處理它的消息
        MqttUtil.subscribe("test01", new MessageListener());
    }

}