cjson MQTT

MQTTSerialize_publish len: -2

发布于 2024-07-09 11:50:03 浏览：520 订阅该版

使用pahomqtt，系统运行后开启mqtt，串口接收数据解析正确后发送tx7_mq的消息队列，在mqtt的publish线程中接收到该消息队列后生成json格式数据，然后通过mqtt进行发布，当json数据比较少时，可以正常发送，当数据比较多时，发送的主题toplic会变成很多的乱码“#”，导致 MQTTSerialize_publish len: -2

![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20240709/d2848d7c3276f11c14856e7e312b0207.png)
线程堆栈4096
![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20240709/30f7a513b790f93cef9c8dba225a4b2d.png)
mqtt的publish大小是4096
![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20240709/1970ca9d7d993556b636717f1d929103.png)
问题就出现在当json数据较多时，toplic就会变成乱码，导致长度计算出错，应该怎么解决这个问题呢
MQTT 的启动和初始化程序

```c
static void mq_start(void)
{

/* init condata param by using MQTTPacket_connectData_initializer */
    MQTTPacket_connectData condata = MQTTPacket_connectData_initializer;

rt_memset(&client, 0, sizeof(MQTTClient));
    struct netif *netif = netif_list; // 获取网络接口列表的第一个接口，根据实际情况可能需要选择特定的接口

/* config MQTT context param */
    {
        client.uri = MQTT_TEST_SERVER_URI;

/* config connect param */
        memcpy(&client.condata, &condata, sizeof(condata));

if (netif != RT_NULL)
        {
            client.condata.clientID.cstring = ipaddr_ntoa(&(netif->ip_addr));
        }
        else
        {
            client.condata.clientID.cstring = MQTT_CLIENTID;
        }
        rt_kprintf("ClientID=%s\n", client.condata.clientID.cstring);
        client.condata.keepAliveInterval = 60;
        client.condata.cleansession = 1;
        client.condata.username.cstring = MQTT_USERNAME;
        client.condata.password.cstring = MQTT_PASSWORD;

/* config MQTT will param. */
        client.condata.willFlag = 1;
        client.condata.will.qos = MQTT_TEST_QOS;
        client.condata.will.retained = 0;
        strcpy(buffer, MQTT_PUB_ORDER); // 将第一个字符串复制到缓冲区
        strcat(buffer, MQTT_DEVICE_ID); // 将第二个字符串追加到缓冲区
        strcat(buffer, MQTT_PUBTOPIC); // 将第二个字符串追加到缓冲区
        client.condata.will.topicName.cstring = buffer;
        rt_kprintf("Publish=%s\n", client.condata.will.topicName.cstring);
        client.condata.will.message.cstring = MQTT_WILLMSG;

/* malloc buffer. */
        client.buf_size = client.readbuf_size = MQTT_PUB_SUB_BUF_SIZE;
        client.buf = rt_malloc(client.buf_size);
        client.readbuf = rt_malloc(client.readbuf_size);
        if (!(client.buf && client.readbuf))
        {
            rt_kprintf("no memory for MQTT client buffer!\n");
            goto _exit;
        }

/* set event callback function */
        client.connect_callback = mqtt_connect_callback;
        client.online_callback = mqtt_online_callback;
        client.offline_callback = mqtt_offline_callback;

/* set subscribe table and event callback */
        client.messageHandlers[0].topicFilter = rt_strdup(MQTT_SUBTOPIC);
        client.messageHandlers[0].callback = mqtt_sub_callback;
        client.messageHandlers[0].qos = MQTT_TEST_QOS;

/* set default subscribe event callback */
        client.defaultMessageHandler = mqtt_sub_callback;
    }

/* run mqtt client */
    paho_mqtt_start(&client);

return;

_exit: if (client.buf)
    {
        rt_free(client.buf);
        client.buf = RT_NULL;
    }
    if (client.readbuf)
    {
        rt_free(client.readbuf);
        client.readbuf = RT_NULL;
    }
    return;
}
```
mqtt的publish线程

```c
void thread_pub(void *parameter)
{
    data_packet_t packet;
    rt_kprintf("thread_pub start\n");
    while (1)
    {
        if (rt_mq_recv(tx7_mq, &packet, sizeof(data_packet_t), RT_WAITING_FOREVER) == RT_EOK)
        {
            rt_kprintf("rt_mq_recv(tx7_mq)\n");
            switch (packet.mode)
            {
            case READ_INFO_BACK:
            {
                // 创建根对象
                cJSON *root = cJSON_CreateObject();

// 添加 DeviceID
                cJSON_AddItemToObject(root, "DeviceID", cJSON_CreateNumber(1));

// 添加 ArgCount
                cJSON_AddItemToObject(root, "ArgCount", cJSON_CreateNumber(30));

// 添加 Result
                cJSON_AddItemToObject(root, "Result", cJSON_CreateString("Success"));

cJSON_AddItemToObject(root, "Message", cJSON_CreateString("ok"));

// 创建 DeviceInfo 对象
                cJSON *deviceInfo = cJSON_CreateObject();
                cJSON_AddItemToObject(root, "DeviceInfo", deviceInfo);

// 向 DeviceInfo 添加字段
                cJSON_AddItemToObject(deviceInfo, "UserID", cJSON_CreateNumber(0));
                cJSON_AddItemToObject(deviceInfo, "PowerCurveEnable", cJSON_CreateString("true"));
                cJSON_AddItemToObject(deviceInfo, "CountFunctionEnable", cJSON_CreateString("false"));
                cJSON_AddItemToObject(deviceInfo, "RepeatResultEnable", cJSON_CreateString("true"));
                cJSON_AddItemToObject(deviceInfo, "MonitorModeEnable", cJSON_CreateString("true"));
                cJSON_AddItemToObject(deviceInfo, "BusType", cJSON_CreateString("true"));
                cJSON_AddItemToObject(deviceInfo, "Spots4Bytes", cJSON_CreateString("false"));
                cJSON_AddItemToObject(deviceInfo, "DeviceID", cJSON_CreateNumber(1));
                cJSON_AddItemToObject(deviceInfo, "SampleSource", cJSON_CreateString("true"));
                cJSON_AddItemToObject(deviceInfo, "ControllerBoxType", cJSON_CreateNumber(70400));
                cJSON_AddItemToObject(deviceInfo, "DisplayRate", cJSON_CreateNumber(1000));
                cJSON_AddItemToObject(deviceInfo, "SamplingRate", cJSON_CreateNumber(1000));
                cJSON_AddItemToObject(deviceInfo, "MainBoardType", cJSON_CreateNumber(0));
                cJSON_AddItemToObject(deviceInfo, "MainBoardSoftVersion", cJSON_CreateString("1.01"));
                cJSON_AddItemToObject(deviceInfo, "DutyCollectEnable", cJSON_CreateString("false"));
//                cJSON_AddItemToObject(deviceInfo, "IOMonitorEnable", cJSON_CreateString("false"));

char *updated_json_str = cJSON_Print(root);
                if (!updated_json_str)
                {
                    rt_kprintf("Failed to print JSON\n");
                    cJSON_Delete(root);
                    return;
                }
                if (client.isconnected)
                {
                    rt_kprintf("mqtt_publish %d:%s:\n", pub_count, buffer);
                    print_long_string(updated_json_str);
//                    rt_kprintf("%s\n", updated_json_str);
                    if (!paho_mqtt_publish(&client, QOS1, buffer, updated_json_str))
                    {
                        ++pub_count;
                    }
                }
                else
                {
                    rt_kprintf("client.is not connected\n");
                }

if (updated_json_str)
                {
                    rt_free(updated_json_str);
                    updated_json_str = RT_NULL;
                }
                cJSON_Delete(root);
            }
                break;
            default:
                break;
            }
        }
    }
}
```