1、简介
1.1 开发环境与连接平台
本文主要介绍W5500如何通过MQTT协议将设备连接到阿里云IoT,并通过MQTT协议实现通信。MQTT协议是基于TCP的协议,所以我们只需要在单片机端实现TCP客户端代码之后就很容易移植MQTT了, +W5500实现TCP客户端的代码我们以前已经实现过,程序下载:
软件环境:Windows
- 硬件环境:STM32F103+W5500
- 开发工具:Keil uVision5
- 调试工具:Wireshark、串口调试助手
- 连接平台:阿里云-华东2节点(https://www.aliyun.com)
-
1.2 MQTT简介:
MQTT官网地址:(http://mqtt.org/)
-
1.2.1 MQTT协议特点
MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的,这些特点使它适用范围非常广泛。在很多情况下,包括受限的环境中,如:机器与机器(M2M)通信和物联网(IoT)。其在,通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。
MQTT协议当前版本为,2014年发布的MQTT v3.1.1。除标准版外,还有一个简化版MQTT-SN,该协议主要针对嵌入式设备,这些设备一般工作于百TCP/IP网络,如:ZigBee。
MQTT协议运行在TCP/IP或其他网络协议,提供有序、无损、双向连接。其特点包括:
- 使用的发布/订阅消息模式,它提供了一对多消息分发,以实现与应用程序的解耦。
- 对负载内容屏蔽的消息传输机制。
- 对传输消息有三种服务质量(QoS):
- 最多一次,这一级别会发生消息丢失或重复,消息发布依赖于底层TCP/IP网络。即:<=1
- 至多一次,这一级别会确保消息到达,但消息可能会重复。即:>=1
- 只有一次,确保消息只有一次到达。即:=1。在一些要求比较严格的计费系统中,可以使用此级别
数据传输和协议交换的最小化(协议头部只有2字节),以减少网络流量
通知机制,异常中断时通知传输双方
- MQTT协议原理及实现方式
实现MQTT协议需要:客户端和服务器端
MQTT协议中有三种身份:发布者(Publish)、代理(Broker)(服务器)、订阅者(Subscribe)。其中,消息的发布者和订阅者都是客户端,消息代理是服务器,消息发布者可以同时是订阅者。
MQTT传输的消息分为:主题(Topic)和消息的内容(payload)两部分
Topic,可以理解为消息的类型,订阅者订阅(Subscribe)后,就会收到该主题的消息内容(payload)
payload,可以理解为消息的内容,是指订阅者具体要使用的内容
- 连接
- 阿里云连接步骤:
- 以aliyun账号直接进入IoT控制台,如果还没有开通阿里云物联网套件服务,则 申请开通
- 接入引导
(1)、创建产品
(2)、添加设备
(3)、获取设备的Topic
- 创建产品
初步进入控制台后,需要创建产品。点击创建产品。产品相当于某一类设备的集合,用户可以根据产品管理其设备等。
- 产品名称:对产品命名,例如可以填写产品型号。产品名称在账号内保持唯一。
- productKey:阿里云IoT为产品颁发的全局唯一标识符
添加设备
创建完产品之后,可以为该产品添加设备。进入产品管理页面下的设备管理,点击添加设备。
- 说明:用户可以自定义设备名称(即deviceName),这个名称即可作为设备唯一标识符,用户可以基于该设备名称与IoT Hub进行通信,需要指出的是,用户需要保证deviceName产品内唯一。
- 设备证书:添加设备之后,物联网套件为设备颁发的唯一标识符,设备证书用于设备认证以及设备通信,详细的请参考设备接入文档。
- deviceName:用户自定义设备唯一标识符,用于设备认证以及设备通信,用户保证产品维度内唯一。
- deviceSecret:物联网套件为设备颁发的设备秘钥,用于认证加密,与deviceName或者deviceId成对出现。
- 获取设备的Topic
添加设备之后,可以获取设备的Topic。点击Topic列表
- 说明:创建产品之后,物联网套件都会为产品默认定义三个Topic类。那么,在添加设备之后,每个设备都会默认有三个Topic,即图中所示。如果想要增加、修改、删除Topic,请到消息通信重新定义Topic类。
- 设备可以基于Topic列表中的Topic进行Pub/Sub通信,例如列表中有/1000118502/test9/update,且设备拥有的权限是发布,这就意味着设备可以往这个Topic发布消息;同样,列表中/1000118502/test9/get,权限是订阅,这就意味着设备可以从这个Topic订阅消息。
- 设备接入
获得productKey、设备证书以及设备的Topic这些参数,就可以基于aliyun IoT device SDK for C将设备连接上IoT Hub并进行通信,具体请参考《MQTT配置》部分
- MQTT移植步骤:
MQTT代码源码下载地址:(http://www.eclipse.org/paho/)
MQTT的移植非常简单,将C/C++ MQTT Embedded clients的代码添加到工程中,然后我们只需要再次封装4个函数即可:
int transport_sendPacketBuffer(unsigned char* buf, int buflen);
通过网络以TCP的方式发送数据;
int transport_getdata(unsigned char* buf, int count);
TCP方式从服务器端读取数据,该函数目前属于阻塞函数;
int transport_open(void);
打开一个网络接口,其实就是和服务器建立一个TCP连接;
int transport_close(void);
关闭网络接口。
如果已经移植好了socket方式的TCP客户端的程序,那么这几个函数的封装也是非常简单的,程序代码如下所示:
/** * @brief 通过TCP方式发送数据到TCP服务器 * @param buf数据首地址 * @param buflen数据长度 * @retval 小于0表示发送失败 */ /*订阅消息*/ int Subscribe_sendPacketBuffer(unsigned char* buf, int buflen) { return send(SOCK_TCPS,buf,buflen); } /*发布消息*/ int Published_sendPacketBuffer(unsigned char* buf, int buflen) { return send(SOCK_TCPC,buf,buflen); } /** * @brief 阻塞方式接收TCP服务器发送的数据 * @param buf数据存储首地址· * @param count数据缓冲区长度 * @retval 小于0表示接收数据失败 */ int Subscribe_getdata(unsigned char* buf, int count) { return recv(SOCK_TCPS,buf,count); } int Published_getdata(unsigned char* buf, int count) { return recv(SOCK_TCPC,buf,count); } /** * @brief 打开一个socket并连接到服务器 * @param 无 * @retval 小于0表示打开失败 */ int Subscribe_open(void) { int32_t ret; //新建一个socket并绑定本地端口5000 ret = socket(SOCK_TCPS,Sn_MR_TCP,50000,0x00); if (ret != 1) { printf("%d:Socket Error\r\n",SOCK_TCPS); while (1); } else { printf("%d:Opened\r\n",SOCK_TCPS); } while (getSn_SR(SOCK_TCPS)!=SOCK_ESTABLISHED) { printf("connecting\r\n"); //连接TCP服务器÷ ret = connect(SOCK_TCPS,server_ip,1883); //端口必须为1883 } if (ret != 1) { printf("%d:Socket Connect Error\r\n",SOCK_TCPS); while (1); } else { printf("%d:Connected\r\n",SOCK_TCPS); } return 0; } int Published_open(void) { int32_t ret; ret = socket(SOCK_TCPC,Sn_MR_TCP,5001,0x00); if (ret != 1) { printf("%d:Socket1 Error1\r\n",SOCK_TCPC); while (1); } else { printf("%d:socket1 Opened\r\n",SOCK_TCPC); } while (getSn_SR(SOCK_TCPC)!=SOCK_ESTABLISHED) { ret = connect(SOCK_TCPC,server_ip,1883); //端口必须为1883 } if (ret != 1) { printf("%d:Socket Connect1 Error\r\n",SOCK_TCPC); while (1); } else { printf("%d:Connected1\r\n",SOCK_TCPC); } return 0; } } /** * @brief 关闭socket * @param 无 * @retval 小于0表示关闭失败 */ int Subscribe_close(void) { disconnect(SOCK_TCPS); printf("close0\n\r"); while (getSn_SR(SOCK_TCPC)!=SOCK_CLOSED) { ; } return 0; } int Published_close(void) { disconnect(SOCK_TCPC); printf("close1\n\r"); while (getSn_SR(SOCK_TCPC)!=SOCK_CLOSED) { ; } return 0; }
- MQTT配置
- MQTT连接参数说明
举例:
- MQTT与阿里云连接函数:
参考阿里云内MQTT设备接入手册,计算出设备连接的各项参数,例如下列程序中框中的部分为本例程MQTT与阿里云连接的参数的配置,详细内容如下:
clientId = 192.168.207.115 deviceName = MQTT1 productKey = TKKMt4nMF8U timestamp = 789(毫秒值) signmethod = hmacsha1(算法类型) deviceSecret = secret
那么使用tcp方式提交给mqtt参数分别如下:
-
mqttClientId:clientId+"|securemode=3,signmethod=hmacsha1,timestamp=789|"
-
clientId=192.168.207.115|securemode=3,signmethod=hmacsha1,timestamp=789|
- keepalive时间需要设置超过60秒以上,否则会拒绝连接。
- Cleansession为1;
-
mqttUsername: deviceName+"&"+productKey
-
username = "MQTT1&TKKMt4nMF8U"
-
password=hmacsha1("secret","clientId168.207.115deviceNameMQTT1productKeyTKKMt4nMF8Utimestamp789").toHexString();
最后是二进制转16制字符串大小写不敏感。这个例子结果为 9076b0ebc04dba8a8ebba1f0003552dbc862c9b9
MQTT连接函数原型,tcp_client.c文件中的MQTT_CON_ALI函数中调用make_con_msg函数并通过阿里云设备的参数,设置MQTT连接阿里云函数的参数:
- void make_con_msg(char* clientID,int keepalive, uint8 cleansession,
- char*username,char* password,unsigned char*buf,int
- buflen)
- {
- int32_t len,rc;
- MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
- data.clientID.cstring = clientID;
- data.keepAliveInterval = keepalive;
- data.cleansession = cleansession;
- data.username.cstring = username;
- data.password.cstring = password;
- len = MQTTSerialize_connect(buf, buflen, &data);
- //构造链接报文
- return;
- } MQTT连接过程:
void MQTT_CON_ALI(void) { int len; int type; switch (getSn_SR(0)) { //获取socket0的状态 case SOCK_INIT: //Socket处于初始化完成(打开)状态 connect(0, server_ip,server_port); //配置Sn_CR为CONNECT,并向TCP服务器发出连接请求¢ break; case SOCK_ESTABLISHED: // Socket处于连接建立状态 if (getSn_IR(0) & Sn_IR_CON) { setSn_IR(0, Sn_IR_CON); // Sn_IR的CON位置1,通知W5500连接已建立 } memset(msgbuf,0,sizeof(msgbuf)); if ((len=getSn_RX_RSR(0))==0) { if (1==CONNECT_FLAG) { printf("send connect\r\n"); /*MQTTÆ拼接连接报文 *根据阿里云平台MQTT设备接入手册配置 */ //void make_con_msg(char* clientID,int keepalive, uint8 cleansession,char*username, char* password,unsigned char*buf, int buflen) make_con_msg("192.168.207.115|securemode=3, signmethod=hmacsha1,timestamp=789|",180, 1,"MQTT1&TKKMt4nMF8U", "9076b0ebc04dba8a8ebba1f0003552dbc862c9b9" ,msgbuf,sizeof(msgbuf)); //printf(" server_ip: %d.%d.%d.%d\r\n", server_ip[0], server_ip[1],server_ip[2],server_ip[3]); //printf("connect ALY\r\n"); CONNECT_FLAG = 0; send(0,msgbuf,sizeof(msgbuf)); Delay_s(2); while ((len=getSn_RX_RSR(0))==0) { Delay_s(2); send(0,msgbuf,sizeof(msgbuf)); }; recv(0,msgbuf,len); while (mqtt_decode_msg(msgbuf)!=CONNACK) { //判断是不是CONNACK printf("wait ack\r\n"); } } else if (SUB_FLAG == 1) { memset(msgbuf,0,sizeof(msgbuf)); make_sub_msg(topic,msgbuf,sizeof(msgbuf)); // make_pub_msg(topic,msgbuf,sizeof(msgbuf),"hello"); send(0,msgbuf,sizeof(msgbuf)); // 接收到数据后再回给服务器,完成数据回环 SUB_FLAG = 0; Delay_s(2); while ((len=getSn_RX_RSR(0))==0) { Delay_s(2); send(0,msgbuf,sizeof(msgbuf)); }; recv(0,msgbuf,len); while (mqtt_decode_msg(msgbuf)!=SUBACK) { //判断是不是SUBACK printf("wait suback\r\n"); } TIM_Cmd(TIM2, ENABLE); printf("send sub\r\n"); } #if 1 else { //count++; // Delay_s(2); if (count>10000) { count = 0; make_ping_msg(msgbuf,sizeof(msgbuf)); send(0,msgbuf,sizeof(msgbuf)); while ((len=getSn_RX_RSR(0))==0) { //Delay_s(2); //send(0,msgbuf,sizeof(msgbuf)); printf("wait pingresponse"); }; recv(0,msgbuf,len); printf("ping len : %d\r\n",len); if (len>2) { if (PUBLISH==mqtt_decode_msg(msgbuf+2)) { printf("publish\r\n"); MQTTDeserialize_publish(&dup, &qos, &retained, &mssageid, &receivedTopic, &payload_in, &payloadlen_in, msgbuf+2, len-2); // printf("message arrived %d: %s\n\r", payloadlen_in, payload_in); memset(topic,0,sizeof(topic)); memset(ser_cmd,0,sizeof(ser_cmd)); memcpy(topic,receivedTopic.lenstring.data, receivedTopic.lenstring.len); replace_string(new_topic,topic , "request", "response"); printf("topic:%s\r\n",topic); strcpy(ser_cmd,(const char *)payload_in); //parse_topic(ser_cmd); // printf("message is %s\r\n",ser_cmd); memset(msgbuf,0,sizeof(msgbuf)); make_pub_msg(new_topic,msgbuf,sizeof( msgbuf),"hello"); send(0,msgbuf,sizeof(msgbuf)); } } } } #endif #if 0 if (PUB_FLAG==1) { memset(msgbuf,0,sizeof(msgbuf)); // make_sub_msg(topic,msgbuf,sizeof(msgbuf)); make_pub_msg(topic,msgbuf,sizeof(msgbuf),"hello"); if (count == 10000) { PUB: send(0,msgbuf,sizeof(msgbuf)); // 接收到数据后再回给服务器,完成数据回环 Delay_s(2); // while((len=getSn_RX_RSR(0))==0) // { // Delay_s(2); //send(0,msgbuf,sizeof(msgbuf)); // printf("puback\r\n"); // }; // recv(0,msgbuf,len); // if(mqtt_decode_msg(msgbuf)!=PUBACK) // { // goto PUB; // printf("wait Puback\r\n"); // } printf("send Pub\r\n"); } } #endif } #if 1 if ((len=getSn_RX_RSR(0))>0) { recv(0,msgbuf,len); if (PUBLISH== mqtt_decode_msg(msgbuf)) { printf("publish\r\n"); MQTTDeserialize_publish(&dup, &qos, &retained, &mssageid, &receivedTopic, &payload_in, &payloadlen_in, msgbuf, len); // printf("message arrived %d: %s\n\r", payloadlen_in, payload_in); memset(topic,0,sizeof(topic)); memcpy(topic,receivedTopic.lenstring.data, receivedTopic.lenstring.len); replace_string(new_topic,topic , "request","response"); printf("topic:%s\r\n",topic); memset(ser_cmd,0,sizeof(ser_cmd)); memcpy(ser_cmd,(const char *)payload_in,strlen((char*) payload_in)); memset(msgbuf,0,sizeof(msgbuf)); make_pub_msg(new_topic,msgbuf,sizeof(msgbuf),rebuf); send(0,msgbuf,sizeof(msgbuf)); //printf("%s\n",msgbuf); } else if (PINGRESP== mqtt_decode_msg(msgbuf)) { if (len>2) { if (PUBLISH==mqtt_decode_msg(msgbuf+2)) { printf("publish\r\n"); MQTTDeserialize_publish(&dup, &qos, &retained, &mssageid, &receivedTopic, &payload_in, &payloadlen_in, msgbuf+ 2, len-2); // printf("message arrived %d: %s\n\r", payloadlen_in, payload_in); memset(topic,0,sizeof(topic)); memcpy(topic,receivedTopic.lenstring.data, receivedTopic.lenstring.len); replace_string(new_topic,topic,"request", "response"); printf("topic:%s\r\n",topic); memset(ser_cmd,0,sizeof(ser_cmd)); strcpy(ser_cmd,(const char *)payload_in); // printf("message is %s\r\n",ser_cmd); //parse_topic(ser_cmd); memset(msgbuf,0,sizeof(msgbuf)); make_pub_msg(new_topic,msgbuf,sizeof(msgbuf), "hello"); send(0,msgbuf,sizeof(msgbuf)); } } } else { printf("wait publish\r\n"); } } // printf("send ping\r\n"); #endif break; case SOCK_CLOSE_WAIT: //Socket处于等待关闭状态 close(0); // 关闭Socket0 break; case SOCK_CLOSED: // Socket处于关闭状态 socket(0,Sn_MR_TCP,local_port,Sn_MR_ND); // 打开Socket0,并配置为TCP无延时模式,打开一个本地端口 break; } }
- Password有两种获得方法:
- 通过网页“在线加密解密”HamcSHA1获得;(http://encode.chahuo.com/)通过hmacsha1算法解析获得解析步骤如下:
void hmac_sha1(uint8_t *key, uint16_t key_length, uint8_t *data, uint16_t data_length, uint8_t *digest) { uint8_t b = 64; /* blocksize */ uint8_t ipad = 0x36; uint8_t opad = 0x5c; uint8_t k0[64]; uint8_t k0xorIpad[64]; uint8_t step7data[64]; uint8_t step5data[MAX_MESSAGE_LENGTH+128]; uint8_t step8data[64+20]; uint16_t i; for (i=0; i<64; i++) { k0[i] = 0x00; } /* Step 1 */ if (key_length != b) { //判断秘钥K字节长度是否等于B /* Step 2 */ if (key_length > b) { //如果大于B,则另K0=H(K) sha1(key, key_length, digest); for (i=0; i<20; i++) { k0[i]=digest[i]; } } /* Step 3 */ else if (key_length < b) { //如果小于B,则在末尾添加B-length(K) 位的0 for (i=0; i<key_length; i++) { k0[i] = key[i]; } } } else { for (i=0; i<b; i++) { k0[i] = key[i]; } } #ifdef HMAC_DEBUG debug_out("k0",k0,64); #endif /* Step 4 */ for (i=0; i<64; i++) { k0xorIpad[i] = k0[i] ^ ipad; //将K0和ipad进行异或运算 } #ifdef HMAC_DEBUG debug_out("k0 xor ipad",k0xorIpad,64); #endif /* Step 5 */ for (i=0; i<64; i++) { step5data[i] = k0xorIpad[i]; } for (i=0; i<data_length; i++) { step5data[i+64] = data[i]; //将数据添加在第4步生成的字节串 后面 } #ifdef HMAC_DEBUG debug_out("(k0 xor ipad) || text",step5data,data_length+64); #endif /* Step 6 */ sha1(step5data, data_length+b, digest); //将第5步的结果运用H函数 #ifdef HMAC_DEBUG debug_out("Hash((k0 xor ipad) || text)",digest,20); #endif /* Step 7 */ for (i=0; i<64; i++) { step7data[i] = k0[i] ^ opad; //将K0和opad进行异或运算 } #ifdef HMAC_DEBUG debug_out("(k0 xor opad)",step7data,64); #endif /* Step 8 */ for (i=0; i<64; i++) { step8data[i] = step7data[i]; } for (i=0; i<20; i++) { step8data[i+64] = digest[i]; } #ifdef HMAC_DEBUG debug_out("(k0 xor opad) || Hash((k0 xor ipad) || text)",step8data, 20+64); #endif /* Step 9 */ sha1(step8data, b+20, digest); #ifdef HMAC_DEBUG debug_out("HASH((k0 xor opad) || Hash((k0 xor ipad) || text))", digest,20); #endif }
- 配置远程服务器IP地址和服务器端口
通过域名解析获取IP地址有两种方法:
a、通过在终端下ping域名的方法获取IP地址
b、通过DNS域名解析的方法获取IP地址
- 通过在终端下ping域名的方法获取IP地址
把 ${productKey}替换为您的产品key,并在终端对MQTT进行ping操作,来获取服务器IP地址
举例:
- 通过DNS域名解析的方法获取IP地址
首先完成W5500的DNS域名解析例程的移植,把DNS相关部分移植到本程序中,再进行相关配置即可完成(DNS相关例程下载地址http://www.w5500.com/) ,DNS解析域名成功后,把解析出的IP地址赋值给MQTT的 server_ip,用于MQTT与阿里云的连接,完成MQTT协议通信
连接成功后,通过串口调试助手验证DNS域名解析是否正确,若正确则MQTT与阿里云连接成功,并可成功的发布订阅消息:
- 设置发布订阅的主题:
在tcp_client.c文件中设置MQTT与阿里云连接参数,并通过调用mqtt_fun.c文件中的相关底层函数来完成MQTT与阿里云连接:
底层的订阅发布函数
/*****************拼接订阅报文**************************************/ void make_sub_msg(char *Topic,unsigned char*msgbuf,int buflen) { int msgid = 1; int req_qos = 0; unsigned char topic[100]; MQTTString topicString= MQTTString_initializer; memcpy(topic,Topic,strlen(Topic)); topicString.cstring = (char*)topic; //topicString.lenstring.len=4; MQTTSerialize_subscribe(msgbuf, buflen, 0, msgid, 1, &topicString, &req_qos); return; } /*********拼接发布报文******************/ void make_pub_msg(char *Topic,unsigned char*msgbuf,int buflen,char*msg) { unsigned char topic[100]; int msglen = strlen(msg); MQTTString topicString = MQTTString_initializer; memset(topic,0,sizeof(topic)); memcpy(topic,Topic,strlen(Topic)); topicString.cstring = (char*)topic; MQTTSerialize_publish(msgbuf, buflen, 0, 2, 0, 0, topicString, ( unsigned char*)msg, msglen); return; }
此发布订阅的主题根据阿里云中设备管理的Topic列表设置设备可以基于Topic列表中的Topic进行Pub/Sub通信,例如列表中有/TKKMt4nMF8U/MQTT1/mqtt,且设备拥有的权限是发布和订阅,这就意味着设备可以往这个Topic发布消息,同样设备可以从这个Topic订阅消息。
- 简单测试:
-
把程序下载到测试板并连接,登陆阿里云,到添加的设备,开启测试板,状态显示在线,说明MQTT与阿里云已经初步连接上
通过设备的Topic列表,选择程序中设置的发布订阅的Topic进行发布消息的操作:
串口打印接收到的服务器端发送的消息:
同时可在日志服务中查询相关设备的相关消息:
此时MQTT协议通信成功。
说明:在串口通信中会一直打印消息,是因为程序中设置了对MQTT的ping操作,防止MQTT离线。 - 注意:
在MQTT与阿里云连接时,会出现离线的状态,在离线状态时重启测试板并手动刷新阿里云即可。因为状态不是实时的显示,会有一段时间的延迟,可耐心等待。
MQTT CONNECT协议设置时的注意事项:错误码