MessagePack 学习笔记-阿里云开发者社区

开发者社区> 大数据> 正文

MessagePack 学习笔记

简介: 封装和解析类似json的  key-value 示例 {"ID" = 333,"name"="zds","3333"="ende"} msgpack::sbuffer sBuf; msgpack::packer pker(&sBuf); pker.

封装和解析类似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.

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:
大数据
使用钉钉扫一扫加入圈子
+ 订阅

大数据计算实践乐园,近距离学习前沿技术

其他文章