10 KAWAII MQTT 问题求助

发布于 2021-06-13 15:36:40 浏览：1208 订阅该版

各位朋友端午节快乐！！！！

求助一下，在做室内定位产品，使用到 KAWAII MQTT 与服务器通讯。

先介绍一下我的配置：
1. 创建一个 mqtt client
2. 创建两个发布消息，分别为**pm/status** 和 **pm/rawdata**；并启动对应线程，线程栈大小2048；这两个发布分别为**发送本机状态数据** 和 **发送原始数据**。
3. 创建两个订阅消息，分别为**pm/$00000b07/configuration** 和 **pm/configuration**;这两个订阅分别为订阅本机设置消息，订阅广播设置消息，有接收就要有回应，同时开启了两个回应线程，线程栈大小2048，回应的名字都是**pm/events**。
4. Qos 等级为 1。
5. 由于QOS 1 需要有服务器的响应，所以需要把发出去的消息暂存一下等服务器响应了销毁，但是这样带来的问题就是栈空间占用太大，并且我还将发送的二进制 base64 一下，base64 还是在线程内 malloc 的。所以有了以下的思路，判断一下mqtt_ack_handler_number，反正本地有缓存实时性不是要求特别高：
5.1 pm/status 线程
```C
static void thread_mqttpub_status(void *parameter)
{
    PackerMQTTPub_Type_Def* packer_status_param = (PackerMQTTPub_Type_Def*) parameter;

while (1)
    {
        if(packer_status_param->clientparam.client_param->mqtt_ack_handler_number == 0)
        {
            packer_pack_mqttpub(packer_status_param, RT_NULL, 0, 0);
            rt_thread_delay(10000);
        }
        else
        {
            rt_thread_delay(RT_TICK_PER_SECOND);
        }
    }
}
```
5.2 pm/rawdata 线程; 我是将数据放入缓存，再发送，并且每次发送不大于512字节
```C
static void thread_mqttpub_rawdata(void *parameter)
{
    PackerMQTTPub_Type_Def* packer_rawdata_param = (PackerMQTTPub_Type_Def*) parameter;
	rt_uint8_t PayloadCnt_Param = 0;

while (1)
    {
#ifdef EXIST_DWT2        
        if(((type_dwt1 == DEVICE_TYPE_ANCHOR) && (mode_dwt1.param.b_tdoa)) || 
			((type_dwt2 == DEVICE_TYPE_ANCHOR) && (mode_dwt2.param.b_tdoa)))
#else
        if((type_dwt1 == DEVICE_TYPE_ANCHOR) && (mode_dwt1.param.b_tdoa))
#endif            
        {
            if((packer_rawdata_param->clientparam.client_param->mqtt_ack_handler_number == 0) && 
                (SendDataforTdoa_Param->lwrt_datastoretdoa_handle.w != SendDataforTdoa_Param->lwrt_datastoretdoa_handle.r))
            {
                if((SendDataforTdoa_Param->lwrt_datastoretdoa_handle.w - SendDataforTdoa_Param->lwrt_datastoretdoa_handle.r) >= PACKET_PAYLOAD_PUBDATA_LEN_MQTT)
                {
                    PayloadCnt_Param = PACKET_PAYLOAD_PUBDATA_LEN_MQTT / sizeof(DataStoreTdoa_Type_Def);
                }
                else
                {
                    PayloadCnt_Param = ((SendDataforTdoa_Param->lwrt_datastoretdoa_handle.w - SendDataforTdoa_Param->lwrt_datastoretdoa_handle.r) / sizeof(DataStoreTdoa_Type_Def));
                }
            
                rt_uint8_t* TdoaData_SendbufferParam = (rt_uint8_t*) rt_malloc(sizeof(DataStoreTdoa_Type_Def) * PayloadCnt_Param);
                if(TdoaData_SendbufferParam == RT_NULL)
                {
                    LOG_E("TdoaData_SendbufferParam memory malloc failed = %d !", sizeof(DataStoreTdoa_Type_Def) * PayloadCnt_Param);
                }
            
                lwrb_read(&SendDataforTdoa_Param->lwrt_datastoretdoa_handle, TdoaData_SendbufferParam, sizeof(DataStoreTdoa_Type_Def) * PayloadCnt_Param);

packer_pack_mqttpub(packer_rawdata_param, TdoaData_SendbufferParam, sizeof(DataStoreTdoa_Type_Def), PayloadCnt_Param);
                
                rt_free(TdoaData_SendbufferParam);
            }
            else
            {
                rt_thread_delay(RT_TICK_PER_SECOND);    
            }
        }
        else
        {
            rt_thread_delay(RT_TICK_PER_SECOND);
        }
    }
}
```
5.3 pm/$00000b07/configuration 响应pm/events
```C
static void thread_mqttresponse_unicast(void *parameter)
{
    PackerMQTTSub_Type_Def* packer_unicast_param = (PackerMQTTSub_Type_Def*) parameter;

while (1)
    {
        if((packer_unicast_param->clientparam.client_param->mqtt_ack_handler_number == 0) && 
            (subunicast_response.senddata_status == DATA_TYPE_EXIST))
        {
            packer_pack_mqttsub(packer_unicast_param, subunicast_response.buffer, subunicast_response.senddata_length, subunicast_response.senddata_packetcount);
            subunicast_response.senddata_status &= (~DATA_TYPE_EXIST);
        }
        else
        {
            rt_thread_delay(RT_TICK_PER_SECOND);
        }
    }
}
```
5.4 pm/configuration 响应pm/events
```C
static void thread_mqttresponse_radio(void *parameter)
{
    PackerMQTTSub_Type_Def* packer_radio_param = (PackerMQTTSub_Type_Def*) parameter;

while (1)
    {
        if((packer_radio_param->clientparam.client_param->mqtt_ack_handler_number == 0) && 
            (subradio_response.senddata_status == DATA_TYPE_EXIST))
        {
            packer_pack_mqttsub(packer_radio_param, subradio_response.buffer, subradio_response.senddata_length, subradio_response.senddata_packetcount);
            subradio_response.senddata_status &= (~DATA_TYPE_EXIST);
        }
        else
        {
            rt_thread_delay(RT_TICK_PER_SECOND);
        }
    }
}
```
6. 现在在测试阶段，pm/status 10s 发送一次数据帧；pm/rawdata 不停的发送，没包数据不大于 512个字节；两个订阅不会发生， pm/events 也不会发生。

7. 平时使用很好，但是偶尔会报错，找了几天问题没发现，连接笔记本 开启 终端 不停监控，终于在昨天发现了。问题如下：
![Error.png](https://oss-club.rt-thread.org/uploads/20210613/980925eca452e2f028e48ed2bba3e778.png.webp)

8. 恳请各位路过的大神朋友给予指点，上述问题说的是 空间不足，但是我再发送之前都 if 了 ack num呀。并且 这个问题有时候一天不出，有是有2小时 3小时就出现。