MQTT规则配置说明

概述

1.Jointech提供的MQTT推送服务，是一种将设备数据实时发送给数据使用方的服务。 当设备上传数据后，推送服务会第一时间感知，在将数据整理打包后，发送至MQTT服务器，客户数据接收服务从MQTT服务器获取数据，然后客户根据自身业务自行处理数据。
2.MQTT规则配置界面，主要是方便用户配置哪些设备进行MQTT推送服务
3.订阅主题说明

MQTT规则配置

用户可以在此界面中，配置需要进行MQTT推送的终端，通过选择数据源，可以进行多方式的选择所需要推送的终端。

数据源说明

1.公司全部设备

表示当前公司及其子公司下的所有终端

2.公司直属设备

表示当前公司及下的所有终端

3.公司指定设备

表示当前公司及其子公司下的指定终端

订阅主题及负载说明

1.服务器地址及说明

2.客户端账户及密码

为当前MQTT规则配置的登录账户及密码（MD5加密后），也可直接在该规则的详情界面中查看。

3.订阅说明

将{accessKey}替换为系统分配的实际值，{deviceNum}替换为设备的ID，{subLockNum}替换为设备子锁的ID

3.1 终端状态上报

主题：upload/{accessKey}/{deviceNum}/location
描述：终端的定位状态数据
示例：

{"assetId":"8052400203","battery":255,"cellSignal":31,"cells":"460,0,10352,188975300","direction":0,"expandInfo":"{\"angle\":\"null\",\"backBattery\":\"null\",\"fuels\":\"-1,-1,-1\",\"humidity\":\"0\",\"lux\":\"null\",\"networkType\":\"0\",\"pressure\":\"null\",\"reportType\":\"null\",\"temperature\":\"-1000.0\"}","gnssSignal":0,"gpsTime":"2025-02-07T07:26:30Z","latitude":22.6700711039462,"locType":2,"longitude":113.922999834147,"mileage":0,"recvTime":"2025-02-07T07:25:20.832Z","speed":0,"statusJson":"{\"lockRope\":0,\"lockStatus\":0}","voltage":"0.0"}

示例Json字符串说明：

锁状态JSON说明

车载状态JSON说明

拓展信息JSON说明

3.2 终端报警上报

主题：upload/{accessKey}/{deviceNum}/alarm
描述：终端的报警数据，如低电量，撞击报警
示例：

{"alarmType":44,"assetId":"8454601010","cells":"460,0,9383,149428936","dateTime":"2020-04-01T00:00:25Z","describe":"","fileIndex":"1585728025","latitude":22.6766686619127,"longitude":113.928864975252,"mileage":59,"speed":0}

示例Json字符串说明：

报警类型说明：

3.3 终端事件上报

主题：upload/{accessKey}/{deviceNum}/event
描述：终端的事件数据，如开锁，关锁
示例：

{"assetId":"8454601010","card":"","cells":"460,0,9383,149428936","dateTime":"2025-02-07T06:54:52Z","describe":"","eventType":1,"latitude":22.6766686619127,"longitude":113.928864975252,"mileage":59,"password":"","speed":0,"unLockType":2}

示例Json字符串说明：

3.4 指令应答上报

主题：upload/{accessKey}/{deviceNum}/cmd
描述：终端响应服务器下发的指令
指令应答说明：不同的指令应答结果不一样，具体指令及其说明，请参考协议文档
示例：

 //正常回复
{"assetId":"794308010642","commandType":"BASE1","content":"[\"794308010642\",\"8\",\"001\",\"BASE\",\"1\",\"JT709A_20241115_HW-V1.2_RFID_7600_V2_8\",\"0\",\"Jointech\",\"LE20B05SIM7600M21-A_CUS_JT\",\"898604B81022C1046673\",\"868822046635957\",\"460\",\"0\",\"188975300\",\"10352\"]","dateTime":"2025-02-07T03:52:39.964Z"}
//失败回复 （Device not online）
{"assetId":"8294630003","commandType":"MQTT_CMD","content":"Device not online！","dateTime":"2025-01-07T09:49:43.158Z"}
//失败回复 （Device has not been registered）
{"assetId":"8294630481","commandType":"MQTT_CMD","content":"Device has not been registered！","dateTime":"2025-01-07T09:48:32.975Z"}

示例Json字符串说明：
可参考数据引擎服务说明中：数据类型说明-> 指令应答数据

3.5 从机状态上报

主题：upload/{accessKey}/{deviceNum}/{subLockNum}/sublock
描述：从机的定位状态数据
示例：

{"assetId":"8052400203","battery":41,"direction":0,"eventType":3,"gpsTime":"2025-02-07T07:16:07Z","humidity":0,"latitude":-0.0,"locRope":0,"locStatus":0,"locTimes":54,"longitude":-0.0,"recvTime":"2025-02-07T07:14:57.315Z","rssi":15,"sensorType":4,"speed":0,"statusJson":"{\"lockRope\":0,\"gateway\":0}","subAssetID":"E0171E0366","subGpsTime":"2025-02-07T07:15:14Z","subGpsTimestamp":1738912514000,"temperature":-1000.0,"voltage":"3.75"}

示例Json字符串说明：

statusJson字段说明

“FStatusJson”:”{"bottomDisassembly":0,"emergencyKey":0,"lockMotor":0,"structuralDisassembly":0,"gateway":0,"lockValve":0}”

3.6 服务器指令下发

主题：download/{accessKey}/{deviceNum}/cmd
描述：下发指令到终端的数据
指令说明：不同设备类型有不同的指令功能，具体指令及其说明，请参考协议文档
示例：

(794308010642,1,001,BASE,1)

代码示例

引入paho依赖

<dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.2.0</version>
</dependency>

建立MQTT连接
建立连接的时候需使用AccessId、secret、topic、clientId均为系统分配，其中AccessId对应客户端的user，secret对应客户端的password。

String user         = "u8vngl";//user
String password     = "xxxxxx";//Password(MD5-32)
String topic        = "upload/{accessKey}/{deviceNum}/location";
//String topic        = "upload/B176C0FF1DFF41D0126BFF7908C76E17/8041254836/location"; //订阅此设备的位置数据
//String topic        = "upload/B176C0FF1DFF41D0126BFF7908C76E17/+/location"; //订阅此配置设备的位置数据
//String topic        = "upload/B176C0FF1DFF41D0126BFF7908C76E17/#"; //订阅此配置设备的位置、报警、事件等数据
int qos             = 1;//qos2消耗较大，请使用1或0
String broker       = "mqtt://mqtt.assetscontrols.com:1883";//mqtt服务器地址
String clientId     = "u8vngl_consumer";
//qos为1或2时，mqttclient使用
MemoryPersistence persistence = new MemoryPersistence();

try {
    MqttClient client = new MqttClient(broker, clientId, persistence);
    MqttConnectOptions connOpts = new MqttConnectOptions();
    connOpts.setUserName(user);
    connOpts.setPassword(password.toCharArray());
    //cleanSession为false时，下次以相同clientId登录将可以获取存储的所有消息
    //如果为true，将获取到retained标记的最后一条消息
    //cleanSession调试测试阶段设置为true，生产设置为false
    connOpts.setCleanSession(false);
    connOpts.setAutomaticReconnect(true);//设置自动重连

    client.setCallback(new MqttCallback());//获取订阅消息
    client.connect(connOpts);
    client.subscribe(topic, qos);//订阅topic
} catch(MqttException me) {
    me.printStackTrace();
}

获取订阅消息
paho的消息获取通过实现MqttCallback接口，MqttCallback接口通过messageArrived方法获取消息

@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
    //messge的处理请放入其他线程中，在该方法中消耗时间过多将影响qos为1或2的响应，使服务器认为未成功投递消息
    System.out.println("topic:" + topic + " msg:" + message);
}


/**
 * [重要提示]
 * 连接丢失后的处理逻辑
 * 1、连接重连
 * 2、重新订阅Topic数据
 * @param cause
 */
@Override
public void connectionLost(Throwable cause) {
    cause.printStackTrace();
    while (true) {
        try {
            //重连
            if (!client.isConnected()) {
                client.reconnect();
            }
            //重新订阅Topic数据
            if (client.isConnected()) {
                client.subscribe(topic, qos);
                System.out.println("has resubscribed");
                break;
            }
            TimeUnit.SECONDS.sleep(100);
        } catch (MqttException | InterruptedException e1) {
            e1.printStackTrace();
        }
        System.out.println("client not connect");
    }
}