OneNET物联网MQTT协议之上传数据点（T1）

MQTT协议为了考虑低功耗的问题，在上传数据时，还是有些说法的。本节我们就先以OneNET平台为例，演示如何使用MQTT协议进行数据点的上传。在进行讲解之前，我们先明确两个问题。

Q：网络传输数据格式大端还是小端？

A：在网络传输中，统一使用大端模式传输。

Q：MQTT协议的数据包格式是什么？

A：MQTT协议中，一个MQTT数据包由：固定头(Fixed header)、可变头(Variable header)、消息体(Payload)三部分构成。
我们在使用第三方的MQTT协议库时，可以只关心Payload的内容。

MQTT协议上传数据点

我们先去OneNET的文档中心查阅一些资料。

首先，平台文档在介绍MQTT时，指明了MQTT协议支持的数据格式

整型（int）

浮点数（float）

字符串（string）

JSON格式

二进制数据

然后，我们从官网下载《设备终端接入协议-MQTT》文档

打开文档，查看上传数据点相关内容，如下图：

从上图我们看出，上传数据点的topic主题是"$dp"

我们再来看看Payload

从上表中，我们可以看到payload内容的第一个字节表示上传的数据类型。其表示的意义为：

Type	内容
1	JSON格式1字符串
2	二进制数据
3	JSON格式2字符串
4	JSON格式3字符串
5	自定义分隔符
6	带时间自定义分隔符
7	可离散浮点数数据流

上面类型好多，我们接下来以JSON格式1字符串来讲解。

JSON格式1字符串

当Type=1时，文档里是这样描述的

表格描述的比较清晰了，Byte2和Byte3表示后面的json格式数据转字符串后的长度，并且Byte2和Bytes3是表示一个无符号16位数据，同时按照大端模式传输。

那这样我们就清楚了。比如，我用python代码实现payload最终的转换，如下部分代码：

# 转换Json格式1为字符串数组
def convert_json_format1(_body):
    packet_type = 0x01
    packet = bytearray()
    packet.extend(struct.pack("!B", packet_type))   # 首先填充数据点类型值
    packet_content = json.dumps(_body)      # 将Json格式数据转换为字符串
    if isinstance(packet_content, str):
        packet_content = packet_content.encode('utf-8')
        packet_content_length = len(packet_content)
        # 填充json字符串长度和json字符串
        packet.extend(struct.pack("!H" + str(packet_content_length) + "s", packet_content_length, packet_content))
    return packet


# 按照16进制打印字符串
def print_hex(_bytes):
    _val = [hex(int(i)) for i in _bytes]
    print(' '.join(_val))

使用此代码，我做如下实验：


if __name__ == "__main__":
    json_body = {
        'datastreams': [
            {
                'id': 'temperature',
                'datapoints': [
                    {
                        'value': 1234
                    }
                ]
            }
        ]
    }
    packet = convert_json_format1(json_body)
    print_hex(packet)

输出内容为：

可以看到第一个箭头所指的字节表示数据点类型

第2和第3个字节内容为json格式的数据长度。并且我使用struct.pack时加入了一个感叹号表示转换为网络传输模式。

接下来，我们就使用这段代码进行测试上传一个数据点到OneNET平台。

上传一个数据点到OneNET平

我们在设备列表的数据流模板里添加一个数据流模板，如下：

然后我们编写python代码，使用MQTT连接，在连接成功之后按照json格式1上传一个数据点。如下为连接成功之后，发送数据点的代码：

import paho.mqtt.client as mqtt
import json
import struct
import random

BrokerHost = '183.230.40.39'    # OneNET使用TCP方式连接时的主机地址
BrokerPort = 6002               # OneNET使用TCP方式连接时的主机端口号

DeviceId = '610170361'          # 设备ID
ProductId = '360633'            # 产品ID
APIKey = "J0UL5YvdDB3=20qE9ByeLGYctcI="    # 可以是设备单独的APIKey，也可以是产品的MasterAPIKey


# 按照16进制打印字符串
def print_hex(_bytes):
    _val = [hex(int(i)) for i in _bytes]
    print(' '.join(_val))


# 转换Json格式1为字符串数组
def convert_json_format1(_body):
    packet_type = 0x01
    packet = bytearray()
    packet.extend(struct.pack("!B", packet_type))   # 首先填充数据点类型值
    packet_content = json.dumps(_body)      # 将Json格式数据转换为字符串
    if isinstance(packet_content, str):
        packet_content = packet_content.encode('utf-8')
        packet_content_length = len(packet_content)
        # 填充json字符串长度和json字符串
        packet.extend(struct.pack("!H" + str(packet_content_length) + "s", packet_content_length, packet_content))
    # print_hex(packet)
    return packet


# 连接结果
def on_connect(client, userdata, flags, rc):
    if rc != 0:
        print("连接失败:" + mqtt.connack_string(rc))
        return
    print("连接成功")
    # 上传测试数据
    value = random.randint(100,10000)

    test_data = {
        'datastreams': [
            {
                'id': 'mqtt20200713',
                'datapoints': [
                    {
                        'value': value
                    }
                ]
            }
        ]
    }
    payload = convert_json_format1(test_data)
    client.publish("$dp", payload, qos=0)


# 从服务器接收发布消息时的回调。
def on_message(client, userdata, msg):
    print("***** 接收到消息 *****")
    print(msg.topic + ":" + msg.payload.decode("utf-8"))


# 当broker响应订阅请求时被调用
def on_subscribe(client, userdata, mid, granted_qos):
    print("***** Broker响应订阅请求*****")
    print(granted_qos)


# 消息发送回调
def on_publish(client, userdata, mid):
    print("[on_publish] mid:" + str(mid))


def main():
    mqtt_broker_host = BrokerHost
    mqtt_broker_port = BrokerPort
    mqtt_user_name = ProductId
    mqtt_password = APIKey
    mqtt_client_id = DeviceId
    client = mqtt.Client(client_id=mqtt_client_id, protocol=mqtt.MQTTv311)
    client.on_connect = on_connect
    client.on_publish = on_publish
    client.on_message = on_message
    client.on_subscribe = on_subscribe
    client.username_pw_set(username=mqtt_user_name, password=mqtt_password)
    client.connect(host=mqtt_broker_host, port=mqtt_broker_port, keepalive=60)
    client.loop_forever()


if __name__ == "__main__":
    main()

运行可以看到，在OneNet后台已经上传数据点成功。

点开

我这里发布消息时，使用的QOS为0，你当然也可以用QOS1，即当你上传数据点成功时平台会给你一个ACK。

我上面只是演示了json格式1的数据上传演示，其它的格式你可以自己去研究一下。毕竟在对低功耗要求高的设备里，使用这种格式上传还是耗费流量和耗电的。

OneNET物联网MQTT协议之上传数据点（T1）

MQTT协议上传数据点

JSON格式1字符串

上传一个数据点到OneNET平

学习来源