IoT Hub

IoT Hub幫助裝置連線阿里雲IoT服務，IoT Hub支援PUB/SUB與RRPC兩種通訊方式，其中PUB/SUB是基於Topic進行的訊息路由。IoT Hub具有以下特點：

• 高效能擴充套件：支援線性動態擴充套件，可以支撐十億裝置同時連線
• 全鏈路加密：整個通訊鏈路以RSA，AES加密，保證資料傳輸安全
• 訊息實時到達：當裝置與IoT Hub成功建立資料通道後，兩者間將保持長連線，以減少握手時間，保證訊息實時達到
• 支援資料透傳：IoT Hub支援將資料以二進位制透傳的方式傳到自己伺服器上，不儲存裝置資料，從而保證資料的安全可控性
• 支援多種通訊協議：IoT Hub支援RRPC和PUB/SUB兩種通訊模式
• 支援多種裝置接入協議：支援裝置使用CoAP、MQTT、HTTPS協議接入物聯網平臺

阿里雲物接入，IoT服務建立

接下來介紹以下阿里雲物聯網套件接入情況，首先要開通物聯網套件服務，阿里雲的物聯網套件的收費情況和百度天工類似，都是以流量（訊息條數進行收費的），並且每月百萬條內的訊息都是免費，因此研究成本很低，只要花時間即可。
第一步，建立產品，阿里雲的產品泛指一批具有相同功能的裝置，建立產品是為了批量管理裝置。建立產品分基礎版和高階版，高階版有更多的功能，可以考慮直接建立高階版。節點型別分裝置和閘道器，裝置表示下面掛載的是裝置，閘道器則掛載的是閘道器，閘道器下面可以繼續掛載裝置，閘道器相當於多了一個分組。

第二步，給產品新增Topic，阿里雲預設給了三個Topic，當然我們也可以定義自己的Topic類，在產品檢視頁面，選擇訊息通訊，就可以定義Topic類了。定義Topic時，和百度天工定義策略一樣，也需要指定許可權，這裡也是有兩種：釋出、訂閱

第三步，建立裝置，建立時首選要選擇一個產品，裝置是掛在產品下面的。然後就可以建立裝置了，裝置建立成功後，介面會彈出一個三元組資料提示：ProductKey（建立產品時生成）、DeviceName（建立裝置時填寫）、DeviceSecret（裝置建立成功後生成，可以在裝置中修改），在MQTT的客戶端端，我們進行裝置接入時，需要通過這個三元組進行認證。

終端裝置物接入

阿里雲的IoT接入可以自己使用MQTT協議自己實現，也可以使用阿里雲提供的SDK包，目前阿里雲支援SDK的版本有：Java、Python、PHP和.Net，其實SDK裡面也是對標準MQTT協議的封裝而已。

假如自己實現，maven依賴：

<dependency>
   <groupId>org.eclipse.paho</groupId>
   <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
   <version>1.1.0</version>
</dependency>
<dependency>
   <groupId>com.alibaba</groupId>
   <artifactId>fastjson</artifactId>
   <version>1.2.28</version>
</dependency>

簡單客戶端連線IoT實現：

/**
 * aliyun.com Inc.
 * Copyright (c) 2004-2017 All Rights Reserved.
 */
package com.aliyun.iot.demo.iothub;

import java.net.InetAddress;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
import java.util.concurrent.TimeUnit;

import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManager;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import com.aliyun.iot.util.LogUtil;
import com.aliyun.iot.util.SignUtil;

/**
 * IoT套件JAVA版裝置接入demo
 */
public class SimpleClient4IOT {

   /******這裡是客戶端需要的引數即前面說的三元組*******/
    public static String deviceName = "thing001";
    public static String productKey = "b1s2DQ7udsJ";
    public static String secret = "C9LncnlHEYAErdxM0GnubaOURApmrRDle";

    //用於測試的topic
    private static String subTopic = "/" + productKey + "/" + deviceName + "/get";
    private static String pubTopic = "/" + productKey + "/" + deviceName + "/update";

    public static void main(String... strings) throws Exception {
        //客戶端裝置自己的一個標記，建議是MAC或SN，不能為空，32字元內
        String clientId = InetAddress.getLocalHost().getHostAddress();

        //裝置認證
        Map<String, String> params = new HashMap<String, String>();
        params.put("productKey", productKey); //這個是對應使用者在控制檯註冊的 裝置productkey
        params.put("deviceName", deviceName); //這個是對應使用者在控制檯註冊的 裝置name
        params.put("clientId", clientId);
        String t = System.currentTimeMillis() + "";
        params.put("timestamp", t);

        //MQTT伺服器地址，TLS連線使用ssl開頭
        String targetServer = "ssl://" + productKey + ".iot-as-mqtt.cn-shanghai.aliyuncs.com:1883";

        //客戶端ID格式，兩個||之間的內容為裝置端自定義的標記，字元範圍[0-9][a-z][A-Z]
        String mqttclientId = clientId + "|securemode=2,signmethod=hmacsha1,timestamp=" + t + "|";
        String mqttUsername = deviceName + "&" + productKey; //mqtt使用者名稱格式
        String mqttPassword = SignUtil.sign(params, secret, "hmacsha1"); //簽名

        System.err.println("mqttclientId=" + mqttclientId);

        connectMqtt(targetServer, mqttclientId, mqttUsername, mqttPassword, deviceName);
    }

    public static void connectMqtt(String url, String clientId, String mqttUsername,
                                   String mqttPassword, final String deviceName) throws Exception {
        MemoryPersistence persistence = new MemoryPersistence();
        SSLSocketFactory socketFactory = createSSLSocket();
        final MqttClient sampleClient = new MqttClient(url, clientId, persistence);
        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setMqttVersion(4); // MQTT 3.1.1
        connOpts.setSocketFactory(socketFactory);

        //設定是否自動重連
        connOpts.setAutomaticReconnect(true);

        //如果是true，那麼清理所有離線訊息，即QoS1或者2的所有未接收內容
        connOpts.setCleanSession(false);

        connOpts.setUserName(mqttUsername);
        connOpts.setPassword(mqttPassword.toCharArray());
        connOpts.setKeepAliveInterval(65);

        LogUtil.print(clientId + "進行連線, 目的地: " + url);
        sampleClient.connect(connOpts);

        sampleClient.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {
                LogUtil.print("連線失敗,原因:" + cause);
                cause.printStackTrace();
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                LogUtil.print("接收到訊息,來至Topic [" + topic + "] , 內容是:["
                    + new String(message.getPayload(), "UTF-8") + "],  ");
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                //如果是QoS0的訊息，token.resp是沒有回覆的
                LogUtil.print("訊息傳送成功! " + ((token == null || token.getResponse() == null) ? "null"
                    : token.getResponse().getKey()));
            }
        });
        LogUtil.print("連線成功:---");

        //這裡測試傳送一條訊息
        String content = "{'content':'msg from :" + clientId + "," + System.currentTimeMillis() + "'}";

        MqttMessage message = new MqttMessage(content.getBytes("utf-8"));
        message.setQos(0);
        //System.out.println(System.currentTimeMillis() + "訊息釋出:---");
        sampleClient.publish(pubTopic, message);

        //一次訂閱永久生效 
        //這個是第一種訂閱topic方式，回撥到統一的callback
        sampleClient.subscribe(subTopic);

        //這個是第二種訂閱方式, 訂閱某個topic，有獨立的callback
        //sampleClient.subscribe(subTopic, new IMqttMessageListener() {
        //    @Override
        //    public void messageArrived(String topic, MqttMessage message) throws Exception {
        //
        //        LogUtil.print("收到訊息：" + message + ",topic=" + topic);
        //    }
        //});

        //回覆RRPC響應
        final ExecutorService executorService = new ThreadPoolExecutor(2,
            4, 600, TimeUnit.SECONDS,
            new ArrayBlockingQueue<Runnable>(100), new CallerRunsPolicy());

        String reqTopic = "/sys/" + productKey + "/" + deviceName + "/rrpc/request/+";
        sampleClient.subscribe(reqTopic, new IMqttMessageListener() {
            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                LogUtil.print("收到請求：" + message + ", topic=" + topic);
                String messageId = topic.substring(topic.lastIndexOf('/') + 1);
                final String respTopic = "/sys/" + productKey + "/" + deviceName + "/rrpc/response/" + messageId;
                String content = "hello world";
                final MqttMessage response = new MqttMessage(content.getBytes());
                response.setQos(0); //RRPC只支援QoS0
                //不能在回撥執行緒中呼叫publish，會阻塞執行緒，所以使用執行緒池
                executorService.submit(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            sampleClient.publish(respTopic, response);
                            LogUtil.print("回覆響應成功，topic=" + respTopic);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                });
            }
        });
    }

    private static SSLSocketFactory createSSLSocket() throws Exception {
        SSLContext context = SSLContext.getInstance("TLSV1.2");
        context.init(null, new TrustManager[] {new ALiyunIotX509TrustManager()}, null);
        SSLSocketFactory socketFactory = context.getSocketFactory();
        return socketFactory;
    }
}