MQTT接入使用樣本
本文以C Link SDK中的Demo檔案./mqtt_basic_demo.c為例，介紹如何調用Link SDK的API，將MQTT協議的裝置接入物聯網平台並進行訊息收發。
背景資訊
MQTT接入的更多資訊，請參見概述。
步驟一：初始化
添加標頭檔。
#include "aiot_state_api.h" #include "aiot_sysdep_api.h" #include "aiot_mqtt_api.h"配置底層依賴和日誌輸出。
aiot_sysdep_set_portfile(&g_aiot_sysdep_portfile); aiot_state_set_logcb(demo_state_logcb);調用aiot_mqtt_init，建立MQTT用戶端執行個體，並初始化預設參數。
mqtt_handle = aiot_mqtt_init(); if (mqtt_handle == NULL) { printf("aiot_mqtt_init failed\n"); return -1; }
步驟二：配置功能
調用aiot_mqtt_setopt，配置以下功能。
更多功能的配置項，請參見aiot_mqtt_option_t。
配置串連參數。
範例程式碼：
/* TODO: 替換為自己裝置的認證資訊。 */ char *product_key = "a18wP******"; char *device_name = "LightSwitch"; char *device_secret = "uwMTmVAMnGGHaAkqmeDY6cHxxB******"; /* TODO: 替換為自己裝置的接入網域名稱。 */ char *mqtt_host = "iot-06z00ax1o******.mqtt.iothub.aliyuncs.com"; ... ... /* 以下參數為可選，您可使用SDK檔案中預設內容。 */ /* 配置MQTT伺服器位址。 */ aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_HOST, (void *)url); /* 配置MQTT伺服器連接埠。 */ aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_PORT, (void *)&port); /* 配置裝置ProductKey。 */ aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_PRODUCT_KEY, (void *)product_key); /* 配置裝置DeviceName。 */ aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_DEVICE_NAME, (void *)device_name); /* 配置裝置DeviceSecret。 */ aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_DEVICE_SECRET, (void *)device_secret); /* 配置網路連接的安全憑據。 */ aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_NETWORK_CRED, (void *)&cred);相關參數：
參數
樣本
說明
mqtt_host
iot-06z00ax1o******.mqtt.iothub.aliyuncs.com
product_key
a18wP******
裝置認證資訊。更多資訊，請參見擷取裝置認證資訊。
本常式的身份認證方式為一機一密。
device_name
LightSwitch
device_secret
uwMTmVAMnGGHaAkqmeDY6cHxxB******
MQTT保活說明：
重要裝置端在保活時間間隔內，至少需要發送一次報文，包括ping請求。
從物聯網平台發送CONNACK響應CONNECT訊息時，開始心跳計時。收到PUBLISH、SUBSCRIBE、PING或 PUBACK訊息時，會重設計時器。物聯網平台每隔30秒定時檢測一次裝置的保活心跳，裝置上線時間點距離最新定時檢測時間點的時間，是定時檢測的等待時間。定義最大逾時時間為：
保活心跳時間*1.5+定時檢測的等待時間。超過最大逾時時間未收到裝置訊息，伺服器會自動中斷連線。
C Link SDK具備保活能力，您可以設定以下配置項，自訂裝置串連的保活心跳。如果不配置，則取預設值。
配置項
預設值
說明
AIOT_MQTTOPT_HEARTBEAT_MAX_LOST
2
可容忍的心跳丟失閾值。即：心跳請求報文達到設定的次數後，發起重連。
AIOT_MQTTOPT_HEARTBEAT_INTERVAL_MS
25,000
每次發起重連之間的間隔時間。單位毫秒， 取值範圍：1000~1,200,000。
AIOT_MQTTOPT_KEEPALIVE_SEC
1200
可容忍的心跳丟失時間閾值。即：失去心跳後，設定的時間內，允許發起重連。單位秒，取值範圍：30~1,200。建議取值大於300。
配置狀態監控和訊息回調。
配置狀態監控回呼函數。
範例程式碼：
int main(int argc, char *argv[]) { ... ... /* 配置MQTT預設訊息接收回呼函數。 */ aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_RECV_HANDLER, (void *)demo_mqtt_default_recv_handler); /* 配置MQTT事件回呼函數。 */ aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_EVENT_HANDLER, (void *)demo_mqtt_event_handler); ... ... }相關參數：
配置項
樣本值
說明
AIOT_MQTTOPT_RECV_HANDLER
demo_mqtt_default_recv_handler
當接收訊息時，根據該回呼函數定義的處理邏輯，執行對應的處理。
AIOT_MQTTOPT_EVENT_HANDLER
demo_mqtt_event_handler
當裝置串連狀態發生變化時，根據該回呼函數定義的處理邏輯，執行對應的處理。
定義狀態監控的回呼函數。
重要請勿將事件的處理邏輯，定義得過於耗時，以免阻塞收包線程。
串連狀態的變化包括網路異常、自動重連已成功、已中斷連線等。
如果要根據串連狀態的變化做應對處理，可在
TODO處，按照需要修改代碼。
/* MQTT事件回呼函數, 當網路連接、重連或斷開時，觸發該函數, 事件定義見core/aiot_mqtt_api.h。 */ void demo_mqtt_event_handler(void *handle, const aiot_mqtt_event_t *event, void *userdata) { switch (event->type) { /* 調用了aiot_mqtt_connect()介面, 與MQTT伺服器建立串連。 */ case AIOT_MQTTEVT_CONNECT: { printf("AIOT_MQTTEVT_CONNECT\n"); /* TODO: 處理SDK建立串連成功, 不可在此調用耗時較長的阻塞函數。 */ } break; /* SDK因網路狀況被動中斷連線後, 成功自動發起重連。 */ case AIOT_MQTTEVT_RECONNECT: { printf("AIOT_MQTTEVT_RECONNECT\n"); /* TODO: 處理SDK重連成功, 不可在此調用耗時較長的阻塞函數。 */ } break; /* SDK因網路狀況被動斷開了串連, network底層讀寫失敗, heartbeat沒有按預期得到服務端心跳應答。 */ case AIOT_MQTTEVT_DISCONNECT: { char *cause = (event->data.disconnect == AIOT_MQTTDISCONNEVT_NETWORK_DISCONNECT) ? ("network disconnect") : ("heartbeat disconnect"); printf("AIOT_MQTTEVT_DISCONNECT: %s\n", cause); /* TODO: 處理SDK被動中斷連線, 不可在此調用耗時較長的阻塞函數。 */ } break; default: { } } }定義訊息接收的回呼函數。
重要請勿將訊息的處理邏輯，定義得過於耗時，以免阻塞收包線程。
如果您要根據接收的訊息做應對處理，可在
TODO處，按照需要修改代碼。
/* MQTT預設訊息處理回調, 當SDK從伺服器收到MQTT訊息時, 且您未設定對應回調的處理時，以下介面被調用。 */ void demo_mqtt_default_recv_handler(void *handle, const aiot_mqtt_recv_t *packet, void *userdata) { switch (packet->type) { case AIOT_MQTTRECV_HEARTBEAT_RESPONSE: { printf("heartbeat response\n"); /* TODO: 處理伺服器對心跳的回應, 一般不處理。 */ } break; case AIOT_MQTTRECV_SUB_ACK: { printf("suback, res: -0x%04X, packet id: %d, max qos: %d\n", -packet->data.sub_ack.res, packet->data.sub_ack.packet_id, packet->data.sub_ack.max_qos); /* TODO: 處理伺服器對訂閱請求的回應, 一般不處理。 */ } break; case AIOT_MQTTRECV_PUB: { printf("pub, qos: %d, topic: %.*s\n", packet->data.pub.qos, packet->data.pub.topic_len, packet->data.pub.topic); printf("pub, payload: %.*s\n", packet->data.pub.payload_len, packet->data.pub.payload); /* TODO: 處理伺服器下發的業務報文。 */ } break; case AIOT_MQTTRECV_PUB_ACK: { printf("puback, packet id: %d\n", packet->data.pub_ack.packet_id); /* TODO: 處理伺服器對QoS=1上報訊息的回應, 一般不處理。 */ } break; default: { } } }
步驟三：請求串連
調用aiot_mqtt_connect，根據配置串連的參數，向物聯網平台，發起串連認證請求。
/* 與伺服器建立MQTT串連。 */
res = aiot_mqtt_connect(mqtt_handle);
if (res < STATE_SUCCESS) {
/* 嘗試建立串連失敗, 銷毀MQTT執行個體, 回收資源。 */
aiot_mqtt_deinit(&mqtt_handle);
printf("aiot_mqtt_connect failed: -0x%04X\n", -res);
return -1;
}步驟四：開啟保活線程
調用aiot_mqtt_process，向伺服器發送心跳報文，使裝置保持長串連狀態，並重發QoS=1的未應答報文。
開啟保活線程。
res = pthread_create(&g_mqtt_process_thread, NULL, demo_mqtt_process_thread, mqtt_handle); if (res < 0) { printf("pthread_create demo_mqtt_process_thread failed: %d\n", res); return -1; }設定保活線程處理函數。
void *demo_mqtt_process_thread(void *args) { int32_t res = STATE_SUCCESS; while (g_mqtt_process_thread_running) { res = aiot_mqtt_process(args); if (res == STATE_USER_INPUT_EXEC_DISABLED) { break; } sleep(1); } return NULL; }
步驟五：開啟接收線程
調用aiot_mqtt_recv，收取伺服器下發的MQTT訊息，根據訊息回呼函數，執行對應處理。在斷線時自動重連，根據事件回呼函數，執行對應處理。
開啟接收線程。
res = pthread_create(&g_mqtt_recv_thread, NULL, demo_mqtt_recv_thread, mqtt_handle); if (res < 0) { printf("pthread_create demo_mqtt_recv_thread failed: %d\n", res); return -1; }設定接收線程處理函數。
void *demo_mqtt_recv_thread(void *args) { int32_t res = STATE_SUCCESS; while (g_mqtt_recv_thread_running) { res = aiot_mqtt_recv(args); if (res < STATE_SUCCESS) { if (res == STATE_USER_INPUT_EXEC_DISABLED) { break; } sleep(1); } } return NULL; }
步驟六：訂閱Topic
調用aiot_mqtt_sub，訂閱指定Topic。
範例程式碼：
{ char *sub_topic = "/a18wP******/LightSwitch/user/get"; res = aiot_mqtt_sub(mqtt_handle, sub_topic, NULL, 1, NULL); if (res < 0) { printf("aiot_mqtt_sub failed, res: -0x%04X\n", -res); return -1; } }說明完成配置後，請刪除相關代碼兩邊的注釋符號。
相關參數：
參數
樣本
說明
sub_topic
/a18wP******/LightSwitch/user/get
擁有訂閱許可權的Topic。
a18wP******為裝置的ProductKey。LightSwitch為裝置的DeviceName。
本樣本為預設的自訂Topic。
裝置通過該Topic，可接收物聯網平台的訊息。
關於Topic的更多資訊，請參見什麼是Topic。
步驟七：發送訊息
調用aiot_mqtt_pub，向指定Topic發送訊息。
範例程式碼：
{ char *pub_topic = "/a18wP******/LightSwitch/user/update"; char *pub_payload = "{\"id\":\"1\",\"version\":\"1.0\",\"params\":{\"LightSwitch\":0}}"; res = aiot_mqtt_pub(mqtt_handle, pub_topic, (uint8_t *)pub_payload, (uint32_t)strlen(pub_payload), 0); if (res < 0) { printf("aiot_mqtt_sub failed, res: -0x%04X\n", -res); return -1; } }說明完成配置後，請刪除相關代碼兩邊的注釋符號。
相關參數：
參數
樣本
說明
pub_topic
/a18wP******/LightSwitch/user/update
擁有發布許可權的Topic。
a18wP******為裝置的ProductKey。LightSwitch為裝置的DeviceName。
裝置通過該Topic向物聯網平台發送訊息。
關於Topic的更多資訊，請參見什麼是Topic。
pub_payload
{\"id\":\"1\",\"version\":\"1.0\",\"params\":{\"LightSwitch\":0}}
上報至物聯網平台的訊息內容。
由於樣本訊息的Topic類型為自訂，因此資料格式可自訂。
關於資料格式的更多資訊，請參見資料格式。
裝置與物聯網平台建立MQTT通訊後，請確保通訊量不超過閾值。
步驟八：中斷連線
MQTT接入常應用於長串連的裝置，程式通常不會運行至此。
常式的主線程任務為配置參數並成功建立串連。串連建立後，主線程可進入休眠。
調用aiot_mqtt_disconnect，向物聯網平台發送中斷連線的報文，然後斷開網路連接。
res = aiot_mqtt_disconnect(mqtt_handle);
if (res < STATE_SUCCESS) {
aiot_mqtt_deinit(&mqtt_handle);
printf("aiot_mqtt_disconnect failed: -0x%04X\n", -res);
return -1;
}步驟九：退出程式
調用aiot_mqtt_deinit，銷毀MQTT用戶端執行個體，釋放資源。
res = aiot_mqtt_deinit(&mqtt_handle);
if (res < STATE_SUCCESS) {
printf("aiot_mqtt_deinit failed: -0x%04X\n", -res);
return -1;
}後續步驟