如何将librdkafka有效负载转换为json获取参数值？

#include <iostream>
#include <json/json.h>
#include <librdkafka/rdkafka.h>

void message_callback(rd_kafka_t* rk, const rd_kafka_message_t* rkmessage, void* opaque) {
    // 获取有效负载的原始数据
    const char* payload = (const char*)rkmessage->payload;
    size_t len = rkmessage->len;

    // 将原始数据转换为JSON
    Json::Value root;
    Json::Reader reader;
    bool parsingSuccessful = reader.parse(payload, payload + len, root);
    if (!parsingSuccessful) {
        std::cerr << "Failed to parse payload as JSON: " << reader.getFormattedErrorMessages() << std::endl;
        return;
    }

    // 获取参数值
    std::string paramValue = root["param"].asString();
    std::cout << "Param value: " << paramValue << std::endl;
}

int main() {
    // 创建Kafka消费者
    rd_kafka_t* rk;
    rd_kafka_conf_t* conf;
    char errstr[512];

    conf = rd_kafka_conf_new();
    rd_kafka_conf_set(conf, "bootstrap.servers", "your_kafka_bootstrap_servers", errstr, sizeof(errstr));

    rk = rd_kafka_new(RD_KAFKA_CONSUMER, conf, errstr, sizeof(errstr));
    if (!rk) {
        std::cerr << "Failed to create Kafka consumer: " << errstr << std::endl;
        return 1;
    }

    // 订阅主题
    rd_kafka_topic_partition_list_t* topics;
    topics = rd_kafka_topic_partition_list_new();
    rd_kafka_topic_partition_list_add(topics, "your_topic", RD_KAFKA_PARTITION_UA);

    rd_kafka_subscribe(rk, topics);

    // 接收消息
    while (true) {
        rd_kafka_message_t* rkmessage;
        rkmessage = rd_kafka_consumer_poll(rk, 1000);
        if (!rkmessage) {
            continue;
        }

        // 处理消息
        message_callback(rk, rkmessage, nullptr);

        rd_kafka_message_destroy(rkmessage);
    }

    // 清理资源
    rd_kafka_topic_partition_list_destroy(topics);
    rd_kafka_destroy(rk);

    return 0;
}