1. 數(shù)據(jù)集成介紹

數(shù)據(jù)集成標準化的目標是規(guī)范應用之間數(shù)據(jù)的傳遞方式和表達方式。

• 傳遞方式：即應用之間的數(shù)據(jù)如何流通。平臺提供了對數(shù)據(jù)進行增刪改查的4個API，以及HTTP2方式的消息訂閱機制。

• 表達方式：即應用之間如何對數(shù)據(jù)內(nèi)容有一致的理解。為了實現(xiàn)這個目標，需要做到如下兩點，一是數(shù)據(jù)結(jié)構(gòu)需要由小二后臺統(tǒng)一管控；二是應用集成對接之前（比如應用上架的時候）需要聲明本應用對哪些數(shù)據(jù)模型產(chǎn)生什么樣的數(shù)據(jù)操作（如查詢、新增，或者訂閱）。

基于以上邏輯，為了實現(xiàn)應用的數(shù)據(jù)集成能力，分別需要執(zhí)行相應的應用聲明、應用開發(fā)、應用集成。

2. 應用聲明

應用的聲明是指應用在上架到市場時，由應用開發(fā)者自行聲明的，包含兩部分聲明內(nèi)容：應用涉及到的數(shù)據(jù)模型；以及對這些模型所做的相應的操作。這里指出應用對數(shù)據(jù)模型的操作，目的有兩個，一是用戶可以感知應用對數(shù)據(jù)的操作范圍，二是平臺根據(jù)聲明的操作決定對應用的操作權(quán)限。

3. 應用集成

所謂的“數(shù)據(jù)集成”，是指應用之間數(shù)據(jù)共享的方式，比如兩個應用，當涉及到相同的數(shù)據(jù)模型時，是共享一份數(shù)據(jù)，還是獨立管理各自的數(shù)據(jù)。通過應用聲明，應用在上架到應用市場之后，用戶就能感知應用與數(shù)據(jù)之間的關(guān)系。目前通用的集成方案是以項目為隔離維度的。也就是說，在一個項目內(nèi)的所有應用，他們關(guān)聯(lián)的相同的數(shù)據(jù)模型，會被默認放在同一個隔離區(qū)域ScopeID內(nèi)。

4. 應用開發(fā) - 數(shù)據(jù)操作

應用之間共享數(shù)據(jù)時，需要向提平臺操作數(shù)據(jù)，或者從平臺訂閱數(shù)據(jù)。數(shù)據(jù)操作方面，這里主要涉及4個接口：新增數(shù)據(jù)、查詢數(shù)據(jù)、刪除數(shù)據(jù)、修改數(shù)據(jù)。詳見后面的API參考。同時，為了便于應用開發(fā)，平臺提供了SDK，簡化開發(fā)。

4.1 SDK介紹

4.1.1 Java SDK

1. 依賴

依賴的三方庫：

<dependency>
  <groupId>com.aliyun.api.gateway</groupId>
  <artifactId>sdk-core-java</artifactId>
  <version>1.0.4</version>
</dependency>

依賴的工具類： 

git clone https://github.com/aliyun/iotx-api-gateway-client.git

2. 示例代碼

import com.alibaba.cloudapi.sdk.model.ApiResponse;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.aliyun.iotx.api.client.IoTApiClientBuilderParams;
import com.aliyun.iotx.api.client.IoTApiRequest;
import com.aliyun.iotx.api.client.SyncApiClient;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;

public class Demo {

    // 依賴的工具類從 https://github.com/aliyun/iotx-api-gateway-client 獲取
    public static void main(String[] args) throws UnsupportedEncodingException{
        String appKey = "YOUR_APP_KEY";
        String appSecret = "YOUR_APP_SECRET";
        String modelId = "YOUR_MODEL_ID";
        String dataId = "YOUR_DATA_ID";

        dataQuery(appKey, appSecret, modelId, dataId);
    }

     public static void dataQuery(String appKey, String appSecret,String modelId,String dataId) throws UnsupportedEncodingException {
         IoTApiClientBuilderParams ioTApiClientBuilderParams =
                 new IoTApiClientBuilderParams();
         // 填寫應用的appkey信息
         ioTApiClientBuilderParams.setAppKey(appKey);
         ioTApiClientBuilderParams.setAppSecret(appSecret);
         SyncApiClient syncClient = new SyncApiClient(ioTApiClientBuilderParams);
         IoTApiRequest request = new IoTApiRequest();
         // 設(shè)置api的版本
         request.setApiVer("0.0.5");
         // 接口參數(shù)
         request.putParam("modelId",modelId);
         ArrayList returnFields = new ArrayList();
         returnFields.add("*");
         request.putParam("returnFields", returnFields);

         JSONArray conditions = new JSONArray();
         JSONObject condition = new JSONObject();
         condition.put("fieldName", "id");
         condition.put("operate", "eq");
         condition.put("value", dataId);
         conditions.add(condition);
         request.putParam("conditions", conditions);

         ApiResponse response = syncClient.postBody("api.link.aliyun.com",
                 "/data/model/data/query", request, true);
         int code = response.getCode();
         String res = new String(response.getBody());
         System.out.println("response code = " + response.getCode() + " response = " + new String(response.getBody(), "UTF-8"));
     }
}

4.1.2 Python SDK

1. 依賴

依賴的三方庫：無

2. 示例代碼

Step 1: 下載示例工程代碼。示例工程基于Python 2.7，Python3請自行適配。

git clone https://github.com/aliyun/api-gateway-demo-sign-python.git

Step 2: 修改signature_composer.py文件第49行

修改前： string_to_sign.append(_build_resource(uri=uri, body=body))
修改后： string_to_sign.append(uri)

Step 3: 修改ClientDemo.py如下，執(zhí)行調(diào)用

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import os
import sys
import json

sys_path = sys.path[0]
if not os.path.isdir(sys_path):
    sys_path = os.path.dirname(sys_path) 
os.chdir(sys_path)

from com.aliyun.api.gateway.sdk import client
from com.aliyun.api.gateway.sdk.http import request
from com.aliyun.api.gateway.sdk.common import constant

host = "https://api.link.aliyun.com"
url = "/data/model/data/query"
cli = client.DefaultClient(app_key="YOUR_APP_KEY", app_secret="YOUR_APP_SECRET")
req_post = request.Request(host=host, protocol=constant.HTTPS, url=url, method="POST", time_out=30000)
req_post.set_content_type(constant.CONTENT_TYPE_STREAM)

body = {
    "request":{
        "apiVer":"0.0.2"
    },
    "params":{
        "modelId":"EMPLOYEE",
        "returnFields":["employee_no", "name"],
        "conditions":[{"fieldName":"id","value":"1","operate":"mt"}],
        "pageSize": 100,
        "pageNum": 1
    },
    "version":"1.0"
}
bb = json.dumps(body).encode("utf-8")

req_post.set_body(bb)

print '==================================================='
print str(cli.execute(req_post)).decode('string_escape')
print '==================================================='

4.1.3 JS SDK

1. 依賴

$ # save into package.json dependencies with -S
$ npm install aliyun-api-gateway -S
$ # you can use cnpm for fast install
$ cnpm install aliyun-api-gateway -S

2. 示例代碼

'use strict';
const co = require('co');
const Client = require('aliyun-api-gateway').Client;
const client = new Client('YOUR_APP_KEY','YOUR_APP_SECRET');

co(function* () {
  var url = 'http://api.link.aliyun.com/test1234';

  var result = yield client.post(url, {
    data: {
      'testtest': 'query1Value'
    },
    headers: {
      accept: 'application/json'
    }
  });

  console.log(JSON.stringify(result));
});

4.2 數(shù)據(jù)操作API

4.2.1 API 描述

4.2.2 API 約定

1. 日期類型的參數(shù)傳入格式：當前時間到格林威治時間1970年01月01日00時00分00秒的毫秒數(shù)。

2. 數(shù)量查詢的一些約定：

   1. 單次查詢最多返回200條數(shù)據(jù)，未指定分頁參數(shù)情況下，查詢返回滿足條件的前200條，可根據(jù)返回參數(shù)中的hasNext判斷是否有更多數(shù)據(jù)。

   2. 若需要按照指定條件返回數(shù)據(jù)總數(shù)，則指定返回參數(shù)為COUNT，API返回參數(shù)中會帶有COUNT以及對應的值。

3. 更新刪除的約定：單次操作，最多支持200條數(shù)據(jù)。

4. 運算符定義：

1. 數(shù)據(jù)包括系統(tǒng)屬性，API中不允許賦值和更新系統(tǒng)屬性，系統(tǒng)屬性如下：

4.2.3 數(shù)據(jù)新增API

1. 請求參數(shù)

2. 返回參數(shù)

3. 請求示例

{
    "request": {
        "apiVer": "0.0.3" // api版本號
    },
    "id": 1508212818676,// request里的全局唯一id透傳
    "params": {
       "modelId":"XXX123",  //數(shù)據(jù)模型id
       "scopeId":"fdbsdj1dfjdubgxxx",  //數(shù)據(jù)模型id
       "properties":{
            "BRAND":"BMW",
            "MODE":"5",
            "CREATE_DATE":1526969423,
            "LONG_SIZE":3.3,
            "LONG_SIZE":2.2              
        }

    },
    "version": "1.0" // 請求協(xié)議版本
}

4. 返回示例

{
  "code": 200,
  "message": "success",
  "localizedMsg": null,
  "data": 12345      
}

5. 返回碼

4.2.4 數(shù)據(jù)刪除API

1. 請求參數(shù)

其中condition針對不同類型的屬性支持的運算符：

2. 返回參數(shù)

3. 請求示例

{
    "request": {
        "apiVer": "0.1.0" // api版本號
    },
    "id": 1508212818676,// request里的全局唯一id透傳
    "params": {  
       "modelId":"XXX123",
       "conditions":[
          {"fieldName":"BRAND","value":"BMW","operate":"eq"},
          {"fieldName":"MODE","value":"X5","operate":"eq"}
        ]                
    },
    "version": "1.0" // 請求協(xié)議版本
}

4. 返回示例

{
  "code": 200,
  "message": "success",
  "localizedMsg": null,
  "data":100   
}

5. 返回碼

4.2.5 數(shù)據(jù)修改API

1. 請求參數(shù)

其中condition針對不同類型的屬性支持的運算符：

2. 返回參數(shù)

3. 請求示例

{
    "request": {
        "apiVer": "0.0.3" // api版本號
    },
    "id": 1508212818676,// request里的全局唯一id透傳
    "params": {  
       "modelId":"XXX123",
       "conditions":[
          {"fieldName":"BRAND","value":"BMW","operate":"eq"},
          {"fieldName":"MODE","value":"X5","operate":"eq"}
        ],
       "updateDetails":{
          "LONG_SIZE":4,
          "WIDTH_SIZE":3
        }                     
    },
    "version": "1.0" // 請求協(xié)議版本
}

4. 返回示例

{
  "code": 200,
  "message": "success",
  "localizedMsg": null,
  "data":100   
}

5. 返回碼

4.2.6 數(shù)據(jù)查詢API

1. 請求參數(shù)

其中condition針對不同類型的屬性支持的運算符：

2. 返回參數(shù)

3. 請求示例-1

{
    "request": {
        "apiVer": "0.0.2" // api版本號
    },
    "id": 1508212818676,// request里的全局唯一id透傳
    "params": {        
       "modelId":"XXX123",
       "returnFields":["MODE","ENGINE"],
       "conditions":[
          {"fieldName":"BRAND","value":"BMW","operate":"eq"},
          {"fieldName":"WIDTH_SIZE","value":1,"value2":2,"operate":"bt"},
          {"fieldName":"LONG_SIZE","value":"3","operate":"mt"}
        ],
       "orderBy":{
          "asc":true,
          "orderByFields":["CREATE_DATA","MODE"]
        },
        "pageNum":1,
        "pageSize":10
       }            
    },
    "version": "1.0" // 請求協(xié)議版本
}

4. 返回示例-1

{
    "code": 200,
    "localizedMsg": null,
    "data": "{\"count\":1,
                      \"hasNext\":false,
            \"items\":[{\"gmt_create\":1551872701000,\"MODE\":\"33\",\"ENGINE\":\"44\",\"id\":2,\"gmt_modified\":1551872714000}],
            \"pageNum\":1,
            \"pageSize\":10}",
    "message": "success"
}

5. 請求示例-2

{
    "request": {
        "apiVer": "0.1.0" // api版本號
    },
    "id": 1508212818676,// request里的全局唯一id透傳
    "params": {        
       "modelId":"XXX123",
       "returnFields":["COUNT",
       "conditions":[
          {"fieldName":"BRAND","value":"BMW","operate":"eq"},
          {"fieldName":"WIDTH_SIZE","value":1,"value2":2,"operate":"bt"},
          {"fieldName":"LONG_SIZE","value":"3","operate":"mt"}
        ],
        "pageNum":1,
        "pageSize":10
       }            
    },
    "version": "1.0" // 請求協(xié)議版本
}

6. 返回示例-2

{
    "code": 200,
    "localizedMsg": null,
    "data": "{
                      \"hasNext\":false,
            \"items\":[{
                    \"COUNT\":100
             }]
            }"
    "message": "success"
}

7. 返回碼

4.2.7 獲取文件上傳地址API

1. 請求參數(shù)

2. 返回參數(shù)

3. 請求示例

{
    "request": {
        "apiVer": "0.0.1"   // api版本號
    },
    "id": 1508212818671,    // request里的全局唯一id透傳
    "params": {
       "modelId":"XXX123",  //數(shù)據(jù)模型id
       "fileSize":"3.2",    //文件大小  
       "attrName":"name",   //屬性名稱
       "fileType":"jpg",    //文件類型 
       "version":"1.0",     //模型版本  
    },
    "version": "1.0"        // 請求協(xié)議版本
}

4. 返回示例

{
    "code": 200, // 返回是否成功，只要不是200說明返回不成功
    "message": "success",  // 如果失敗，會返回失敗的信息描述
    "localizedMsg": null,
    "data": {"url":"http://xxx.xxx.xx","fileName":"xxx.jpg"} // 返回的數(shù)據(jù)   
}

5. 返回碼

4.2.8 獲取文件下載地址API

1. 請求參數(shù)

2. 返回參數(shù)

3. 請求示例

{
    "request": {
        "apiVer": "0.0.1" // api版本號
    },
    "id": 1508212818671, // request里的全局唯一id透傳
    "params": {
       "modelId":"XXX123",  //數(shù)據(jù)模型id
       "scopeId":"fdbsdj1dfjdubgxxx",  //業(yè)務隔離id 
       "attrName":"name",  //屬性名稱
       "fileName":"xdsfxv.jpg",  //文件類型 
       "version":"1.0",  //模型版本  
    },
    "version": "1.0" // 請求協(xié)議版本
}

4. 返回示例

{
    "code": 200, // 返回是否成功，只要不是200說明返回不成功
    "message": "success",  // 如果失敗，會返回失敗的信息描述
    "localizedMsg": null,
    "data": {"url":"http://xxx.xxx.xx"} // 返回的數(shù)據(jù)下載查看的url   
}

5. 返回碼

5. 應用開發(fā) - 數(shù)據(jù)訂閱

應用可以通過AMQP方式，訂閱數(shù)據(jù)的變更消息（新增、刪除、修改）。應用自身不需要發(fā)布數(shù)據(jù)的變更消息到通道中，這些消息的產(chǎn)生是由應用通過上面的數(shù)據(jù)操作API，對數(shù)據(jù)進行操作之后，由平臺產(chǎn)生。平臺產(chǎn)生時間之后，會將消息通過訂閱關(guān)系發(fā)送到訂閱方。在這個消息通信中，每一個應用實例，由AppKey標識身份。對于單租戶型的應用，每一個AppKey代表了一次應用分發(fā)的實例；但是，對于SaaS應用，他的一個AppKey代表了該應用對應的所有租戶的身份，因此，在SaaS應用按照AppKey得到消息之后，需要自行根據(jù)訂閱到的數(shù)據(jù)內(nèi)容中的AppID字段，將數(shù)據(jù)對應到不用的用戶中。

5.1 SDK介紹

5.1.1 Java SDK

1. 依賴引用

在工程中添加 maven 依賴接入 SDK。

<!-- amqp 1.0 qpid client -->
<dependency>
  <groupId>org.apache.qpid</groupId>
  <artifactId>qpid-jms-client</artifactId>
  <version>0.40.0</version>
</dependency>
<!-- util for base64-->
<dependency>
  <groupId>commons-codec</groupId>
  <artifactId>commons-codec</artifactId>
  <version>1.10</version>
</dependency>

2. 身份認證

使用服務端訂閱功能，需要基于AppKey進行身份認證并建立連接。該AppKey根據(jù)應用類型不同，有兩種來源：

• 對于獨立單租戶托管并分發(fā)的應用，該AppKey是由托管平臺在應用實例化分發(fā)部署應用時，自動產(chǎn)生，并注入在主機的環(huán)境變量中，應用事先并不知道該值，需要動態(tài)從環(huán)境變量中獲取。

• 對于共享型應用，該AppKey是該SaaS應用在注冊多租戶接口時，從平臺創(chuàng)建，并硬編碼在應用中。

public static void main(String[] args) throws Exception {
        String clientId = UUID.randomUUID().toString();  //填寫唯一的id，比如uuid
        String random = UUID.randomUUID().toString().replaceAll("-", "").substring(0, 6);

        //按照阿里云IoT規(guī)范，組裝UserName
        String userName = clientId + "|authMode=appkey"
            + ",signMethod=SHA256"
            + ",random=" + random
            + ",appKey=" + "{appkey:純數(shù)字}"
            //+ ",iotInstanceId=" + iotInstanceId, //如果是企業(yè)實例ID，那么需要補充這個參數(shù)。如果是公共實例ID，那么隱藏這個參數(shù)
            + ",groupId=" + "{appkey:純數(shù)字}"
            + "|";

        //按照阿里云IoT規(guī)范，計算簽名，組裝password
        String signContent = "random=" + random;
        String password = doSign(signContent, "{app-secret}", "HmacSHA256");

        //如果是公共實例，那么使用如下鏈接地址。如果是企業(yè)實例，需要在實例管理中查看amqp鏈接地址。
        String connectionUrl =
            "failover:(amqps://{aliyun-uid}.iot-amqp.cn-shanghai.aliyuncs.com:5671?amqp.idleTimeout=80000)"
                + "?failover.maxReconnectAttempts=10&failover.reconnectDelay=30";

        Hashtable<String, String> hashtable = new Hashtable<String, String>();
        hashtable.put("connectionfactory.SBCF", connectionUrl);
        hashtable.put("queue.QUEUE", "default");
        hashtable.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.qpid.jms.jndi.JmsInitialContextFactory");
        Context context = new InitialContext(hashtable);
        ConnectionFactory cf = (ConnectionFactory)context.lookup("SBCF");
        Destination queue = (Destination)context.lookup("QUEUE");
        Connection connection = cf.createConnection(userName, password);
        MyJmsConnectionListener myJmsConnectionListener = new MyJmsConnectionListener();
        ((JmsConnection)connection).addConnectionListener(myJmsConnectionListener);
        // Create Session
        // Session.CLIENT_ACKNOWLEDGE: 收到消息后，需要手動調(diào)用message.acknowledge()
        // Session.AUTO_ACKNOWLEDGE: SDK自動ack（推薦）
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        connection.start();
        // Create Receiver Link
        MessageConsumer consumer = session.createConsumer(queue);
  			consumer.setMessageListener(messageListener);
    }

    /**
     * 按照指定簽名算法計算password簽名
     */
    private static String doSign(String toSignString, String secret, String signMethod) throws Exception {
        SecretKeySpec signingKey = new SecretKeySpec(secret.getBytes(), signMethod);
        Mac mac = Mac.getInstance(signMethod);
        mac.init(signingKey);
        byte[] rawHmac = mac.doFinal(toSignString.getBytes());
        return Hex.encodeHexString(rawHmac);
    }
}

注：connectionUrlTemplate 的寫法如下所示：

上海：
公共實例：amqps://${uid}.iot-amqp.cn-shanghai.aliyuncs.com:5671
購買的實例：在LP控制臺“實例管理”--“查看終端節(jié)點”里找到AMQP接入點

3. 設(shè)置消息接受接口

使用AMQP方式，建立連接之前，需要提供消息接收接口，用于處理回調(diào)的消息，需要通過consumer.setMessageListener(messageListener)來設(shè)置，其中messageListener是一個自定義實現(xiàn)MessageListener接口的類，如下所示：

private static MessageListener messageListener = new MessageListener() {
    @Override
    public void onMessage(final Message message) {
        try {
            //1.收到消息之后一定要ACK。
            // 推薦做法：創(chuàng)建Session選擇Session.AUTO_ACKNOWLEDGE，這里會自動ACK。
            // 其他做法：創(chuàng)建Session選擇Session.CLIENT_ACKNOWLEDGE，這里一定要調(diào)message.acknowledge()來ACK。
            // message.acknowledge();
            //2.建議異步處理收到的消息，確保onMessage函數(shù)里沒有耗時邏輯。
            // 如果業(yè)務處理耗時過程過長阻塞住線程，可能會影響SDK收到消息后的正常回調(diào)。
            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    processMessage(message);
                }
            });
        } catch (Exception e) {
            logger.error("submit task occurs exception ", e);
        }
    }
};

private static class MyJmsConnectionListener implements JmsConnectionListener() {
    /**
     * 連接成功建立
     * @param remoteURI
     */
    @Override
    public void onConnectionEstablished(URI remoteURI) {
        logger.info("onConnectionEstablished, remoteUri:{}", remoteURI);
    }

    /**
     * 最終連接失敗，嘗試過最大重試次數(shù)之后
     * @param error
     */
    @Override
    public void onConnectionFailure(Throwable error) {
        logger.error("onConnectionFailure, {}", error.getMessage());
    }

    /**
     * 連接中斷
     * @param remoteURI
     */
    @Override
    public void onConnectionInterrupted(URI remoteURI) {
        logger.info("onConnectionInterrupted, remoteUri:{}", remoteURI);
    }

    /**
     * 連接中斷后又自動重連上
     * @param remoteURI
     */
    @Override
    public void onConnectionRestored(URI remoteURI) {
        logger.info("onConnectionRestored, remoteUri:{}", remoteURI);
    }

    @Override
    public void onInboundMessage(JmsInboundMessageDispatch envelope) {
        logger.info("onInboundMessage, {}", envelope);
    }

    @Override
    public void onSessionClosed(Session session, Throwable cause) {
        logger.error("onSessionClosed, " + session, cause);
    }

    @Override
    public void onConsumerClosed(MessageConsumer consumer, Throwable cause) {
        logger.error("onConsumerClosed, " + consumer, cause);
    }

    @Override
    public void onProducerClosed(MessageProducer producer, Throwable cause) {
        logger.error("onProducerClosed, " + producer, cause);
    }
}

其中的content內(nèi)容是一個JSON對象字符串，字段信息如下：

   // 模型ID
    String modelId;

    // 變更數(shù)據(jù)的ID
    List<Long> dataIds;

    // 操作類型：insert/update/delete
    String operateType;

    // 訂閱的appId, 以appKey授權(quán)時為空
    String appId;

    String scopeId;

    // 模型的邏輯隔離id
    String logicalModelIsoId;

    // 模型實例id
    String modelInstanceId;

    //  數(shù)據(jù)實例id
    String dataInstanceId;