快速使用MQTT的Java SDK收發消息（終端和云端消息收發）
更新時間：
本文介紹如何快速使用云消息隊列 MQTT 版的Java SDK實現MQTT終端和云端服務的消息收發。
前提條件
背景信息
本文以公網環境為例，介紹使用Java SDK實現消息收發。更多消息收發示例代碼，請參見接入點說明
終端和云端服務與云消息隊列 MQTT 版通信時，需要在各自的SDK代碼中設置云消息隊列 MQTT 版實例的接入點信息，通過接入點和云消息隊列 MQTT 版服務端連接。
終端SDK接入點格式
使用終端SDK接入云消息隊列 MQTT 版時，需要填寫的接入點格式如下：
公網接入點：
MQTT實例ID.mqtt.aliyuncs.comVPC 接入點：
MQTT實例ID-internal-vpc.mqtt.aliyuncs.com
終端SDK接入點也可以直接在云消息隊列 MQTT 版控制臺實例詳情頁面的接入點頁簽中查看。
云端SDK接入點格式
使用云端SDK接入云消息隊列 MQTT 版時，需要填寫的接入點格式如下：
重要僅實例地域屬于中國內地的實例支持云端SDK接入。
公網接入點：
MQTT實例ID-server-internet.mqtt.aliyuncs.comVPC 接入點：
MQTT實例ID-server-internal.mqtt.aliyuncs.com
說明
MQTT實例ID可在云消息隊列 MQTT 版控制臺實例詳情頁面的基礎信息區域查看。
終端SDK接入點和云端SDK接入點同時支持公網接入點和VPC 接入點。公網接入點為本地公網環境訪問的IP地址，一般用于物聯網和移動互聯網場景中；VPC 接入點為云上私網訪問的IP地址，一般用于云端應用接入云消息隊列 MQTT 版。
重要
SDK使用接入點連接服務時務必使用域名接入，不得直接使用域名背后的IP地址直接連接，因為IP地址隨時會變化。在以下使用情況中出現的問題云消息隊列 MQTT 版產品方概不負責：
終端或云端不使用域名接入而是使用IP地址接入，產品方更新了域名解析導致原有IP地址失效。
終端或云端網絡側對IP地址設置網絡防火墻策略，產品方更新了域名解析后新IP地址被您的防火墻策略攔截。
調用終端SDK發送消息
- 下載第三方的開源Java SDK。下載地址為Eclipse Paho Java Client。
下載終端SDK的Demo示例作為您代碼開發的參考。下載地址為mqtt-java-demo。
- 解壓該Demo工程包至您指定的文件夾。
在IntelliJ IDEA中，導入解壓后的文件以創建相應的工程，并確認pom.xml中已包含以下依賴。
<dependencies> <dependency> <groupId>commons-codec</groupId> <artifactId>commons-codec</artifactId> <version>1.10</version> </dependency> <dependency> <groupId>org.eclipse.paho</groupId> <artifactId>org.eclipse.paho.client.mqttv3</artifactId> <version>1.2.2</version> </dependency> <dependency> <groupId>org.apache.httpcomponents</groupId> <artifactId>httpclient</artifactId> <version>4.5.2</version> </dependency> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>1.2.83</version> </dependency> <dependency> <groupId>com.aliyun</groupId> <artifactId>aliyun-java-sdk-onsmqtt</artifactId> <version>1.0.3</version> </dependency> <dependency> <groupId>com.aliyun</groupId> <artifactId>aliyun-java-sdk-core</artifactId> <version>4.5.0</version> </dependency> </dependencies>在MQ4IoTProducerDemo.java類中，按代碼注釋說明填寫相應參數，主要涉及您已在創建資源中所創建的MQTT資源信息。然后執行Main函數運行代碼完成消息發送。
示例代碼如下。
說明在使用示例代碼前，需要配置環境變量，通過環境變量讀取訪問憑證。關于配置環境變量的方法，請參見配置訪問憑證。
云消息隊列 MQTT 版的AccessKey ID和AccessKey Secret的環境變量名稱分別為MQTT_AK_ENV和MQTT_SK_ENV。
package com.aliyun.openservices.lmq.example.demo; import com.aliyun.openservices.lmq.example.util.ConnectionOptionWrapper; import java.util.concurrent.ExecutorService; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; import org.eclipse.paho.client.mqttv3.MqttCallbackExtended; import org.eclipse.paho.client.mqttv3.MqttClient; import org.eclipse.paho.client.mqttv3.MqttException; import org.eclipse.paho.client.mqttv3.MqttMessage; import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; public class MQ4IoTProducerDemo { public static void main(String[] args) throws Exception { /** * 您創建的云消息隊列 MQTT 版的實例ID。 */ String instanceId = "XXXXX"; /** * 設置終端SDK的接入點，進入云消息隊列 MQTT 版控制臺實例詳情頁面的接入點頁簽查看。 * 接入點地址必須填寫分配的域名，不得使用IP地址直接連接，否則可能會導致客戶端異常。 */ String endPoint = "XXXXX.mqtt.aliyuncs.com"; /** * AccessKey ID，阿里云身份驗證，在阿里云RAM控制臺創建。 * 阿里云賬號AccessKey擁有所有API的訪問權限，建議您使用RAM用戶進行API訪問或日常運維。 * 強烈建議不要把AccessKey ID和AccessKey Secret保存到工程代碼里，否則可能導致AccessKey泄露，威脅您賬號下所有資源的安全。 * 本示例以將AccessKey 和 AccessKeySecret 保存在環境變量為例說明。 */ String accessKey = System.getenv("MQTT_AK_ENV"); /** * AccessKey Secret，阿里云身份驗證，在阿里云RAM控制臺創建。僅在簽名鑒權模式下需要設置。 */ String secretKey = System.getenv("MQTT_SK_ENV"); /** * MQTT客戶端ID，由業務系統分配，需要保證每個TCP連接都不一樣，保證全局唯一，如果不同的客戶端對象（TCP連接）使用了相同的clientId會導致連接異常斷開。 * clientId由兩部分組成，格式為GroupID@@@DeviceID，其中GroupID在云消息隊列 MQTT 版控制臺創建，DeviceID由業務方自己設置，clientId總長度不得超過64個字符。 */ String clientId = "GID_XXXXX@@@XXXXX"; /** * 云消息隊列 MQTT 版消息的一級Topic，需要在控制臺創建才能使用。 * 如果使用了沒有創建或者沒有被授權的Topic會導致鑒權失敗，服務端會斷開客戶端連接。 */ final String parentTopic = "XXXXX"; /** * 云消息隊列 MQTT 版支持子級Topic，用來做自定義的過濾，此處為示例，可以填寫任意字符串。 * 需要注意的是，完整的Topic長度不得超過128個字符。 */ final String mq4IotTopic = parentTopic + "/" + "testMq4Iot"; /** * QoS參數代表傳輸質量，可選0，1，2。詳細信息，請參見名詞解釋。 */ final int qosLevel = 0; ConnectionOptionWrapper connectionOptionWrapper = new ConnectionOptionWrapper(instanceId, accessKey, secretKey, clientId); final MemoryPersistence memoryPersistence = new MemoryPersistence(); /** * 客戶端協議和端口。客戶端使用的協議和端口必須匹配，如果是SSL加密則設置ssl://endpoint:8883。 */ final MqttClient mqttClient = new MqttClient("tcp://" + endPoint + ":1883", clientId, memoryPersistence); /** * 設置客戶端發送超時時間，防止無限阻塞。 */ mqttClient.setTimeToWait(5000); final ExecutorService executorService = new ThreadPoolExecutor(1, 1, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); mqttClient.setCallback(new MqttCallbackExtended() { @Override public void connectComplete(boolean reconnect, String serverURI) { /** * 客戶端連接成功后就需要盡快訂閱需要的Topic。 */ System.out.println("connect success"); } @Override public void connectionLost(Throwable throwable) { throwable.printStackTrace(); } @Override public void messageArrived(String s, MqttMessage mqttMessage) throws Exception { /** * 消費消息的回調接口，需要確保該接口不拋異常，該接口運行返回即代表消息消費成功。 * 消費消息需要保證在規定時間內完成，如果消費耗時超過服務端約定的超時時間，對于可靠傳輸的模式，服務端可能會重試推送，業務需要做好冪等去重處理。 */ System.out.println( "receive msg from topic " + s + " , body is " + new String(mqttMessage.getPayload())); } @Override public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) { System.out.println("send msg succeed topic is : " + iMqttDeliveryToken.getTopics()[0]); } }); mqttClient.connect(connectionOptionWrapper.getMqttConnectOptions()); for (int i = 0; i < 10; i++) { MqttMessage message = new MqttMessage("hello mq4Iot pub sub msg".getBytes()); message.setQos(qosLevel); /** * 發送普通消息時，Topic必須和接收方訂閱的Topic一致，或者符合通配符匹配規則。 */ mqttClient.publish(mq4IotTopic, message); /** * 云消息隊列 MQTT 版支持點對點消息，即如果發送方明確知道該消息只需要給特定的一個設備接收，且知道對端的clientId，則可以直接發送點對點消息。 * 點對點消息不需要經過訂閱關系匹配，可以簡化訂閱方的邏輯。點對點消息的Topic格式規范是 {{parentTopic}}/p2p/{{targetClientId}}。 */ String receiverId = "xxx"; final String p2pSendTopic = parentTopic + "/p2p/" + receiverId; message = new MqttMessage("hello mq4Iot p2p msg".getBytes()); message.setQos(qosLevel); mqttClient.publish(p2pSendTopic, message); } Thread.sleep(Long.MAX_VALUE); } }
調用云端SDK接收消息
下載云消息隊列 MQTT 版提供的云端SDK。下載地址為云端SDK版本說明。
下載云端SDK的Demo示例做為您代碼開發的參考。下載地址為mqtt-server-sdk-demo。
- 解壓該Demo工程包至您指定的文件夾。
在IntelliJ IDEA中，導入解壓后的文件以創建相應的工程，并確認pom.xml中已包含以下依賴。
<dependencies> <dependency> <groupId>com.alibaba.mqtt</groupId> <artifactId>server-sdk</artifactId> <version>1.0.0.Final</version> </dependency> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>1.2.83</version> </dependency> </dependencies>在MQTTConsumerDemo.java類中，按代碼注釋說明填寫相應參數，主要涉及您已在創建資源中所創建好的MQTT資源信息。然后執行Main函數運行代碼完成消息接收。
示例代碼如下。
說明在使用示例代碼前，需要配置環境變量，通過環境變量讀取訪問憑證。關于配置環境變量的方法，請參見配置訪問憑證。
云消息隊列 MQTT 版的AccessKey ID和AccessKey Secret的環境變量名稱分別為MQTT_AK_ENV和MQTT_SK_ENV。
package com.aliyun.openservices.lmq.example; import com.alibaba.fastjson.JSONObject; import com.alibaba.mqtt.server.ServerConsumer; import com.alibaba.mqtt.server.callback.MessageListener; import com.alibaba.mqtt.server.config.ChannelConfig; import com.alibaba.mqtt.server.config.ConsumerConfig; import com.alibaba.mqtt.server.model.MessageProperties; public class MQTTConsumerDemo { public static void main(String[] args) throws Exception { /** * 設置云端SDK的接入點，請參見接入點說明中的云端SDK接入點格式。 * 接入點地址必須填寫分配的域名，不得使用IP地址直接連接，否則可能會導致服務端異常。 */ String domain = "post-cn-jaj3h8i****.mqtt.aliyuncs.com"; /** * 使用的協議和端口必須匹配，該參數值固定為5672。 */ int port = "5672"; /** * 您創建的云消息隊列 MQTT 版的實例ID。 */ String instanceId = "post-cn-jaj3h8i****"; /** * AccessKey ID，阿里云身份驗證，在阿里云RAM控制臺創建。 * 阿里云賬號AccessKey擁有所有API的訪問權限，建議您使用RAM用戶進行API訪問或日常運維。 * 強烈建議不要把AccessKey ID和AccessKey Secret保存到工程代碼里，否則可能導致AccessKey泄露，威脅您賬號下所有資源的安全。 * 本示例以將AccessKey 和 AccessKeySecret 保存在環境變量為例說明。 */ String accessKey = System.getenv("MQTT_AK_ENV"); /** * AccessKey Secret，阿里云身份驗證，在阿里云RAM控制臺創建。僅在簽名鑒權模式下需要設置。 */ String secretKey = System.getenv("MQTT_SK_ENV"); /** * 云消息隊列 MQTT 版消息的一級Topic，需要在控制臺創建才能使用。 * 由于云端SDK訂閱消息一般用于云上應用進行消息匯總和分析等場景，因此，云端SDK訂閱消息不支持設置子級Topic。 * 如果使用了沒有創建或者沒有被授權的Topic會導致鑒權失敗，服務端會斷開客戶端連接。 */ String firstTopic = "firstTopic"; ChannelConfig channelConfig = new ChannelConfig(); channelConfig.setDomain(domain); channelConfig.setPort(port); channelConfig.setInstanceId(instanceId); channelConfig.setAccessKey(accessKey); channelConfig.setSecretKey(secretKey); ServerConsumer serverConsumer = new ServerConsumer(channelConfig, new ConsumerConfig()); serverConsumer.start(); serverConsumer.subscribeTopic(firstTopic, new MessageListener() { @Override public void process(String msgId, MessageProperties messageProperties, byte[] payload) { System.out.println("Receive:" + msgId + "," + JSONObject.toJSONString(messageProperties) + "," + new String(payload)); } }); } }說明云端SDK消息發送的示例代碼，請參見MQTTProducerDemo.java。
文檔內容是否對您有幫助？