Msgpack反序列化报错msgpack::v1::insufficient_bytes求助

发布端通过ZeroMQ库发送经Msgpack打包的字符串消息，运行状态正常，可正常发送消息。订阅端接收数据时编译无报错，但运行至代码行msgpack::unpack(msg, static_cast<const char*>(got.data()), got.size());时程序中断，抛出错误msgpack::v1::insufficient_bytes。

发布端代码 #

#include <future>
#include <iostream>
#include <string>
#include <thread>

#include "zmq.hpp"
#include "zmq_addon.hpp"
#include<msgpack.hpp>

int main()
{
    zmq::context_t ctx(1);
    zmq::socket_t publisher(ctx, zmq::socket_type::pub);

    std::stringstream ss;
    
    publisher.bind("tcp://127.0.0.1:5041");
    
    std::this_thread::sleep_for(std::chrono::milliseconds(500));
    std::string msg_str = "Test";

    msgpack::sbuffer sbuf;

    
    msgpack::pack(sbuf, msg_str);

    zmq::message_t msg_send(sbuf.size());
    memcpy(msg_send.data(), sbuf.data(), sizeof(sbuf));
    
    
    while (true)
    {

        publisher.send(msg_send, zmq::send_flags::none);
        std::cout << "Mitteilung ist online: " << std::endl;
        std::this_thread::sleep_for(std::chrono::milliseconds(500));
        
    }

}

订阅端代码 #

#include <iostream>
#include<zmq.h>
#include<zmq.hpp>
#include<msgpack.hpp>
#include<vector>
#include<string>
#include <thread>

int main()
{
    zmq::context_t ctx(1);
    zmq::socket_t subscriber(ctx, zmq::socket_type::sub);

    subscriber.connect("tcp://127.0.0.1:5041");

    subscriber.set(zmq::sockopt::subscribe, "");
    
    while (1)
    {
        std::string um_str_;
        msgpack::unpacked msg;
        zmq::message_t got;
       
        subscriber.recv(got);
        msgpack::object obj = msg.get();

        std::cout << "Hier" << std::endl;

        msgpack::unpack(msg, static_cast<const char*>(got.data()), got.size());
       // obj.convert(um_str_);

    }

}

1. 发布端的数据拷贝错误 #

发布端中memcpy(msg_send.data(), sbuf.data(), sizeof(sbuf));存在致命错误：sizeof(sbuf)获取的是msgpack::sbuffer对象本身的内存大小，而非它实际存储的Msgpack打包数据的字节长度。这导致发送的消息长度远小于真实的打包数据长度，订阅端收到不完整的数据后，解包时就会触发insufficient_bytes错误。

修正方式：将sizeof(sbuf)替换为sbuf.size()，确保拷贝完整的打包数据：

memcpy(msg_send.data(), sbuf.data(), sbuf.size());

2. 订阅端的代码执行顺序错误 #

订阅端中msgpack::object obj = msg.get();在msgpack::unpack之前调用，此时msgpack::unpacked msg还未被初始化，调用msg.get()会导致未定义行为，同时打乱了解包的正常流程。

修正方式：调整代码顺序，先执行解包操作，再获取msgpack::object并转换数据：
修正后的订阅端循环部分代码：

while (1)
{
    std::string um_str_;
    msgpack::unpacked msg;
    zmq::message_t got;
   
    subscriber.recv(got);
    std::cout << "Hier" << std::endl;

    // 先执行解包
    msgpack::unpack(msg, static_cast<const char*>(got.data()), got.size());
    msgpack::object obj = msg.get();
    // 转换数据
    obj.convert(um_str_);
    std::cout << "解析到消息：" << um_str_ << std::endl;
}

发布端完整代码 #

#include <future>
#include <iostream>
#include <string>
#include <thread>

#include "zmq.hpp"
#include "zmq_addon.hpp"
#include<msgpack.hpp>

int main()
{
    zmq::context_t ctx(1);
    zmq::socket_t publisher(ctx, zmq::socket_type::pub);

    std::stringstream ss;
    
    publisher.bind("tcp://127.0.0.1:5041");
    
    std::this_thread::sleep_for(std::chrono::milliseconds(500));
    std::string msg_str = "Test";

    msgpack::sbuffer sbuf;
    msgpack::pack(sbuf, msg_str);

    zmq::message_t msg_send(sbuf.size());
    memcpy(msg_send.data(), sbuf.data(), sbuf.size());
    
    while (true)
    {
        publisher.send(msg_send, zmq::send_flags::none);
        std::cout << "消息已发送" << std::endl;
        std::this_thread::sleep_for(std::chrono::milliseconds(500));
    }
}

订阅端完整代码 #

#include <iostream>
#include<zmq.h>
#include<zmq.hpp>
#include<msgpack.hpp>
#include<vector>
#include<string>
#include <thread>

int main()
{
    zmq::context_t ctx(1);
    zmq::socket_t subscriber(ctx, zmq::socket_type::sub);

    subscriber.connect("tcp://127.0.0.1:5041");
    subscriber.set(zmq::sockopt::subscribe, "");
    
    while (1)
    {
        std::string um_str_;
        msgpack::unpacked msg;
        zmq::message_t got;
       
        subscriber.recv(got);
        std::cout << "收到原始数据" << std::endl;

        msgpack::unpack(msg, static_cast<const char*>(got.data()), got.size());
        msgpack::object obj = msg.get();
        obj.convert(um_str_);

        std::cout << "解析后消息：" << um_str_ << std::endl;
    }
}

内容的提问来源于stack exchange，提问作者Trixxx128