MessagePack 学习笔记

2021-08-03 22:54:45

封装和解析类似json的 key-value 示例

{"ID" = 333,"name"="zds","3333"="ende"}

    msgpack::sbuffer sBuf;
    msgpack::packer<msgpack::sbuffer> pker(&sBuf);

    pker.pack_map(3);
    pker.pack(std::string("ID"));
    pker.pack(333);
    pker.pack(std::string("name"));
    pker.pack(std::string("zds"));
    pker.pack(std::string("333"));
    pker.pack(std::string("ende"));


    //unserilized
    msgpack::unpacked unpack;
    msgpack::unpack(unpack, sBuf.data(), sBuf.size());

    msgpack::object obj = unpack.get();
    std::cout << obj << std::endl;


    if (obj.type == msgpack::type::ARRAY)
        std::cout << "是array" << std::endl;
    else if (obj.type == msgpack::type::MAP)
        std::cout << "是map" << std::endl;




    if(obj.via.map.size > 0)
    {
        auto pkv = obj.via.map.ptr;
        auto pkv_end = obj.via.map.ptr + obj.via.map.size;

        do 
        {
            auto key = pkv->key;
            auto val = pkv->val;
            std::cout << "key:" << key << " value:" << val << std::endl;
            ++pkv;
        } while (pkv < pkv_end);
    }

解析Socket示例

各类数据结构:

msgpack::object 他是一个引用，拷贝他的代价少，因为他是浅拷贝
msgpack::object_handle 他管理了一个对象的生命周期。他如果释放了，所有从他生成的object都是无效的引用

解析Socket示例

下列代码解析socke收包数据

unpacker.reserve_buffer 分配要收的数据的内存字节数
unpacker..buffer() 返回数据地址
unpacker.buffer_consumed() 设置实际收到的数据
unpacker.next(object_handle& oh) 循环解析数据

int main() {
    boost::asio::io_service ios;
    std::uint16_t const port = 12345;

    // Server
    std::size_t const window_size = 10;
    boost::asio::ip::tcp::acceptor ac(ios, boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), port));
    boost::asio::ip::tcp::socket ss(ios);
    std::function<void()> do_accept;
    std::function<void()> do_async_read_some;

    msgpack::unpacker unp;

    do_accept = [&] {
        ac.async_accept(
            ss,
            [&]
            (boost::system::error_code const& e) {
                if (e) {
                    std::cout << __LINE__ << ":" << e.message() << std::endl;
                    return;
                }
                unp.reserve_buffer(window_size);
                do_async_read_some = [&] {
                    ss.async_read_some(
                        boost::asio::buffer(unp.buffer(), window_size),
                        [&](boost::system::error_code const& e, std::size_t bytes_transferred) {
                            if (e) {
                                std::cout << __LINE__ << ":" << e.message() << std::endl;
                                return;
                            }
                            std::cout << bytes_transferred << " bytes read." << std::endl;
                            unp.buffer_consumed(bytes_transferred);
                            msgpack::object_handle oh;
                            while (unp.next(oh)) {
                                std::cout << oh.get() << std::endl;
                                // In order to finish the program,
                                // return if one complete msgpack is processed.
                                // In actual server, don't return here.
                                return;
                            }
                            do_async_read_some();
                        }
                    );
                };
                do_async_read_some();
            }
        );
    };
    do_accept();

    // Client
    auto host = "localhost";
    boost::asio::ip::tcp::resolver r(ios);
    boost::asio::ip::tcp::resolver::query q(host, boost::lexical_cast<std::string>(port));
    auto it = r.resolve(q);
    boost::asio::ip::tcp::socket cs(ios);
    boost::asio::async_connect(
        cs,

        it,
        [&]
        (boost::system::error_code const& e, boost::asio::ip::tcp::resolver::iterator) {
            if (e) {
                std::cout << __LINE__ << ":" << e.message() << std::endl;
                return;
            }
            std::cout << __LINE__ << ":client connected" << std::endl;
            msgpack::sbuffer sb;
            msgpack::pack(sb, std::make_tuple(42, false, "hello world", 12.3456));
            write(cs, boost::asio::buffer(sb.data(), sb.size()));
        }
    );

    // Start
    ios.run();
}

详解：

msgpack controls a buffer

msgpack provides a buffer management functionality named msgpack::unpacker. msgpack::unpacker is sutable for the following motivations:

msgpack data is chopped, and the client doesn't know when it will complete. This is a typical situation when you develop streaming applications.
You want to minimize copy opperations without careful memory management.

Here is the basic (not all) interface of msgpack::unpacker:

#ifndef MSGPACK_UNPACKER_INIT_BUFFER_SIZE
#define MSGPACK_UNPACKER_INIT_BUFFER_SIZE (64*1024)
#endif

#ifndef MSGPACK_UNPACKER_RESERVE_SIZE
#define MSGPACK_UNPACKER_RESERVE_SIZE (32*1024)
#endif

class unpacker {
public:
    unpacker(unpack_reference_func f = &unpacker::default_reference_func,
             void* user_data = nullptr,
             std::size_t init_buffer_size = MSGPACK_UNPACKER_INIT_BUFFER_SIZE,
             unpack_limit const& limit = unpack_limit());
    void reserve_buffer(std::size_t size = MSGPACK_UNPACKER_RESERVE_SIZE);
    char* buffer();
    void buffer_consumed(std::size_t size);
    bool next(unpacked& result);
};

Here is a basic pattern using msgpack::unpacker:

// The size may decided by receive performance, transmit layer's protocol and so on.
std::size_t const try_read_size = 100;

msgpack::unpacker unp;

// Message receive loop
while (/* block until input becomes readable */) {
    unp.reserve_buffer(try_read_size);
    // unp has at least try_read_size buffer on this point.

    // input is a kind of I/O library object.
    // read message to msgpack::unpacker's internal buffer directly.
    std::size_t actual_read_size = input.readsome(unp.buffer(), try_read_size);

    // tell msgpack::unpacker actual consumed size.
    unp.buffer_consumed(actual_read_size);

    msgpack::unpacked result;
    // Message pack data loop
    while(unp.next(result)) {
        msgpack::object obj(result.get());
        // Use obj
    }
    // All complete msgpack message is proccessed at this point,
    // then continue to read addtional message.
}

msgpack::unpacker::next() returns true if one complete msgpack messege is proccessed. If msgpack message is correct but insufficient, it returns false. However, parsing proccess is proceeded and the context information is preserved in the msgpack::unpacker. It helps leveling the load of parse.

When msgpack message contains binary data, string data, or ext data, they are not copied but referenced from msgpack::object by default. See the following implementation:

inline bool unpacker::default_reference_func(type::object_type type, uint64_t len, void*)
{
    return true;
}

You can also customize unpack_reference_func. Even if you use references, you don't need to control buffer's lifetime. The buffers' lifetime is controled by msgpack using msgpack::zone's finalizer_array and msgpack::unpacker's reference counting mechanism.

So, in most cases, the default behavior is enough. If you want to control the peak of memory consumption when receiving msgpack data patterns are predictable, customizing unpack_reference_func might be useful.

You can get a reference information from msgpack::unpacker::next() using the following function:

    bool next(unpacked& result, bool& referenced);

However, mostly you don't need to use that version of next() because referenced memories are managed by unpacker.

码农公寓

msgpack controls a buffer

相关文章