Java連接MQTT服務-wss方式
阿新 • • 發佈：2017-11-06
logs ber 修改 get row clean let ref token
特別提示：本人博客部分有參考網絡其他博客，但均是本人親手編寫過並驗證通過。如發現博客有錯誤，請及時提出以免誤導其他人，謝謝！歡迎轉載，但記得標明文章出處：http://www.cnblogs.com/mao2080/
說明：前面介紹的tcp、ws方式適合Java程序在局域網內使用，不涉及到安全問題。但由於Android手機APP需要通過websocket方式來連接,就必須考慮安全性問題了，這時候就采用了wss+CA證書方式進行認證，而且在數據傳輸中也是加密的。大致與ws方式相同，只不過是加了證書。
1、Java代碼
1 package com.mao.mqtt; 2 3 import java.io.FileInputStream;4 import java.security.KeyStore; 5 import java.security.cert.CertificateFactory; 6 import java.security.cert.X509Certificate; 7 import java.text.SimpleDateFormat; 8 import java.util.Date; 9 10 import javax.net.ssl.SSLContext; 11 import javax.net.ssl.SSLSocketFactory; 12 import javax.net.ssl.TrustManagerFactory;13 14 import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; 15 import org.eclipse.paho.client.mqttv3.MqttCallback; 16 import org.eclipse.paho.client.mqttv3.MqttClient; 17 import org.eclipse.paho.client.mqttv3.MqttConnectOptions; 18 import org.eclipse.paho.client.mqttv3.MqttException; 19 importorg.eclipse.paho.client.mqttv3.MqttMessage; 20 import org.eclipse.paho.client.mqttv3.MqttTopic; 21 22 /** 23 * 24 * 功能描述：MQTT測試 25 * 創建人： [email protected] 26 * 創建時間：2017年7月4日 下午5:08:59 27 * 修改人： [email protected] 28 * 修改時間：2017年7月4日 下午5:08:59 29 */ 30 public class MQTTTest_wss { 31 32 /**MQTT服務端ip及端口*/ 33 private static String host = "wss://ip:443"; 34 35 /**賬號*/ 36 private static String username = "li2080"; 37 38 /**密碼*/ 39 private static String password = "123"; 40 41 /**訂閱的主題*/ 42 private static String subTopic = "a/b/c"; 43 44 /**clientID*/ 45 private static String clientId = "li2080"; 46 47 /**發布的主題*/ 48 private static String pubTopic = "a/b/c"; 49 50 /**MQTT-Client*/ 51 private static MqttClient client; 52 53 /**證書路徑*/ 54 private static String caPath = "E:\\mqtt-demo\\certfile\\CA.crt"; 55 56 /** 57 * @throws InterruptedException 58 * @throws MqttException */ 59 public static void main(String[] args) throws InterruptedException, MqttException { 60 61 // 訂閱消息的方法 62 subscribe(); 63 // 64 publish(); 65 } 66 67 /** 68 * 69 * 描述：訂閱信息 70 * @author [email protected] 71 * @created 2017年7月4日 下午4:53:47 72 * @since 73 * @return 74 */ 75 public static void subscribe() { 76 try { 77 // 創建MqttClient 78 MQTTTest_wss.getClient().setCallback(new MqttCallback() { 79 80 public void connectionLost(Throwable arg0) { 81 82 } 83 84 public void messageArrived(String topic, MqttMessage message) throws Exception { 85 System.out.println("MQTT Rece:" + message.toString()); 86 } 87 88 public void deliveryComplete(IMqttDeliveryToken token) { 89 90 } 91 92 }); 93 MQTTTest_wss.getClient().subscribe(subTopic, 0); 94 System.out.println("連接狀態:" + client.isConnected()); 95 } catch (Exception e) { 96 e.printStackTrace(); 97 } 98 } 99 100 /** 101 * 102 * 描述：獲取MqttClient 103 * @author [email protected] 104 * @created 2017年7月6日 上午9:56:37 105 * @since 106 * @return 107 * @throws MqttException 108 */ 109 public static MqttClient getClient() throws MqttException{ 110 try { 111 if(client == null){ 112 client = new MqttClient(host, clientId); 113 MqttConnectOptions conOptions = new MqttConnectOptions(); 114 conOptions.setUserName(username); 115 conOptions.setPassword(password.toCharArray()); 116 conOptions.setCleanSession(true); 117 conOptions.setSocketFactory(getSSLSocktet(caPath)); 118 client.connect(conOptions); 119 } 120 if(!client.isConnected()){ 121 client.reconnect(); 122 } 123 } catch (Exception e) { 124 e.printStackTrace(); 125 } 126 return client; 127 } 128 129 /** 130 * 131 * 描述：發布信息 132 * @author [email protected] 133 * @throws MqttException 134 * @created 2017年7月4日 下午4:53:32 135 * @since 136 */ 137 public static void publish() throws MqttException { 138 SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); 139 String sendMsg = "{time:"+sdf.format(new Date())+", content:"+com.lds.iot.common.util.UUIDUtil.getLowerLetterNumber(15)+", from: java console}"; 140 try { 141 MqttTopic topic = MQTTTest_wss.getClient().getTopic(pubTopic); 142 MqttMessage message = new MqttMessage(sendMsg.getBytes()); 143 message.setQos(0); 144 topic.publish(message); 145 System.out.println("MQTT Send:" + sendMsg); 146 } catch (Exception e) { 147 e.printStackTrace(); 148 } 149 } 150 151 /** 152 * 獲取SSLSocketFactory 153 * @param caPath 154 * @return 155 * @throws Exception 156 */ 157 public static SSLSocketFactory getSSLSocktet(String caPath) throws Exception { 158 CertificateFactory cAf = CertificateFactory.getInstance("X.509"); 159 FileInputStream caIn = new FileInputStream(caPath); 160 X509Certificate ca = (X509Certificate) cAf.generateCertificate(caIn); 161 KeyStore caks = KeyStore.getInstance(KeyStore.getDefaultType()); 162 caks.load(null, null); 163 caks.setCertificateEntry("ca-certificate", ca); 164 TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()); 165 tmf.init(caks); 166 caIn.close(); 167 SSLContext context = SSLContext.getInstance("TLSv1"); 168 context.init(null, tmf.getTrustManagers(), null); 169 return context.getSocketFactory(); 170 } 171 172 }
2、Maven配置
1 <dependency> 2 <groupId>org.eclipse.paho</groupId> 3 <artifactId>org.eclipse.paho.client.mqttv3</artifactId> 4 <version>1.2.0</version> 5 </dependency>
3、運行效果
Java連接MQTT服務-wss方式