目录

前言

一、熟悉MQTT的配置流程

二、熟悉bk7238的SDK中MQTT相关函数

三，熟悉关键结构体

四，封装连接函数

五，封装主题的订阅和发布函数

六，编写demo代码

---

前言

MQTT 是一种基于标准的消息传递协议或规则集，用于机器对机器的通信。智能传感器、可穿戴设备和其他物联网（IoT）设备通常必须通过带宽有限的资源受限网络传输和接收数据。这些物联网设备使用 MQTT 进行数据传输，因为它易于实施，并且可以有效地传输物联网数据。MQTT 支持设备到云端和云端到设备之间的消息传递。

一、熟悉MQTT的配置流程

1. MQTT 参数结构体准备
2. MQTT 客户端初始化
3. 设置回调函数（连接、消息、断线等）
4. 建立 MQTT 连接（TCP/SSL）
5. 订阅主题（SUBSCRIBE）
6. 发布消息（PUBLISH）
7. 接收消息（回调函数）
8. 断线重连处理（可选）

二、熟悉bk7238的SDK中MQTT相关函数

extern void mqtt_core_handler(void);

extern int mqtt_client_subscribe(struct mqtt_client_session *cs,char *topic);
extern int mqtt_client_publish(mqtt_client_session *client, enum QoS qos, const char *topic, const char *msg_str);

extern int matt_client_connect(mqtt_client_session* cs, MQTTPacket_connectData* options);
extern int mqtt_client_disconnect(mqtt_client_session* cs);

extern int mqtt_client_session_init(mqtt_client_session* cs);
extern int mqtt_client_session_init2(mqtt_client_session* cs,unsigned char *send_buf,
                    unsigned int sdb_len,unsigned char *read_buf,unsigned int rdb_len);


extern int mqtt_net_connect(mqtt_client_session* cs,char *host,int port);
extern int mqtt_net_disconnect(mqtt_client_session* cs);
这些函数足以帮我们建立起mqtt的连接了，接下来开始进行配置

三，熟悉关键结构体

其实就一个mqtt会话结构体，可以用来保存一个客户端的所有状态和资源

struct mqtt_client_session
{
    LIST_HEADER_T node;
 
    void *lock;
    
    unsigned int next_packetid;
    
    unsigned int command_timeout_ms;
 
    unsigned net_is_connected : 1;
    unsigned is_connected : 1;
    unsigned cleansession : 1;
    
    
    unsigned char *buf,*readbuf;
    unsigned short buf_size,readbuf_size;
    
    unsigned int keepAliveInterval;
    unsigned int tick_ping;
    
    tmqtt_client_netport netport;
    
    void (*connect_callback)(mqtt_client_session *);
    void (*online_callback)(mqtt_client_session *);
    void (*offline_callback)(mqtt_client_session *,MQTT_EVNT_T);
    void (*mqtt_notice_cb)(mqtt_client_session *,MQTT_EVNT_T);
    
    struct{
        unsigned short sub_topic_num;
        struct sub_msg_handlers *messageHandlers;
    }sub;
    
   void (*defaultMessageHandler)(mqtt_client_session *, MessageData *);
 
   LIST_HEADER_T msg_hd;
} ;

四，封装连接函数

接下来就是对该结构体进行配置，连接mqtt服务器（首先要先确保设备已经完成配网操作） 封装

连接函数

//连接mqtt，需要输入mqtt服务器地址，用户id，用户名，密码
int user_mqtt_start_connect(char *host, char *client_id, char *username, char *password)
{
 
    char *link_name = "broker.emqx.io";
    int ret;
    
    if(host != NULL)
    {
        link_name = host;
    }
    
    MQTTPacket_connectData connectData = MQTTPacket_connectData_initializer;
 
    if(mqtt_client_session_init(&_mqttc_ses) != MQTT_OK)
    {
        goto _exit;
    }
//使用TCP明文传输
#if 1
    tcp_mqtt_client_api_register( &_mqttc_ses.netport );
 
    if(mqtt_net_connect(&_mqttc_ses,link_name,1883) != MQTT_OK)
    {
        goto _exit;
    }
#else
    ssl_mqtt_client_api_register( &_mqttc_ses.netport );
 
    if(mqtt_net_connect(&_mqttc_ses,link_name,8883) != MQTT_OK)
    {
        goto _exit;
    }
 
#endif
    
    connectData.MQTTVersion = 3;
    connectData.clientID.cstring = client_id;
    connectData.username.cstring = username;
    connectData.password.cstring = password;
    connectData.keepAliveInterval = 30000;  //心跳包间隔
    connectData.cleansession = 1;
 
    _mqttc_ses.buf_size = MQTT_DEFUALT_BUF_SIZE;
    _mqttc_ses.buf = os_malloc(_mqttc_ses.buf_size);
    _mqttc_ses.readbuf_size = MQTT_DEFUALT_BUF_SIZE;
    _mqttc_ses.readbuf = os_malloc(_mqttc_ses.readbuf_size);
 
    if (!(_mqttc_ses.readbuf && _mqttc_ses.buf))
    {
        MQTT_LOG("no memory for MQTT client buffer!\n");
        goto _exit;
    }
    
    /* set event callback function */
    _mqttc_ses.connect_callback = mqtt_connect_callback;
    _mqttc_ses.online_callback = mqtt_online_callback;
    _mqttc_ses.offline_callback = mqtt_offline_callback;
    _mqttc_ses.mqtt_notice_cb = mqtt_notice_event_callback;
    
    /* set subscribe table and event callback */
    // _mqttc_ses.sub.messageHandlers = sub_topic;
    // if(sub_topic[0].topicFilter == NULL)
    // {
    //     sub_topic[0].topicFilter = mqt_strdup("testtopic/1111");
    // }
    // sub_topic[0].callback = mqtt_sub_callback;
    // sub_topic[0].qos = QOS0;
    // if(sub_topic[1].topicFilter == NULL)
    // {
    //     sub_topic[1].topicFilter = mqt_strdup("testtopic/2222");
    // }
    // sub_topic[1].callback = mqtt_sub_callback;
    // sub_topic[1].qos = QOS1;
    // _mqttc_ses.sub.sub_topic_num = sizeof(sub_topic)/sizeof(struct sub_msg_handlers);
 
 
    _mqttc_ses.defaultMessageHandler =  mqtt_default_sub_callback;
    
    ret = matt_client_connect(&_mqttc_ses,&connectData);
    if(MQTT_OK != ret)
    {
        MQTT_LOG("[MQTT]connect failed\r\n");
        goto _exit;
    }
    MQTT_LOG("[MQTT]connect succeed\r\n");
 
    return 0;
_exit:
    if(_mqttc_ses.readbuf)
    {
        os_free(_mqttc_ses.readbuf);
    }
    if(_mqttc_ses.buf)
    {
        os_free(_mqttc_ses.buf);
    }
    extern int mqtt_client_session_deinit(mqtt_client_session* cs);
    mqtt_client_session_deinit(&_mqttc_ses);
    return -1;
}

需要对配置一些基础的回调函数，其中较为主要的就是接收到主题消息的回调函数，当订阅的主题有消息时，触发回调函数，我们可以在这个函数中获取到发布到该主题的消息

static void mqtt_default_sub_callback(mqtt_client_session *c, MessageData *msg_data)
{
	os_memset(&mqtt_msg_data, 0, sizeof(mqtt_msg_data));
	MQTT_LOG("default Message arrived on topic %.*s: %.*s\n", msg_data->topicName->lenstring.len,
		msg_data->topicName->lenstring.data,
		msg_data->message->payloadlen, msg_data->message->payload);
 
    return;
}
 
//mqtt连接回调函数，当mqtt连接成功时，会调用此函数。不一定保活
static void mqtt_connect_callback(mqtt_client_session *c)
{
    MQTT_LOG("inter mqtt_connect_callback!\r\n");
}
 
//mqtt完全上线回调函数， 表示mqtt已经完全上线，可以进行通信
static void mqtt_online_callback(mqtt_client_session *c)
{
    MQTT_LOG("inter mqtt_online_callback!\r\n");
}
 
//MQTT断开连接或异常掉线时，会调用此函数
static void mqtt_offline_callback(mqtt_client_session *c,MQTT_EVNT_T ev)
{
    MQTT_LOG("inter mqtt_offline_callback event %d\r\n",ev);
}
 
static void mqtt_notice_event_callback(mqtt_client_session *c,MQTT_EVNT_T ev)
{
    MQTT_LOG("mqtt nitice event %d\r\n",ev);
}

 五，封装主题的订阅和发布函数

//订阅主题，输入主题名字符串和qos服务质量级别
void user_mqtt_subscribe_topic(char *topic, int qos)
{
    mqtt_client_subscribe(&_mqttc_ses,topic);  //其底层只能配置qos2,已经写死了
}
 
//发布主题,输入主题名字符串，消息字符串，qos服务质量级别，qos只能是qos1
void user_mqtt_publish_topic(char *topic, char *msg, int qos)
{
    mqtt_client_publish(&_mqttc_ses, qos, topic, msg);
}
 
//断开mqtt连接,包括断开TCP层的网络连接
void user_mqtt_disconnect_about_net(void)
{
    mqtt_client_disconnect(&_mqttc_ses);
    mqtt_net_disconnect(&_mqttc_ses);
}
 
//断开mqtt连接,不包括断开TCP层的网络连接
void user_mqtt_disconnect(void)
{
    mqtt_client_disconnect(&_mqttc_ses);
}

六，编写demo代码

以上已经将bk7238连接mqtt所需的函数都封装准备好了，只需要在demo示例中调用即可：

 
//初始化mqtt任务，包括连接mqtt，订阅对应的主题，创建总理线程
void user_mqtt_task_init(void)
{
	int ret = 0;
    MQTT_LOG("[UAER MQTT]%s:%d\r\n",__FUNCTION__,__LINE__);
	wifi_connect("WIFI_MAC", "WIFI_NAME", NAME_LEN, "PASSWD", PASSWD_LEN);
	rtos_delay_milliseconds(3000);
	if(wifi_is_connected() == true)
	{
		MQTT_LOG("wifi is connected\r\n");
	}
 
	char clientID[32] = { 0 }; // 对应 MAC
    //char product_key[32] = {0}; // 对应 PID
    char password[128] = { 0 };   // 对应 SECRET
	strcpy(clientID, MQTT_MAC);
    strcpy(password, MQTT_DEVICE_SECRET);
 
	char subscribeTopic[128] = {0};
	char publishTopic[128] = {0};
 
	snprintf(subscribeTopic, MQTT_SUBSCRIBE_LEN + 1, MQTT_SUBSCRIBE, PRODUCT_ID, clientID);
    snprintf(publishTopic, MQTT_PUBLISH_LEN + 1, MQTT_PUBLISH, PRODUCT_ID, clientID);
 
	MQTT_LOG("clientID:%s\r\n", clientID);
	MQTT_LOG("password:%s\r\n", password);
	MQTT_LOG("subscribeTopic:%s\r\n", subscribeTopic);
	MQTT_LOG("publishTopic:%s\r\n", publishTopic);
 
 
    ret = user_mqtt_start_connect(NULL, clientID, clientID, password);
	if(ret != 0)
	{
		MQTT_LOG("mqtt connect fail\r\n");
		goto _exit;
	}
 
	user_mqtt_subscribe_topic(subscribeTopic, 2);
 
	user_mqtt_publish_topic(publishTopic, "hello world", 1); 
	
	mqtt_core_handler_thread_init(); //这个会启动一个线程，用于总理mqtt，发送心跳
 
 
_exit:
	if(ret != 0)
		user_mqtt_disconnect_about_net(); //断开mqtt连接,包括断开TCP层网络连接
}

在user_main中调用该函数，会自动连接mqtt，订阅get主题，并且向set主题发布一条信息