MessagePack简介与在Python中使用msgpack

什么是MessagePack？

It’s like JSON. but fast and small。
MessagePack is an efficient binary serialization format. It lets you exchange data among multiple languages like JSON. But it’s faster and smaller. Small integers are encoded into a single byte, and typical short strings require only one extra byte in addition to the strings themselves.

这是官网对于MessagePack的描述，可见这是一种类似JSON格式，但是比JSON更快更小的高效二进制序列化数据传输格式。它可以像JSON一样在多种语言之间交换数据。
小的整数被编码成单个字节，典型的短字符串除了字符串本身之外只需要一个额外的字节进行描述。

这是官方的示意图，同样的数据，使用JSON格式占用27bytes，使用MessagePack只占用18bytes。从图中可以看出，MessagePack对数据进行了重新编码和压缩，如，去掉了花括号、冒号、引号等符号，改用对应编码来描述数据的结构。如82表示这是一个包含两个元素的map，A7中A表示string，7表示长度，后面则是字符串数据内容，C3表示布尔值true等。同时使用二进制传输，
这样就大大减少了无用字符的数量，压缩了数据所占用的空间，在网络传输等情况下提供更快的速度。

对于MessagePack压缩原理感兴趣可以阅读：https://github.com/msgpack/msgpack/blob/master/spec.md

MessagePack支持的语言：

现在MessagePack支持超过50种编程语言和应用环境，可以在官网查看支持的实现列表。

单就对于Python的支持就提供了msgpack, vsergeev, lekma, jakm, polyglotted, aviramha等多个库以供使用。

Redis、fluentd、Celery等软件也提供对MessagePack格式的支持。

下面我们介绍MessagePack官方提供的Python库：msgpack。

msgpack python库的使用：

项目地址：https://github.com/msgpack/msgpack-python

提示：msgpack从1.0版本开始不再支持python2，如果想在python2中使用msgpack可以使用msgpack.fallback库，不过该库是纯python实现，相比于使用 C 语言做 binding 的msgpack 库会慢很多。

进入正题，如何在Python3中使用msgapck进行消息传输？

安装：pip install msgpack
使用：

一次性打包和解包：
使用packb方法打包，使用unpackb方法解包。同时msgpack还提供了dumps和loads方法作为这两个方法的别名，以兼容json库和pickle库。

import msgpackdata = {'name': 'Jack', 'age': 20}# pack
pack_data = msgpack.packb(data)
print(pack_data)
pack_data = msgpack.dumps(data)
print(pack_data)# unpack
unpack_data = msgpack.unpackb(pack_data)
print(unpack_data)
unpack_data = msgpack.loads(pack_data)
print(unpack_data)

执行结果：

b'\x82\xa4name\xa4Jack\xa3age\x14'
b'\x82\xa4name\xa4Jack\xa3age\x14'
{'name': 'Jack', 'age': 20}
{'name': 'Jack', 'age': 20}

当数据装包含列表时，解包时默认解包为list，如果设置use_list=False，则解包为元组。
List是python默认的序列类型，但是元组比列表要“轻”，当性能对于程序很重要时，可以在解包时使用use_list=False。

import msgpackdata = [1, 2, 3, 4]pack_data = msgpack.packb(data)
print(pack_data)unpack_data = msgpack.unpackb(pack_data)
print(unpack_data)unpack_data = msgpack.unpackb(pack_data, use_list=False)
print(unpack_data)

执行结果：

b'\x94\x01\x02\x03\x04'
[1, 2, 3, 4]
(1, 2, 3, 4)

对流进行解包Streaming unpacking：
Unpacker是一个streaming unpacker。它从一个流中解包多个对象(或者从其提要方法提供的字节中解包)。

import msgpack
from io import BytesIO# write msgpack to stream
buf = BytesIO()
for i in range(10):buf.write(msgpack.packb(i))buf.seek(0)# read and unpack msgpack from stream
unpacker = msgpack.Unpacker(buf)
for unpacked in unpacker:print(unpacked, end='\t')

执行结果：

0  1   2   3   4   5   6   7   8   9

同样, 可以使用Unpacker从文件流中读取和解包数据：

import msgpack# write msgpack to file
with open('data', 'wb') as f:for i in range(10):f.write(msgpack.packb(i))# read and unpack msgpack from file
with open('data', 'rb') as f:unpacker = msgpack.Unpacker(f)for unpacked in unpacker:print(unpacked, end='\t')

执行结果：

0  1   2   3   4   5   6   7   8   9

对自定义数据类型进行打包/解包：
msgpack也可以对自定义的数据类型解析处理，需要提供对应的处理方法作为参数传给packb和unpackb。这里以datetime.datetime为例：

import msgpack
import datetimedata = {'id': 1, 'created': datetime.datetime.now()}def encode_datetime(obj):"""对datetime数据进行编码"""if isinstance(obj, datetime.datetime):return {'__datetime__': True, 'as_str': obj.strftime('%Y%m%dT%H%M%S.%f')}return objdef decode_datetime(obj):"""对datetime数据进行解码"""if '__datetime__' in obj:obj = datetime.datetime.strptime(obj['as_str'], '%Y%m%dT%H%M%S.%f')return objpacked_data = msgpack.packb(data, default=encode_datetime)
unpacked_data = msgpack.unpackb(packed_data, object_hook=decode_datetime)
print(packed_data)
print(unpacked_data)
print(data == unpacked_data)

执行结果：

b'\x82\xa2id\x01\xa7created\x82\xac__datetime__\xc3\xa6as_str\xb620220607T185631.676527'
{'id': 1, 'created': datetime.datetime(2022, 6, 7, 18, 56, 31, 676527)}
True

其中，Unpacker的object_hook回调接收字典对象，object_pairs_hook回调可以用来接收键值对的列表。

扩展类型：
还可以使用 ext类型对自定义数据类型进行打包和解包。

import msgpack
import arraydef default(obj):if isinstance(obj, array.array) and obj.typecode == 'd':return msgpack.ExtType(42, obj.tobytes())raise TypeError('Unknown Type: %r' % obj)def ext_hook(code, data):if code == 42:a = array.array('d')a.frombytes(data)return a raise msgpack.ExtType(code, data)data = array.array('d', [1.2, 3.4, 5.6])
packed = msgpack.packb(data, default=default)
unpacked = msgpack.unpackb(packed, ext_hook=ext_hook)
print(packed)
print(unpacked)
print(data == unpacked)

执行结果：

b'\xc7\x18*333333\xf3?333333\x0b@ffffff\x16@'
array('d', [1.2, 3.4, 5.6])
True

专业解包控制：
作为对迭代的替代，Unpacker 提供了 unpack、skip、read_array_header 和 read_map_header 方法。前两个从流中读取整个消息，分别对结果进行反序列化和返回，或者忽略它。后两个方法返回即将到来的容器中的元素数量，以便数组中的每个元素或map中的键值对可以单独解包或跳过。

注意：
msgpack不强制区分 bytes 和 str，二者都可以进行打包和解包。
安全问题：
如果要解包来自于不信任的数据源信息，msgpack提供了两个安全选项：
- max_buffer_size (默认: 100*1024*1024) 限制内部缓冲区大小。它也用于限制预分配的列表大小。
- strict_map_key (默认: True) 将map的键类型限制为bytes和 str。虽然 msgpack 规范没有限制映射键的类型，但是存在 hashdos 的风险。如果需要支持其他类型的映射键，设置 strict_map_key = False。
性能提醒：
CPython的GC（垃圾回收）会在增加分配对象时开始执行。这意味着解包时可能会造成无用的GC。你可以在解包大的消息时设置gc.disable()。