139      阿裏雲  技術社區[雲棲]

SSL 方式接入示例__Java 接入示例_MQTT 接入(物聯)_消息隊列 MQ-阿裏雲

本文主要介紹如何使用 MQTT 的 SSL 加密通道來收發消息。

注意：

SSL 方式接入需要服務端支持，目前各個 Region 對 SSL 通道的支持情況請參考環境準備章節的文檔。

本文給出的示例均基於 Eclipse Paho Java SDK 實現，SDK 下載請參見 MQTT 接入準備。如使用其他第三方的客戶端，請適當修改。

1. 證書下載

如果需要使用 SSL 加密通道，需要首先下載服務端的安全證書到本地。證書下載請點擊證書下載。

2. 客戶端設置 SSL 屬性

在添加好本地的信任證書倉庫後，應用的代碼中即可設置該倉庫到 SSL 的屬性中。具體示例如下：

public static void main(String[] args) throws IOException {
        /**
         * 設置MQTT的接入點，請根據應用所在環境選擇合適的Region，不支持跨Region訪問
         */
        final String broker ="ssl://mqtt-test.cn-qingdao.aliyuncs.com:8883";
        /**
         * 設置阿裏雲的AccessKey，用於鑒權
         */
        final String acessKey ="XXXXXX";
        /**
         * 設置阿裏雲的SecretKey，用於鑒權
         */
        final String secretKey ="XXXXXXX";
        /**
         * 發消息使用的一級Topic，需要先在MQ控製台裏申請
         */
        final String topic ="XXXX";
        /**
         * MQTT的ClientID，一般由兩部分組成，GroupID@@@DeviceID
         * 其中GroupID在MQ控製台裏申請
         * DeviceID由應用方設置，可能是設備編號等，需要唯一，否則服務端拒絕重複的ClientID連接
         */
        final String clientId ="GID_XXX@@@ClientID_XXXX";
        String sign;
        MemoryPersistence persistence = new MemoryPersistence();
        try {
            final MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
            final MqttConnectOptions connOpts = new MqttConnectOptions();
            System.out.println("Connecting to broker: " + broker);
            /**
             * 計算簽名，將簽名作為MQTT的password。
             * 簽名的計算方法，參考工具類MacSignature，第一個參數是ClientID的前半部分，即GroupID
             * 第二個參數阿裏雲的SecretKey
             */
            sign = MacSignature.macSignature(clientId.split("@@@")[0], secretKey);
            connOpts.setUserName(acessKey);
            connOpts.setServerURIs(new String[] { broker });
            connOpts.setPassword(sign.toCharArray());
            connOpts.setCleanSession(false);
            connOpts.setKeepAliveInterval(100);
            /**
            *導入證書的路徑，生成加密連接
            */
            SocketFactory socketFactory = initSSLSocket("intermedia.crt");
            connOpts.setSocketFactory(socketFactory);
            sampleClient.setCallback(new MqttCallback() {
                public void connectionLost(Throwable throwable) {
                    System.out.println("mqtt connection lost");
                    throwable.printStackTrace();
                    while(!sampleClient.isConnected()){
                        try {
                            sampleClient.connect(connOpts);
                        } catch (MqttException e) {
                            e.printStackTrace();
                        }
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
                public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
                    System.out.println("messageArrived:" + topic + "------" + new String(mqttMessage.getPayload()));
                }
                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                    System.out.println("deliveryComplete:" + iMqttDeliveryToken.getMessageId());
                }
            });
            sampleClient.connect(connOpts);
            for (int i = 0; i < 10; i++) {
                try {
                    String scontent = new Date()+"MQTT Test body" + i;
                    final MqttMessage message = new MqttMessage(scontent.getBytes());
                    message.setQos(0);
                    System.out.println(i+" pushed at "+new Date()+" "+ scontent);
                    /**
                     *消息發送到某個主題Topic，所有訂閱這個Topic的設備都能收到這個消息
                     * 遵循MQTT的發布訂閱規範，Topic也可以是多級Topic。此處設置了發送到二級Topic
                     */
                    sampleClient.publish(topic+"/notice/", message);
                    /**
                     * 如果發送P2P消息，二級Topic必須是“p2p”，三級Topic是目標的ClientID
                     * 此處設置的三級Topic需要是接收方的ClientID
                     */
                    String p2pTopic =topic+"/p2p/GID_mqttdelay3@@@DEVICEID_001";
                    sampleClient.publish(p2pTopic,message);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        } catch (Exception me) {
            me.printStackTrace();
        }
    }

private static SSLSocketFactory initSSLSocket(String certFileName) throws Exception {
        InputStream caInput = new BufferedInputStream(ClassLoader.getSystemResourceAsStream(certFileName));
        CertificateFactory cf = CertificateFactory.getInstance("X.509");
        Certificate ca = null;
        try {
            ca = cf.generateCertificate(caInput);
        } catch (CertificateException e) {
            e.printStackTrace();
        } finally {
            caInput.close();
        }
        String keyStoreType = KeyStore.getDefaultType();
        KeyStore keyStore = KeyStore.getInstance(keyStoreType);
        keyStore.load(null, null);
        keyStore.setCertificateEntry("ca", ca);
        String tmfAlgorithm = TrustManagerFactory.getDefaultAlgorithm();
        TrustManagerFactory tmf = TrustManagerFactory.getInstance(tmfAlgorithm);
        tmf.init(keyStore);
        SSLContext context = SSLContext.getInstance("TLSV1.2");
        context.init(null, tmf.getTrustManagers(), null);
        SSLSocketFactory socketFactory = context.getSocketFactory();
        return socketFactory;
    }