W5500通過MQTT連接阿里云平臺

1、簡介

1.1 開發(fā)環(huán)境與連接平臺

本文主要介紹W5500如何通過MQTT協(xié)議將設備連接到阿里云IoT，并通過MQTT協(xié)議實現(xiàn)通信。MQTT協(xié)議是基于TCP的協(xié)議，所以我們只需要在單片機端實現(xiàn)TCP客戶端代碼之后就很容易移植MQTT了， +W5500實現(xiàn)TCP客戶端的代碼我們以前已經實現(xiàn)過，程序下載： 

軟件環(huán)境：Windows

• 硬件環(huán)境：STM32F103+W5500
• 開發(fā)工具：Keil uVision5
• 調試工具：Wireshark、串口調試助手
• 連接平臺：阿里云-華東2節(jié)點（https://www.aliyun.com）

• 1.2 MQTT簡介：

MQTT官網(wǎng)地址：（http://mqtt.org/）

• 1.2.1 MQTT協(xié)議特點

MQTT是一個基于客戶端-服務器的消息發(fā)布/訂閱傳輸協(xié)議。MQTT協(xié)議是輕量、簡單、開放和易于實現(xiàn)的，這些特點使它適用范圍非常廣泛。在很多情況下，包括受限的環(huán)境中，如：機器與機器（M2M）通信和物聯(lián)網(wǎng)（IoT）。其在，通過衛(wèi)星鏈路通信傳感器、偶爾撥號的醫(yī)療設備、智能家居、及一些小型化設備中已廣泛使用。

MQTT協(xié)議當前版本為，2014年發(fā)布的MQTT v3.1.1。除標準版外，還有一個簡化版MQTT-SN，該協(xié)議主要針對嵌入式設備，這些設備一般工作于百TCP/IP網(wǎng)絡，如：ZigBee。

MQTT協(xié)議運行在TCP/IP或其他網(wǎng)絡協(xié)議，提供有序、無損、雙向連接。其特點包括：

1. 使用的發(fā)布/訂閱消息模式，它提供了一對多消息分發(fā)，以實現(xiàn)與應用程序的解耦。
2. 對負載內容屏蔽的消息傳輸機制。
3. 對傳輸消息有三種服務質量（QoS）：

• 最多一次，這一級別會發(fā)生消息丟失或重復，消息發(fā)布依賴于底層TCP/IP網(wǎng)絡。即：<=1
• 至多一次，這一級別會確保消息到達，但消息可能會重復。即：>=1
• 只有一次，確保消息只有一次到達。即：＝1。在一些要求比較嚴格的計費系統(tǒng)中，可以使用此級別

數(shù)據(jù)傳輸和協(xié)議交換的最小化（協(xié)議頭部只有2字節(jié)），以減少網(wǎng)絡流量

通知機制，異常中斷時通知傳輸雙方

• MQTT協(xié)議原理及實現(xiàn)方式

實現(xiàn)MQTT協(xié)議需要：客戶端和服務器端

MQTT協(xié)議中有三種身份：發(fā)布者（Publish）、代理（Broker）（服務器）、訂閱者（Subscribe）。其中，消息的發(fā)布者和訂閱者都是客戶端，消息代理是服務器，消息發(fā)布者可以同時是訂閱者。

MQTT傳輸?shù)南⒎譃椋褐黝}（Topic）和消息的內容（payload）兩部分

Topic，可以理解為消息的類型，訂閱者訂閱（Subscribe）后，就會收到該主題的消息內容（payload）

payload，可以理解為消息的內容，是指訂閱者具體要使用的內容

• 連接

1. 阿里云連接步驟：

• 以aliyun賬號直接進入IoT控制臺，如果還沒有開通阿里云物聯(lián)網(wǎng)套件服務，則 申請開通
• 接入引導

（1）、創(chuàng)建產品

（2）、添加設備

（3）、獲取設備的Topic

• 創(chuàng)建產品

初步進入控制臺后，需要創(chuàng)建產品。點擊創(chuàng)建產品。產品相當于某一類設備的集合，用戶可以根據(jù)產品管理其設備等。

• 產品名稱：對產品命名，例如可以填寫產品型號。產品名稱在賬號內保持唯一。
• productKey：阿里云IoT為產品頒發(fā)的全局唯一標識符

添加設備

創(chuàng)建完產品之后，可以為該產品添加設備。進入產品管理頁面下的設備管理，點擊添加設備。

• 說明：用戶可以自定義設備名稱(即deviceName)，這個名稱即可作為設備唯一標識符，用戶可以基于該設備名稱與IoT Hub進行通信，需要指出的是，用戶需要保證deviceName產品內唯一。
• 設備證書：添加設備之后，物聯(lián)網(wǎng)套件為設備頒發(fā)的唯一標識符，設備證書用于設備認證以及設備通信，詳細的請參考設備接入文檔。
• deviceName：用戶自定義設備唯一標識符，用于設備認證以及設備通信，用戶保證產品維度內唯一。
• deviceSecret：物聯(lián)網(wǎng)套件為設備頒發(fā)的設備秘鑰，用于認證加密，與deviceName或者deviceId成對出現(xiàn)。
• 獲取設備的Topic

添加設備之后，可以獲取設備的Topic。點擊Topic列表

• 說明：創(chuàng)建產品之后，物聯(lián)網(wǎng)套件都會為產品默認定義三個Topic類。那么，在添加設備之后，每個設備都會默認有三個Topic，即圖中所示。如果想要增加、修改、刪除Topic，請到消息通信重新定義Topic類。
• 設備可以基于Topic列表中的Topic進行Pub/Sub通信，例如列表中有/1000118502/test9/update，且設備擁有的權限是發(fā)布，這就意味著設備可以往這個Topic發(fā)布消息；同樣，列表中/1000118502/test9/get,權限是訂閱，這就意味著設備可以從這個Topic訂閱消息。
• 設備接入

獲得productKey、設備證書以及設備的Topic這些參數(shù)，就可以基于aliyun IoT device SDK for C將設備連接上IoT Hub并進行通信，具體請參考《MQTT配置》部分

2. MQTT移植步驟：

MQTT代碼源碼下載地址:（http://www.eclipse.org/paho/）

MQTT的移植非常簡單，將C/C++ MQTT Embedded clients的代碼添加到工程中，然后我們只需要再次封裝4個函數(shù)即可：

int transport_sendPacketBuffer(unsigned char* buf, int buflen);

通過網(wǎng)絡以TCP的方式發(fā)送數(shù)據(jù)；

int transport_getdata(unsigned char* buf, int count);

TCP方式從服務器端讀取數(shù)據(jù)，該函數(shù)目前屬于阻塞函數(shù)；

int transport_open(void);

打開一個網(wǎng)絡接口，其實就是和服務器建立一個TCP連接；

int transport_close(void);

關閉網(wǎng)絡接口。

如果已經移植好了socket方式的TCP客戶端的程序，那么這幾個函數(shù)的封裝也是非常簡單的，程序代碼如下所示：

/**

* @brief  通過TCP方式發(fā)送數(shù)據(jù)到TCP服務器

* @param  buf數(shù)據(jù)首地址

* @param  buflen數(shù)據(jù)長度

* @retval 小于0表示發(fā)送失敗

*/

/*訂閱消息*/

int Subscribe_sendPacketBuffer(unsigned char* buf, int buflen)

{

return send(SOCK_TCPS,buf,buflen);

}

/*發(fā)布消息*/

int Published_sendPacketBuffer(unsigned char* buf, int buflen)

{

return send(SOCK_TCPC,buf,buflen);

}

/**

* @brief  阻塞方式接收TCP服務器發(fā)送的數(shù)據(jù)

* @param  buf數(shù)據(jù)存儲首地址·

* @param  count數(shù)據(jù)緩沖區(qū)長度

* @retval 小于0表示接收數(shù)據(jù)失敗

*/

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連接參數(shù)說明

舉例：

• MQTT與阿里云連接函數(shù)：

參考阿里云內MQTT設備接入手冊，計算出設備連接的各項參數(shù)，例如下列程序中框中的部分為本例程MQTT與阿里云連接的參數(shù)的配置，詳細內容如下：

clientId = 192.168.207.115

deviceName = MQTT1

productKey = TKKMt4nMF8U

timestamp = 789（毫秒值）

signmethod = hmacsha1（算法類型）

deviceSecret = secret

那么使用tcp方式提交給mqtt參數(shù)分別如下：

• 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連接函數(shù)原型，tcp_client.c文件中的MQTT_CON_ALI函數(shù)中調用make_con_msg函數(shù)并通過阿里云設備的參數(shù)，設置MQTT連接阿里云函數(shù)的參數(shù)：

1. void make_con_msg(char* clientID,int keepalive, uint8 cleansession,
2. char*username,char* password,unsigned char*buf,int
3.                   buflen)
4. {
5.     int32_t len,rc;
6.     MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
7.     data.clientID.cstring = clientID;
8.     data.keepAliveInterval = keepalive;
9.     data.cleansession = cleansession;
10.     data.username.cstring = username;
11.     data.password.cstring = password;
12.     len = MQTTSerialize_connect(buf, buflen, &data);
13.           //構造鏈接報文
14.     return;
15. } MQTT連接過程：

void MQTT_CON_ALI(void)

{

int len;

int type;

switch (getSn_SR(0)) {

//獲取socket0的狀態(tài)

case SOCK_INIT:

//Socket處于初始化完成（打開）狀態(tài)

connect(0, server_ip,server_port);

//配置Sn_CR為CONNECT，并向TCP服務器發(fā)出連接請求￠

break;

case SOCK_ESTABLISHED:                //

Socket處于連接建立狀態(tài)

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?拼接連接報文

*根據(jù)阿里云平臺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));

// 接收到數(shù)據(jù)后再回給服務器，完成數(shù)據(jù)回環(huán)

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));  //

接收到數(shù)據(jù)后再回給服務器，完成數(shù)據(jù)回環(huán)

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處于等待關閉狀態(tài)

close(0);                               // 關閉Socket0

break;

case SOCK_CLOSED:

// Socket處于關閉狀態(tài)

socket(0,Sn_MR_TCP,local_port,Sn_MR_ND);

// 打開Socket0，并配置為TCP無延時模式，打開一個本地端口

break;

}

}

• Password有兩種獲得方法：
• 通過網(wǎng)頁“在線加密解密”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字節(jié)長度是否等于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];

//將數(shù)據(jù)添加在第4步生成的字節(jié)串

后面

}

#ifdef HMAC_DEBUG

debug_out("(k0 xor ipad) || text",step5data,data_length+64);

#endif

/* Step 6 */

sha1(step5data, data_length+b, digest);

//將第5步的結果運用H函數(shù)

#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地址

1. 通過在終端下ping域名的方法獲取IP地址

把 ${productKey}替換為您的產品key，并在終端對MQTT進行ping操作，來獲取服務器IP地址

舉例：

1. 通過DNS域名解析的方法獲取IP地址

首先完成W5500的DNS域名解析例程的移植，把DNS相關部分移植到本程序中，再進行相關配置即可完成(DNS相關例程下載地址http://www.w5500.com/) ，DNS解析域名成功后，把解析出的IP地址賦值給MQTT的 server_ip，用于MQTT與阿里云的連接，完成MQTT協(xié)議通信

連接成功后，通過串口調試助手驗證DNS域名解析是否正確，若正確則MQTT與阿里云連接成功，并可成功的發(fā)布訂閱消息：

• 設置發(fā)布訂閱的主題：

在tcp_client.c文件中設置MQTT與阿里云連接參數(shù)，并通過調用mqtt_fun.c文件中的相關底層函數(shù)來完成MQTT與阿里云連接：

底層的訂閱發(fā)布函數(shù)

/*****************拼接訂閱報文**************************************/

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;

}

/*********拼接發(fā)布報文******************/

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;

}

此發(fā)布訂閱的主題根據(jù)阿里云中設備管理的Topic列表設置設備可以基于Topic列表中的Topic進行Pub/Sub通信，例如列表中有/TKKMt4nMF8U/MQTT1/mqtt，且設備擁有的權限是發(fā)布和訂閱，這就意味著設備可以往這個Topic發(fā)布消息，同樣設備可以從這個Topic訂閱消息。

• 簡單測試：
• 把程序下載到測試板并連接，登陸阿里云，到添加的設備，開啟測試板，狀態(tài)顯示在線，說明MQTT與阿里云已經初步連接上
  通過設備的Topic列表，選擇程序中設置的發(fā)布訂閱的Topic進行發(fā)布消息的操作：
  串口打印接收到的服務器端發(fā)送的消息：
  同時可在日志服務中查詢相關設備的相關消息：
  此時MQTT協(xié)議通信成功。
  說明：在串口通信中會一直打印消息，是因為程序中設置了對MQTT的ping操作，防止MQTT離線。
• 注意：

在MQTT與阿里云連接時，會出現(xiàn)離線的狀態(tài)，在離線狀態(tài)時重啟測試板并手動刷新阿里云即可。因為狀態(tài)不是實時的顯示，會有一段時間的延遲，可耐心等待。
MQTT CONNECT協(xié)議設置時的注意事項：錯誤碼