kawaiimqtt:

 1 /* 5. wifi connect */
 2     rt_wlan_connect(WIFI_SSID, WIFI_KEY);
 3 
 4     /* 6. startup mqtt client  */
 5     mqtt_log_init();
 6     rt_snprintf(cid, sizeof(cid), "rtthread%d", rt_tick_get());
 7     /* check network connection status */
 8 
 9     net_dev = netdev_get_by_name("w0");
10     while(!(netdev_is_internet_up(net_dev)))
11     {
12         rt_thread_mdelay(100);
13         timeout++;
14         if(timeout == 200)
15         {
16             rt_kprintf("wifi connect failed!\r\n");
17             return -RT_ERROR;
18         }
19     }
20 
21     client = mqtt_lease();
22 
23     mqtt_set_host(client, MQTT_URL);
24     mqtt_set_port(client, MQTT_PORT);
25     mqtt_set_user_name(client, "rt-thread");
26     mqtt_set_password(client, "rt-thread");
27     mqtt_set_client_id(client, cid);
28     mqtt_set_clean_session(client, 1);
29 
30     if(mqtt_connect(client))
31     {
32         KAWAII_MQTT_LOG_E("%s:%d %s()... mqtt connect failed...", __FILE__, __LINE__, __FUNCTION__);
33         is_started = 0;
34         return -RT_ERROR;
35     }
36 
37     is_started = 1;
38     mqtt_subscribe(client, SUB1_NAME, QOS0, sub_topic_handle_led);
39     mqtt_subscribe(client, SUB2_NAME, QOS1, sub_topic_handle_num);
40     mqtt_subscribe(client, SUB_OTA, QOS2, sub_topic_handle_ota);
41 
42     tid3 = rt_thread_create("mq_pub", mqtt_t_publish, RT_NULL, 2048, 13, 10);
43     if (tid3 != RT_NULL)
44     {
45         rt_thread_startup(tid3);
46     }

1保证网络连接

2设置mqtt客户端：远端URL,端口，客户端ID,用户名，密码。。。

3连接远程mqtt服务器

4订阅主题：主题,Qos,收到主题时的响应（将主题名和主题内容分离出来，然后匹配响应）

  1 typedef struct {
  2     char **str;     //the PChar of string array
  3     size_t num;     //the number of string
  4 }IString;
  5 
  6 /* 拆分字符串 */
  7 static int Split(char *src, char *delim, IString* istr)//split buf
  8 {
  9     int i;
 10     char *str = NULL, *p = NULL;
 11 
 12     (*istr).num = 1;
 13     str = (char*)rt_calloc(strlen(src)+1,sizeof(char));
 14     if (str == NULL) return 0;
 15     (*istr).str = (char**)rt_calloc(1,sizeof(char *));
 16     if ((*istr).str == NULL) return 0;
 17     strcpy(str,src);
 18 
 19     p = strtok(str, delim);
 20     (*istr).str[0] = (char*)rt_calloc(strlen(p)+1,sizeof(char));
 21     if ((*istr).str[0] == NULL) return 0;
 22     strcpy((*istr).str[0],p);
 23     for(i = 1; (p = strtok(NULL, delim)); i++)
 24     {
 25         (*istr).num++;
 26         (*istr).str = (char**)rt_realloc((*istr).str,(i+1)*sizeof(char *));
 27         if ((*istr).str == NULL) return 0;
 28         (*istr).str[i] = (char*)rt_calloc(strlen(p)+1,sizeof(char));
 29         if ((*istr).str[0] == NULL) return 0;
 30         strcpy((*istr).str[i],p);
 31     }
 32     rt_free(str);
 33     str = p = NULL;
 34 
 35     return 1;
 36 }
 37 static void sub_topic_handle_led(void* client, message_data_t* msg)
 38 {
 39     (void) client;
 40     KAWAII_MQTT_LOG_I("-----------------------------------------------------------------------------------\r\n");
 41     KAWAII_MQTT_LOG_I("%s:%d %s()...\ntopic: %s\nmessage:%s\r\n", __FILE__, __LINE__, __FUNCTION__, msg->topic_name, (char*)msg->message->payload);
 42     KAWAII_MQTT_LOG_I("-----------------------------------------------------------------------------------\r\n");
 43 
 44     int i;
 45     IString istr;
 46     struct file_msg file_msg;
 47     char *tick_num;
 48 
 49     if (Split(msg->message->payload," ",&istr))
 50     {
 51         for (i = 0; i < istr.num; i++)
 52             rt_kprintf("%s\n",istr.str[i]);
 53 
 54         if(i == 2)
 55         {
 56             if(rt_strncmp(istr.str[0], "led", 3))
 57             {
 58                 KAWAII_MQTT_LOG_E("command error!\r\n");
 59             }
 60             else
 61             {
 62                 if(!(rt_strncmp(istr.str[1], "off", 3)))
 63                 {
 64                     rt_pin_write(LED_PIN_RED, PIN_HIGH);
 65                 }
 66                 else if(!(rt_strncmp(istr.str[1], "on", 2)))
 67                 {
 68                     rt_pin_write(LED_PIN_RED, PIN_LOW);
 69                 }
 70                 else
 71                 {
 72                     KAWAII_MQTT_LOG_E("state command error!\r\n");
 73                     for (i = 0; i < istr.num; i++)
 74                         rt_free(istr.str[i]);
 75                     rt_free(istr.str);
 76                     return;
 77                 }
 78 
 79                 tick_num = rt_calloc(1, 10);
 80                 if (tick_num == RT_NULL) {
 81                     rt_kprintf("memory is not enough \r\n");
 82                 }
 83                 else {
 84                     itoa(rt_tick_get(), tick_num, 10);
 85                     file_msg.timestamp = tick_num;
 86                     file_msg.cmd = istr.str[0];
 87                     file_msg.state = istr.str[1];
 88                     file_msg.str = istr.str;
 89 
 90                     rt_mq_send(&mq_demo, &file_msg, sizeof(file_msg));
 91                 }
 92             }
 93         }
 94         else
 95         {
 96             KAWAII_MQTT_LOG_E("command error! \r\n");
 97             for (i = 0; i < istr.num; i++)
 98                 rt_free(istr.str[i]);
 99             rt_free(istr.str);
100         }
101 
102     }
103     else
104     {
105         KAWAII_MQTT_LOG_E("Split failure!\r\n");
106     }
107 }

5 建立发布线程用于发布主题（设置主题Oos,负载内容，然后向指定主题发送（若是数据采集可以用消息队列等待发送）。）

 1 /* 板子发布消息的线程 */
 2 static void mqtt_t_publish(void *parameter)
 3 {
 4     char pub_buf[64];
 5     mqtt_message_t msg;
 6     memset(&msg, 0, sizeof(msg));
 7 
 8     rt_thread_mdelay(2 * RT_TICK_PER_SECOND);
 9 
10     while(1)
11     {
12         if(is_started)
13         {
14             memset(pub_buf, 0, 64);
15 
16             sprintf(pub_buf, "%d", usMRegHoldBuf[SLAVE_ADDR - 1][MB_RECV_REG_NUM]);
17 
18             msg.qos = QOS0;
19             msg.payload = pub_buf;
20 
21             mqtt_publish(client, PUB_NAME, &msg);
22         }
23         rt_thread_mdelay(5 * RT_TICK_PER_SECOND);
24     }
25 }