直连对接

BladeX大约 5 分钟directfeature

概念说明

目前BladeX物联网平台支持设备直连认证，通信协议为MQTT。若有其他协议比如TCP、Modbus等，需要通过Node-Red边缘网关进行统一解析后进行转发至MQTT认证通信。本文以Java语言编写设备模拟器来进行直连认证演示。

一、设备直连认证

创建模拟器类

@Slf4j
@RequiredArgsConstructor
public class WatchSimulator {
    private final String productKey;
    private final String deviceName;
    private final String deviceSecret;
    private final long taskDelay;

    public WatchSimulator(String productKey, String deviceName, String deviceSecret) {
        this(productKey, deviceName, deviceSecret, 3000);
    }
}

构建连接签名参数

@Slf4j
@RequiredArgsConstructor
public class WatchSimulator {
    private final String productKey;
    private final String deviceName;
    private final String deviceSecret;
    private final long taskDelay;

    public WatchSimulator(String productKey, String deviceName, String deviceSecret) {
        this(productKey, deviceName, deviceSecret, 3000);
    }
	
    /**
     * 创建客户端认证信息
     *
     * @return ClientSign
     */
    private ClientSign createClientSign() {
        ClientIdInfo clientIdInfo = new ClientIdInfo();
        clientIdInfo.setClientId(deviceName);
        clientIdInfo.setTenantId("000000");
    
        ClientSign clientSign = new ClientSign();
        clientSign.setClientIdInfo(clientIdInfo);
        clientSign.setProductKey(productKey);
        clientSign.setDeviceName(deviceName);
        clientSign.setDeviceSecret(deviceSecret);
        return clientSign;
    }
}

编写模拟器启动入口

@Slf4j
@RequiredArgsConstructor
public class WatchSimulator {
    private final String productKey;
    private final String deviceName;
    private final String deviceSecret;
    private final long taskDelay;

    public WatchSimulator(String productKey, String deviceName, String deviceSecret) {
        this(productKey, deviceName, deviceSecret, 3000);
    }
	
    /**
    * 启动模拟器
    */
    public void start() {
        // 初始化客户端认证信息
        ClientSign clientSign = createClientSign();
        String username = clientSign.getUsername();
        String password = clientSign.getPassword();
        String clientId = clientSign.getClientIdInfo().getMqttClientId();

        log.info("username: {}, password: {}, clientId: {}", username, password, clientId);

        MqttClientCreator creator = MqttClient.create()
            .username(username)
            .password(password)
            .clientId(clientId);

        // 初始化 MQTT 客户端
        MqttClient client = creator.connectSync();
        StatVo stat = client.getStat();
        log.info("stat: {}", stat);
    }
	
    /**
     * 创建客户端认证信息
     *
     * @return ClientSign
     */
    private ClientSign createClientSign() {
        ClientIdInfo clientIdInfo = new ClientIdInfo();
        clientIdInfo.setClientId(deviceName);
        clientIdInfo.setTenantId("000000");
    
        ClientSign clientSign = new ClientSign();
        clientSign.setClientIdInfo(clientIdInfo);
        clientSign.setProductKey(productKey);
        clientSign.setDeviceName(deviceName);
        clientSign.setDeviceSecret(deviceSecret);
        return clientSign;
    }
}

编写测试类查看是否能连接成功，连接基础参数可以从设备详情页获取

public class WatchTest {
    public static void main(String[] args) {
        // 产品Key
        String productKey = "dwi0gke0xo1t";
        // 设备名称
        String deviceName = "dwi2gcnb6pz5";
        // 设备密钥
        String deviceSecret = "PT0wBbgwNkDnGKw1EKkLemRQUlXqjORm";
        // 初始化模拟器
        WatchSimulator simulator = new WatchSimulator(productKey, deviceName, deviceSecret);
        // 启动模拟器
        simulator.start();
    }
}

二、设备上报数据

添加上报统一方法

@Slf4j
@RequiredArgsConstructor
public class WatchSimulator {
    private final String productKey;
    private final String deviceName;
    private final String deviceSecret;
    private final long taskDelay;

    public WatchSimulator(String productKey, String deviceName, String deviceSecret) {
        this(productKey, deviceName, deviceSecret, 3000);
    }

    /**
     * 启动模拟器
     */
    public void start() {
        // 初始化客户端认证信息
        ClientSign clientSign = createClientSign();
        String username = clientSign.getUsername();
        String password = clientSign.getPassword();
        String clientId = clientSign.getClientIdInfo().getMqttClientId();

        log.info("username: {}, password: {}, clientId: {}", username, password, clientId);

        MqttClientCreator creator = MqttClient.create()
            .username(username)
            .password(password)
            .clientId(clientId);

        // 初始化 MQTT 客户端
        MqttClient client = creator.connectSync();
        StatVo stat = client.getStat();
        log.info("stat: {}", stat);


        // 属性任务
        publishTask(client, "/blade/sys/${productKey}/${deviceName}/thing/event/property/post", createPropertyData());

        // 命令任务
        publishTask(client, "/blade/sys/${productKey}/${deviceName}/thing/service/Reboot", createRebootData());

        // 事件任务
        publishTask(client, "/blade/sys/${productKey}/${deviceName}/thing/event/LowBatteyEvent/post", createLowBatteryEventData());
    }

    /**
     * 创建客户端认证信息
     *
     * @return ClientSign
     */
    private ClientSign createClientSign() {
        ClientIdInfo clientIdInfo = new ClientIdInfo();
        clientIdInfo.setClientId(deviceName);
        clientIdInfo.setTenantId("000000");

        ClientSign clientSign = new ClientSign();
        clientSign.setClientIdInfo(clientIdInfo);
        clientSign.setProductKey(productKey);
        clientSign.setDeviceName(deviceName);
        clientSign.setDeviceSecret(deviceSecret);
        return clientSign;
    }

    /**
     * 发布任务
     *
     * @param client   MqttClient
     * @param endpoint endpoint
     * @param data     数据
     */
    private void publishTask(MqttClient client, String endpoint, Object data) {
        String topic = TemplateUtil.safeProcess(endpoint, Kv.create().set("productKey", productKey).set("deviceName", deviceName));
        client.schedule(() -> client.publish(topic, JsonUtil.toJsonBytes(data)), taskDelay);
    }

    /**
     * 创建属性数据
     *
     * @return DataReq
     */
    private DataReq<Map<String, Object>> createPropertyData() {
        DataReq<Map<String, Object>> req = new DataReq<>();
        req.setId(StringUtil.randomUUID());
        req.setVersion("1.0");
        Map<String, Object> params = new HashMap<>();
        params.put("Brightness", 50);
        params.put("BatteryLevel", 20);
        params.put("AlarmSwitch", 1);
        req.setParams(params);
        return req;
    }

    /**
     * 创建重启数据
     *
     * @return DataReq
     */
    private DataReq<Map<String, Object>> createRebootData() {
        DataReq<Map<String, Object>> req = new DataReq<>();
        req.setId(StringUtil.randomUUID());
        req.setVersion("1.0");
        Map<String, Object> params = new HashMap<>();
        Map<String, Object> input = new HashMap<>();
        input.put("RebootStatus", "true");
        Map<String, Object> output = new HashMap<>();
        output.put("RebootTime", System.currentTimeMillis());
        params.put("input", JsonUtil.toJsonString(input));
        params.put("output", JsonUtil.toJsonString(output));
        params.put("commandName", "重启服务");
        req.setParams(params);
        return req;
    }

    /**
     * 创建低电量事件数据
     *
     * @return DataReq
     */
    private DataReq<Map<String, Object>> createLowBatteryEventData() {
        DataReq<Map<String, Object>> req = new DataReq<>();
        req.setId(StringUtil.randomUUID());
        req.setVersion("1.0");
        Map<String, Object> params = new HashMap<>();
        Map<String, Object> output = new HashMap<>();
        output.put("BatteryLevel", 20);
        params.put("output", JsonUtil.toJsonString(output));
        params.put("eventName", "电量低事件");
        params.put("eventType", "alert");
        req.setParams(params);
        return req;
    }
}

代码解析说明

publishTask：发布任务方法，用于发布属性、命令、事件数据
createPropertyData：创建属性数据
createRebootData：创建重启数据
createLowBatteryEventData：创建低电量事件数据
start：启动模拟器，初始化客户端认证信息，连接MQTT客户端，发布属性、命令、事件数据
createClientSign：创建客户端认证信息
WatchTest：测试类，启动模拟器
WatchSimulator：模拟器类，用于模拟设备上报数据

上报数据格式说明

属性上报topic: /blade/sys/${productKey}/${deviceName}/thing/event/property/post
属性数据格式

{
  "id": "1",
  "version": "1.0",
  "params": {
    "Brightness": 50,
    "BatteryLevel": 20,
    "AlarmSwitch": 1
  }
}

命令上报topic: /blade/sys/${productKey}/${deviceName}/thing/service/Reboot
注意：其中Reboot为标识符，代表对应命令所指定的物模型标识符
命令数据格式

{
  "id": "1",
  "version": "1.0",
  "params": {
    "input": {"RebootStatus":"true"},
    "output": {"RebootTime":1627584000000},
    "commandName": "重启服务"
  }
}

事件上报topic: /blade/sys/${productKey}/${deviceName}/thing/event/LowBatteyEvent/post
注意：其中LowBatteyEvent为标识符，代表对应事件所指定的物模型标识符
事件数据格式

{
  "id": "1",
  "version": "1.0",
  "params": {
    "output": {"BatteryLevel":20},
    "eventName": "电量低事件",
    "eventType": "alert"
  }
}

完整数据格式请见如下工具类

/**  
 * 设备通用数据格式  
 */  
@Data  
@JsonInclude(JsonInclude.Include.NON_NULL)  
public class DataReq<T> implements Serializable {  
  
    /**  
     * 消息ID号。uuid，去掉短横线，32位，全局唯一，用于ack或系统消息追踪  
     */  
    private String id;  
  
    /**  
     * 协议版本号，目前协议版本号唯一取值为1.0  
     */  
    private String version;  
  
    /**  
     * 扩展功能的参数，其下包含各功能字段。平台可扩展，或可自行扩展，  
     * 自行扩展的参数需在自定义解析模块自行解析  
     */  
    private SysBean sys;  
  
    /**  
     * 请求方法。  
     */  
    private String method;  
  
    /**  
     * 请求参数。  
     */  
    private T params;  
  
}

/**  
 * 扩展功能  
 */  
@Data  
@NoArgsConstructor  
@AllArgsConstructor  
public class SysBean implements Serializable {  
  
    /**  
     * 不响应 ack  
     */  
    public static final int ACK_NO = 0;  
  
    /**  
     * 响应 ack  
     */  
    public static final int ACK_NEED = 1;  
  
    /**  
     * 构造方法，根据ackNeed的值设置ack字段  
     *  
     * @param ackNeed 是否需要响应ack  
     */  
    public SysBean(boolean ackNeed) {  
        this(ackNeed ? ACK_NEED : ACK_NO);  
    }  
  
    /**  
     * 扩展功能字段，表示是否返回响应数据。  
     * 0：不返回响应数据  
     * 1：返回响应数据  
     */  
    private int ack;  
  
}

三、测试设备上报数据

启动测试类，查看平台数据结果，可以看到设备处于在线状态，对应的属性、命令、事件数据也在定时上报