高效的序列化/反序列化数据方式 Protobuf

高效的序列化/反序列化数据方式 Protobuf

github地址

目录

  • protocolBuffers 序列化
    • Int32
    • String
    • Map
    • slice
    • 序列化小结
  • protocolBuffers 反序列化
    • Int32
    • String
    • Map
    • slice
    • 序列化小结
  • 序列化/反序列化性能
  • 最后

protocolBuffers序列化

上篇文章中其实已经讲过了 encode 的过程,这篇文章以 golang 为例,从代码实现的层面讲讲序列化和反序列化的过程。

举个 go 使用 protobuf 进行数据序列化和反序列化的例子,本篇文章从这个例子开始。

先新建一个 examplemessage

syntax = "proto2";
package example;enum FOO { X = 17; };message Test {required string label = 1;optional int32 type = 2 [default=77];repeated int64 reps = 3;optional group OptionalGroup = 4 {required string RequiredField = 5;}
}

利用 protoc-gen-go 生成对应的 get/set 方法。代码中就可以用生成的代码进行序列化和反序列化了。

package mainimport ("log""github.com/golang/protobuf/proto""path/to/example"
)func main() {test := &example.Test {Label: proto.String("hello"),Type:  proto.Int32(17),Reps:  []int64{1, 2, 3},Optionalgroup: &example.Test_OptionalGroup {RequiredField: proto.String("good bye"),},}data, err := proto.Marshal(test)if err != nil {log.Fatal("marshaling error: ", err)}newTest := &example.Test{}err = proto.Unmarshal(data, newTest)if err != nil {log.Fatal("unmarshaling error: ", err)}// Now test and newTest contain the same data.if test.GetLabel() != newTest.GetLabel() {log.Fatalf("data mismatch %q != %q", test.GetLabel(), newTest.GetLabel())}// etc.
}

上面代码中 proto.Marshal() 是序列化过程。proto.Unmarshal() 是反序列化过程。这一章节先看看序列化过程的实现,下一章节再分析反序列化过程的实现。

// Marshal takes the protocol buffer
// and encodes it into the wire format, returning the data.
func Marshal(pb Message) ([]byte, error) {// Can the object marshal itself?if m, ok := pb.(Marshaler); ok {return m.Marshal()}p := NewBuffer(nil)err := p.Marshal(pb)if p.buf == nil && err == nil {// Return a non-nil slice on success.return []byte{}, nil}return p.buf, err
}

序列化函数一进来,会先调用 message 对象自身的实现的序列化方法。

// Marshaler is the interface representing objects that can marshal themselves.
type Marshaler interface {Marshal() ([]byte, error)
}

Marshaler 是一个 interface ,这个接口是专门留给对象自定义序列化的。如果有实现,就 return 自己实现的方法。如果没有,接下来就进行默认序列化方式。

p := NewBuffer(nil)
err := p.Marshal(pb)
if p.buf == nil && err == nil {// Return a non-nil slice on success.return []byte{}, nil
}

新建一个 Buffer ,调用 BufferMarshal() 方法。message 经过序列化以后,数据流会放到 Bufferbuf 字节流中。序列化最终返回 buf 字节流即可。

type Buffer struct {buf   []byte // encode/decode byte streamindex int    // read point// pools of basic types to amortize allocation.bools   []booluint32s []uint32uint64s []uint64// extra pools, only used with pointer_reflect.goint32s   []int32int64s   []int64float32s []float32float64s []float64
}

Buffer 的数据结构如上, Buffer 是用于序列化和反序列化 protocol buffers 的缓冲区管理器。它可以在调用的时候重用以减少内存使用量。内部维护了 7 个 pool ,3 个基础数据类型的 pool ,4 个只能被 pointer_reflect 使用的 pool

func (p *Buffer) Marshal(pb Message) error {// Can the object marshal itself?if m, ok := pb.(Marshaler); ok {data, err := m.Marshal()p.buf = append(p.buf, data...)return err}t, base, err := getbase(pb)// 异常处理if structPointer_IsNil(base) {return ErrNil}if err == nil {err = p.enc_struct(GetProperties(t.Elem()), base)}// 用来统计 Encode 次数的if collectStats {(stats).Encode++ // Parens are to work around a goimports bug.}// maxMarshalSize = 1<<31 - 1,这个值是 protobuf 可以 encoded 的最大值。if len(p.buf) > maxMarshalSize {return ErrTooLarge}return err
}

BufferMarshal() 方法依旧先调用一下对象是否实现了 Marshal() 接口,如果实现了,还是让它自己序列化,序列化之后的二进制数据流加入到 buf 数据流中。

func getbase(pb Message) (t reflect.Type, b structPointer, err error) {if pb == nil {err = ErrNilreturn}// get the reflect type of the pointer to the struct.t = reflect.TypeOf(pb)// get the address of the struct.value := reflect.ValueOf(pb)b = toStructPointer(value)return
}

getbase 方法通过 reflect 方法拿到了 message 的类型和对应 value 的结构体指针。拿到结构体指针先做异常处理。

所以序列化最核心的代码其实就一句,p.enc_struct(GetProperties(t.Elem()), base)

// Encode a struct.
func (o *Buffer) enc_struct(prop *StructProperties, base structPointer) error {var state errorState// Encode fields in tag order so that decoders may use optimizations// that depend on the ordering.// https://developers.google.com/protocol-buffers/docs/encoding#orderfor _, i := range prop.order {p := prop.Prop[i]if p.enc != nil {err := p.enc(o, p, base)if err != nil {if err == ErrNil {if p.Required && state.err == nil {state.err = &RequiredNotSetError{p.Name}}} else if err == errRepeatedHasNil {// Give more context to nil values in repeated fields.return errors.New("repeated field " + p.OrigName + " has nil element")} else if !state.shouldContinue(err, p) {return err}}if len(o.buf) > maxMarshalSize {return ErrTooLarge}}}// Do oneof fields.if prop.oneofMarshaler != nil {m := structPointer_Interface(base, prop.stype).(Message)if err := prop.oneofMarshaler(m, o); err == ErrNil {return errOneofHasNil} else if err != nil {return err}}// Add unrecognized fields at the end.if prop.unrecField.IsValid() {v := *structPointer_Bytes(base, prop.unrecField)if len(o.buf)+len(v) > maxMarshalSize {return ErrTooLarge}if len(v) > 0 {o.buf = append(o.buf, v...)}}return state.err
}

上面代码中可以看到,除去 oneof fieldsunrecognized fields 是单独最后处理的,其他类型都是调用的 p.enc(o, p, base) 进行序列化的。

Properties 的数据结构定义如下:

type Properties struct {Name     string // name of the field, for error messagesOrigName string // original name before protocol compiler (always set)JSONName string // name to use for JSON; determined by protocWire     stringWireType intTag      intRequired boolOptional boolRepeated boolPacked   bool   // relevant for repeated primitives onlyEnum     string // set for enum types onlyproto3   bool   // whether this is known to be a proto3 field; set for []byte onlyoneof    bool   // whether this is a oneof fieldDefault     string // default valueHasDefault  bool   // whether an explicit default was providedCustomType  stringStdTime     boolStdDuration boolenc           encodervalEnc        valueEncoder // set for bool and numeric types onlyfield         fieldtagcode       []byte // encoding of EncodeVarint((Tag<<3)|WireType)tagbuf        [8]bytestype         reflect.Type      // set for struct types onlysstype        reflect.Type      // set for slices of structs types onlyctype         reflect.Type      // set for custom types onlysprop         * StructProperties // set for struct types onlyisMarshaler   boolisUnmarshaler boolmtype    reflect.Type // set for map types onlymkeyprop * Properties  // set for map types onlymvalprop * Properties  // set for map types onlysize    sizervalSize valueSizer // set for bool and numeric types onlydec    decodervalDec valueDecoder // set for bool and numeric types only// If this is a packable field, this will be the decoder for the packed version of the field.packedDec decoder
}

Properties 这个结构体中,定义了名为 encencoder 和名为 decdecoder

encoderdecoder 函数定义是完全一样的。

type encoder func(p *Buffer, prop *Properties, base structPointer) error
type decoder func(p *Buffer, prop *Properties, base structPointer) error

encoderdecoder 函数初始化是在 Properties 中:

// Initialize the fields for encoding and decoding.
func (p *Properties) setEncAndDec(typ reflect.Type, f *reflect.StructField, lockGetProp bool) {// 下面代码有删减,类似的部分省略了// proto3 scalar typescase reflect.Int32:if p.proto3 {p.enc = (*Buffer).enc_proto3_int32p.dec = (*Buffer).dec_proto3_int32p.size = size_proto3_int32} else {p.enc = (*Buffer).enc_ref_int32p.dec = (*Buffer).dec_proto3_int32p.size = size_ref_int32}case reflect.Uint32:if p.proto3 {p.enc = (*Buffer).enc_proto3_uint32p.dec = (*Buffer).dec_proto3_int32 // can reusep.size = size_proto3_uint32} else {p.enc = (*Buffer).enc_ref_uint32p.dec = (*Buffer).dec_proto3_int32 // can reusep.size = size_ref_uint32}case reflect.Float32:if p.proto3 {p.enc = (*Buffer).enc_proto3_uint32 // can just treat them as bitsp.dec = (*Buffer).dec_proto3_int32p.size = size_proto3_uint32} else {p.enc = (*Buffer).enc_ref_uint32 // can just treat them as bitsp.dec = (*Buffer).dec_proto3_int32p.size = size_ref_uint32}case reflect.String:if p.proto3 {p.enc = (*Buffer).enc_proto3_stringp.dec = (*Buffer).dec_proto3_stringp.size = size_proto3_string} else {p.enc = (*Buffer).enc_ref_stringp.dec = (*Buffer).dec_proto3_stringp.size = size_ref_string}case reflect.Slice:switch t2 := t1.Elem(); t2.Kind() {default:logNoSliceEnc(t1, t2)breakcase reflect.Int32:if p.Packed {p.enc = (*Buffer).enc_slice_packed_int32p.size = size_slice_packed_int32} else {p.enc = (*Buffer).enc_slice_int32p.size = size_slice_int32}p.dec = (*Buffer).dec_slice_int32p.packedDec = (*Buffer).dec_slice_packed_int32default:logNoSliceEnc(t1, t2)break}}case reflect.Map:p.enc = (*Buffer).enc_new_mapp.dec = (*Buffer).dec_new_mapp.size = size_new_mapp.mtype = t1p.mkeyprop = &Properties{}p.mkeyprop.init(reflect.PtrTo(p.mtype.Key()), "Key", f.Tag.Get("protobuf_key"), nil, lockGetProp)p.mvalprop = &Properties{}vtype := p.mtype.Elem()if vtype.Kind() != reflect.Ptr && vtype.Kind() != reflect.Slice {// The value type is not a message (*T) or bytes ([]byte),// so we need encoders for the pointer to this type.vtype = reflect.PtrTo(vtype)}p.mvalprop.CustomType = p.CustomTypep.mvalprop.StdDuration = p.StdDurationp.mvalprop.StdTime = p.StdTimep.mvalprop.init(vtype, "Value", f.Tag.Get("protobuf_val"), nil, lockGetProp)}p.setTag(lockGetProp)
}

上面代码中,分别把各个类型都进行 switch - case 枚举,每种情况都设置对应的 encode 编码器,decode 解码器,size 大小。proto2proto3 有区别的地方也分成2种不同的情况进行处理。

有以下几种类型,reflect.Boolreflect.Int32reflect.Uint32reflect.Int64reflect.Uint64reflect.Float32reflect.Float64reflect.Stringreflect.Structreflect.Ptr、reflect.Slicereflect.Map 共 12 种大的分类。

下面主要挑 3 类,Int32StringMap 代码实现进行分析。

Int32

func (o *Buffer) enc_proto3_int32(p *Properties, base structPointer) error {v := structPointer_Word32Val(base, p.field)x := int32(word32Val_Get(v)) // permit sign extension to use full 64-bit rangeif x == 0 {return ErrNil}o.buf = append(o.buf, p.tagcode...)p.valEnc(o, uint64(x))return nil
}

处理 Int32 代码比较简单,先把 tagcode 放进 buf 二进制数据流缓冲区,接着序列化 Int32 ,序列化以后紧接着 tagcode 后面放进缓冲区。

// EncodeVarint writes a varint-encoded integer to the Buffer.
// This is the format for the
// int32, int64, uint32, uint64, bool, and enum
// protocol buffer types.
func (p *Buffer) EncodeVarint(x uint64) error {for x >= 1<<7 {p.buf = append(p.buf, uint8(x&0x7f|0x80))x >>= 7}p.buf = append(p.buf, uint8(x))return nil
}

Int32 的编码处理方法在上篇里面讲过,用的 Varint 处理方法。上面这个函数同样适用于处理 int32, int64, uint32, uint64, bool, enum

顺道也可以看看 sint32Fixed32 的具体代码实现。

// EncodeZigzag32 writes a zigzag-encoded 32-bit integer
// to the Buffer.
// This is the format used for the sint32 protocol buffer type.
func (p *Buffer) EncodeZigzag32(x uint64) error {// use signed number to get arithmetic right shift.return p.EncodeVarint(uint64((uint32(x) << 1) ^ uint32((int32(x) >> 31))))
}

针对有符号的 sint32 ,采取的是先 Zigzag ,然后在 Varint 的处理方式。

// EncodeFixed32 writes a 32-bit integer to the Buffer.
// This is the format for the
// fixed32, sfixed32, and float protocol buffer types.
func (p *Buffer) EncodeFixed32(x uint64) error {p.buf = append(p.buf,uint8(x),uint8(x>>8),uint8(x>>16),uint8(x>>24))return nil
}

对于 Fixed32 的处理,仅仅只是位移操作,并没有做什么压缩操作。

String

func (o *Buffer) enc_proto3_string(p *Properties, base structPointer) error {v := *structPointer_StringVal(base, p.field)if v == "" {return ErrNil}o.buf = append(o.buf, p.tagcode...)o.EncodeStringBytes(v)return nil
}

序列化字符串也分2步,先把 tagcode 放进去,然后再序列化数据。

// EncodeStringBytes writes an encoded string to the Buffer.
// This is the format used for the proto2 string type.
func (p *Buffer) EncodeStringBytes(s string) error {p.EncodeVarint(uint64(len(s)))p.buf = append(p.buf, s...)return nil
}

序列化字符串的时候,会先把字符串的长度通过编码 Varint 的方式,写到 buf 中。长度后面再紧跟着 string 。这也就是 tag - length - value 的实现。

Map

// Encode a map field.
func (o *Buffer) enc_new_map(p *Properties, base structPointer) error {var state errorState // XXX: or do we need to plumb this through?v := structPointer_NewAt(base, p.field, p.mtype).Elem() // map[K]Vif v.Len() == 0 {return nil}keycopy, valcopy, keybase, valbase := mapEncodeScratch(p.mtype)enc := func() error {if err := p.mkeyprop.enc(o, p.mkeyprop, keybase); err != nil {return err}if err := p.mvalprop.enc(o, p.mvalprop, valbase); err != nil && err != ErrNil {return err}return nil}// Don't sort map keys. It is not required by the spec, and C++ doesn't do it.for _, key := range v.MapKeys() {val := v.MapIndex(key)keycopy.Set(key)valcopy.Set(val)o.buf = append(o.buf, p.tagcode...)if err := o.enc_len_thing(enc, &state); err != nil {return err}}return nil
}

上述代码也可以序列化字典数组,例如:

map<key_type, value_type> map_field = N;

转换成对应的 repeated message 形式再进行序列化。

message MapFieldEntry {key_type key = 1;value_type value = 2;
}
repeated MapFieldEntry map_field = N;

map 序列化是针对每个 k-v ,都先放入 tagcode ,然后再序列化 k-v 。这里需要化未知长度的结构体的时候需要调用 enc_len_thing() 方法。

// Encode something, preceded by its encoded length (as a varint).
func (o *Buffer) enc_len_thing(enc func() error, state *errorState) error {iLen := len(o.buf)o.buf = append(o.buf, 0, 0, 0, 0) // reserve four bytes for lengthiMsg := len(o.buf)err := enc()if err != nil && !state.shouldContinue(err, nil) {return err}lMsg := len(o.buf) - iMsglLen := sizeVarint(uint64(lMsg))switch x := lLen - (iMsg - iLen); {case x > 0: // actual length is x bytes larger than the space we reserved// Move msg x bytes right.o.buf = append(o.buf, zeroes[:x]...)copy(o.buf[iMsg+x:], o.buf[iMsg:iMsg+lMsg])case x < 0: // actual length is x bytes smaller than the space we reserved// Move msg x bytes left.copy(o.buf[iMsg+x:], o.buf[iMsg:iMsg+lMsg])o.buf = o.buf[:len(o.buf)+x] // x is negative}// Encode the length in the reserved space.o.buf = o.buf[:iLen]o.EncodeVarint(uint64(lMsg))o.buf = o.buf[:len(o.buf)+lMsg]return state.err
}

enc_len_thing() 方法会先预存 4 个字节的长度空位。序列化以后算出长度。如果长度比 4 个字节还要长,则右移序列化的二进制数据,把长度填到 tagcode 和数据之间。如果长度小于 4 个字节,相应的要左移。

slice

最后再举一个数组的例子。以 []int32 为例。

// Encode a slice of int32s ([]int32) in packed format.
func (o *Buffer) enc_slice_packed_int32(p *Properties, base structPointer) error {s := structPointer_Word32Slice(base, p.field)l := s.Len()if l == 0 {return ErrNil}// TODO: Reuse a Buffer.buf := NewBuffer(nil)for i := 0; i < l; i++ {x := int32(s.Index(i)) // permit sign extension to use full 64-bit rangep.valEnc(buf, uint64(x))}o.buf = append(o.buf, p.tagcode...)o.EncodeVarint(uint64(len(buf.buf)))o.buf = append(o.buf, buf.buf...)return nil
}

序列化这个数组,分3步,先把 tagcode 放进去,然后再序列化整个数组的长度,最后把数组的每个数据都序列化放在后面。最后形成 tag - length - value - value - value 的形式。

上述就是 Protocol Buffer 序列化的过程。

序列化小结

Protocol Buffer 序列化采用 Varint、Zigzag 方法,压缩 int 型整数和带符号的整数。对浮点型数字不做压缩(这里可以进一步的压缩,Protocol Buffer 还有提升空间)。编码 .proto 文件,会对 optionrepeated 字段进行检查,若 optionalrepeated 字段没有被设置字段值,那么该字段在序列化时的数据中是完全不存在的,即不进行序列化(少编码一个字段)。

上面这两点做到了压缩数据,序列化工作量减少。

序列化的过程都是二进制的位移,速度非常快。数据都以 tag - length - value (或者 tag - value)的形式存在二进制数据流中。采用了 TLV 结构存储数据以后,也摆脱了 JSON 中的 {、}、; 、这些分隔符,没有这些分隔符也算是再一次减少了一部分数据。

这一点做到了序列化速度非常快。

回到顶部

protocolBuffers反序列化

反序列化的实现完全是序列化实现的逆过程。

func Unmarshal(buf []byte, pb Message) error {pb.Reset()return UnmarshalMerge(buf, pb)
}

在反序列化开始之前,先重置一下缓冲区。

func (p *Buffer) Reset() {p.buf = p.buf[0:0] // for reading/writingp.index = 0        // for reading
}

清空 buf 中的所有数据,并且重置 index

func UnmarshalMerge(buf []byte, pb Message) error {// If the object can unmarshal itself, let it.if u, ok := pb.(Unmarshaler); ok {return u.Unmarshal(buf)}return NewBuffer(buf).Unmarshal(pb)
}

反序列化数据的开始从上面这个函数开始,如果传进来的 message 的结果和 buf 结果不匹配,最终得到的结果是不可预知的。反序列化之前,同样会先调用一下对应自己身自定义的 Unmarshal() 方法。

type Unmarshaler interface {Unmarshal([]byte) error
}

Unmarshal() 是一个可以自己实现的接口。

UnmarshalMerge 中会调用 Unmarshal(pb Message) 方法。

func (p *Buffer) Unmarshal(pb Message) error {// If the object can unmarshal itself, let it.if u, ok := pb.(Unmarshaler); ok {err := u.Unmarshal(p.buf[p.index:])p.index = len(p.buf)return err}typ, base, err := getbase(pb)if err != nil {return err}err = p.unmarshalType(typ.Elem(), GetProperties(typ.Elem()), false, base)if collectStats {stats.Decode++}return err
}

Unmarshal(pb Message) 这个函数只有一个入参,和 proto.Unmarshal() 方法函数签名不同(前面的函数只有 1 个入参,后面的有 2 个入参)。两者的区别在于,1 个入参的函数实现里面并不会重置 buf 缓冲区,二个入参的会先重置 buf 缓冲区。

这两个函数最终都会调用 unmarshalType() 方法,这个函数是最终支持反序列化的函数。

func (o *Buffer) unmarshalType(st reflect.Type, prop *StructProperties, is_group bool, base structPointer) error {var state errorStaterequired, reqFields := prop.reqCount, uint64(0)var err errorfor err == nil && o.index < len(o.buf) {oi := o.indexvar u uint64u, err = o.DecodeVarint()if err != nil {break}wire := int(u & 0x7)// 下面代码有省略dec := p.dec// 中间代码有省略decErr := dec(o, p, base)if decErr != nil && !state.shouldContinue(decErr, p) {err = decErr}if err == nil && p.Required {// Successfully decoded a required field.if tag <= 64 {// use bitmap for fields 1-64 to catch field reuse.var mask uint64 = 1 << uint64(tag-1)if reqFields&mask == 0 {// new required fieldreqFields |= maskrequired--}} else {// This is imprecise. It can be fooled by a required field// with a tag > 64 that is encoded twice; that's very rare.// A fully correct implementation would require allocating// a data structure, which we would like to avoid.required--}}}if err == nil {if is_group {return io.ErrUnexpectedEOF}if state.err != nil {return state.err}if required > 0 {// Not enough information to determine the exact field. If we use extra// CPU, we could determine the field only if the missing required field// has a tag <= 64 and we check reqFields.return &RequiredNotSetError{"{Unknown}"}}}return err
}

unmarshalType() 函数比较长,里面处理的情况比较多,有 oneof,WireEndGroup 。真正处理反序列化的函数在 decErr := dec(o, p, base) 这一行。

dec 函数在 PropertiessetEncAndDec() 函数中进行了初始化。上面序列化的时候谈到过那个函数了,这里就不再赘述了。dec() 函数针对每个不同类型都有对应的反序列化函数。

同样的,接下来也举 4 个例子,看看反序列化的实际代码实现。

Int32

func (o *Buffer) dec_proto3_int32(p *Properties, base structPointer) error {u, err := p.valDec(o)if err != nil {return err}word32Val_Set(structPointer_Word32Val(base, p.field), uint32(u))return nil
}

反序列化 Int32 代码比较简单,原理是按照 encode 的逆过程,还原原来的数据。

func (p *Buffer) DecodeVarint() (x uint64, err error) {i := p.indexbuf := p.bufif i >= len(buf) {return 0, io.ErrUnexpectedEOF} else if buf[i] < 0x80 {p.index++return uint64(buf[i]), nil} else if len(buf)-i < 10 {return p.decodeVarintSlow()}var b uint64// we already checked the first bytex = uint64(buf[i]) - 0x80i++b = uint64(buf[i])i++x += b << 7if b&0x80 == 0 {goto done}x -= 0x80 << 7b = uint64(buf[i])i++x += b << 14if b&0x80 == 0 {goto done}x -= 0x80 << 14b = uint64(buf[i])i++x += b << 21if b&0x80 == 0 {goto done}x -= 0x80 << 21b = uint64(buf[i])i++x += b << 28if b&0x80 == 0 {goto done}x -= 0x80 << 28b = uint64(buf[i])i++x += b << 35if b&0x80 == 0 {goto done}x -= 0x80 << 35b = uint64(buf[i])i++x += b << 42if b&0x80 == 0 {goto done}x -= 0x80 << 42b = uint64(buf[i])i++x += b << 49if b&0x80 == 0 {goto done}x -= 0x80 << 49b = uint64(buf[i])i++x += b << 56if b&0x80 == 0 {goto done}x -= 0x80 << 56b = uint64(buf[i])i++x += b << 63if b&0x80 == 0 {goto done}// x -= 0x80 << 63 // Always zero.return 0, errOverflowdone:p.index = ireturn x, nil
}

Int32 序列化之后,第一个字节一定是 0x80 ,那么除去这个字节以后,后面的每个二进制字节都是数据,剩下的步骤就是通过位移操作把每个数字都加起来。上面这个反序列化的函数同样适用于 int32 , int64 , uint32 , uint64 , bool , 和 enum

顺道也可以看看 sint32Fixed32 的反序列化具体代码实现。

func (p *Buffer) DecodeZigzag32() (x uint64, err error) {x, err = p.DecodeVarint()if err != nil {return}x = uint64((uint32(x) >> 1) ^ uint32((int32(x&1)<<31)>>31))return
}

针对有符号的 sint32 ,反序列化的过程就是先反序列 Varint ,再反序列化 Zigzag

func (p *Buffer) DecodeFixed32() (x uint64, err error) {// x, err already 0i := p.index + 4if i < 0 || i > len(p.buf) {err = io.ErrUnexpectedEOFreturn}p.index = ix = uint64(p.buf[i-4])x |= uint64(p.buf[i-3]) << 8x |= uint64(p.buf[i-2]) << 16x |= uint64(p.buf[i-1]) << 24return
}

Fixed32 反序列化的过程也是通过位移,每个字节的内容都累加,就可以还原出原先的数据。注意这里也要先跳过 tag 的位置。

String

func (p *Buffer) DecodeRawBytes(alloc bool) (buf []byte, err error) {n, err := p.DecodeVarint()if err != nil {return nil, err}nb := int(n)if nb < 0 {return nil, fmt.Errorf("proto: bad byte length %d", nb)}end := p.index + nbif end < p.index || end > len(p.buf) {return nil, io.ErrUnexpectedEOF}if !alloc {// todo: check if can get more uses of alloc=falsebuf = p.buf[p.index:end]p.index += nbreturn}buf = make([]byte, nb)copy(buf, p.buf[p.index:])p.index += nbreturn
}

反序列化 string 先把 length 序列化出来,通过 DecodeVarint 的方式。拿到 length 以后,剩下的就是直接拷贝的过程。在上篇 encode 中,我们知道字符串是不做处理,直接放到二进制流里面的,所以反序列化直接取出即可。

Map

func (o *Buffer) dec_new_map(p *Properties, base structPointer) error {raw, err := o.DecodeRawBytes(false)if err != nil {return err}oi := o.index       // index at the end of this map entryo.index -= len(raw) // move buffer back to start of map entrymptr := structPointer_NewAt(base, p.field, p.mtype) // *map[K]Vif mptr.Elem().IsNil() {mptr.Elem().Set(reflect.MakeMap(mptr.Type().Elem()))}v := mptr.Elem() // map[K]V// 这里省略一些代码,主要是为了 key - value 准备的一些可以双重间接寻址的占位符,具体原因可以见序列化代码里面的 enc_new_map 函数// Decode.// This parses a restricted wire format, namely the encoding of a message// with two fields. See enc_new_map for the format.for o.index < oi {// tagcode for key and value properties are always a single byte// because they have tags 1 and 2.tagcode := o.buf[o.index]o.index++switch tagcode {case p.mkeyprop.tagcode[0]:if err := p.mkeyprop.dec(o, p.mkeyprop, keybase); err != nil {return err}case p.mvalprop.tagcode[0]:if err := p.mvalprop.dec(o, p.mvalprop, valbase); err != nil {return err}default:// TODO: Should we silently skip this instead?return fmt.Errorf("proto: bad map data tag %d", raw[0])}}keyelem, valelem := keyptr.Elem(), valptr.Elem()if !keyelem.IsValid() {keyelem = reflect.Zero(p.mtype.Key())}if !valelem.IsValid() {valelem = reflect.Zero(p.mtype.Elem())}v.SetMapIndex(keyelem, valelem)return nil
}

反序列化 map 需要把每个 tag 取出来,然后紧接着反序列化每个 key - value 。最后会判断 keyelemvalelem 是否为零值,如果是零值要分别调用 reflect.Zero 处理零值的情况。

slice

最后还是举一个数组的例子。以 []int32 为例。

func (o *Buffer) dec_slice_packed_int32(p *Properties, base structPointer) error {v := structPointer_Word32Slice(base, p.field)nn, err := o.DecodeVarint()if err != nil {return err}nb := int(nn) // number of bytes of encoded int32sfin := o.index + nbif fin < o.index {return errOverflow}for o.index < fin {u, err := p.valDec(o)if err != nil {return err}v.Append(uint32(u))}return nil
}

反序列化这个数组,分2步,跳过 tagcode 拿到 length ,反序列化 length 。在 length 这个长度中依次反序列化各个 value

上述就是 Protocol Buffer 反序列化的过程。

序列化小结

Protocol Buffer 反序列化直接读取二进制字节数据流,反序列化就是 encode 的反过程,同样是一些二进制操作。反序列化的时候,通常只需要用到 lengthtag 值只是用来标识类型的,PropertiessetEncAndDec() 方法里面已经把每个类型对应的 decode 解码器初始化好了,所以反序列化的时候,tag 值可以直接跳过,从 length 开始处理。

XML 的解析过程就复杂一些。XML 需要从文件中读取出字符串,再转换为 XML 文档对象结构模型。之后,再从 XML 文档对象结构模型中读取指定节点的字符串,最后再将这个字符串转换成指定类型的变量。这个过程非常复杂,其中将 XML 文件转换为文档对象结构模型的过程通常需要完成词法文法分析等大量消耗 CPU 的复杂计算。

回到顶部

序列化/反序列化性能

Protocol Buffer 一直被人们认为是高性能的存在。也有很多人做过实现,验证了这一说法。例如这个链接里面的实验 jvm-serializers。

在看数据之前,我们可以先理性的分析一下 Protocol BufferJSONXML 这些比有哪些优势:

Protobuf 采用了 VarintZigzag 大幅的压缩了整数类型,也没有 JSON 里面的 {、}、;、这些数据分隔符,有 option 字段标识的,没有数据的时候不会进行反序列化。这几个措施导致 pb 的数据量整体的就比 JSON 少很多。

Protobuf 采取的是 TLV 的形式,JSON 这些都是字符串的形式。字符串比对应该比基于数字的字段 tag 更耗时。Protobuf 在正文前有一个大小或者长度的标记,而 JSON 必须全文扫描无法跳过不需要的字段。
下面这张图来自参考链接里面的 《Protobuf有没有比JSON快5倍?用代码来击破pb性能神话》:


从这个实验来看,确实 Protobuf 在序列化数字这方面性能是非常强悍的。

序列化 / 反序列化数字确实是 Protobuf 针对 JSONXML 的优势,但是它也存在一些没有优势的地方。比如字符串。字符串在 Protobuf 中基本没有处理,除了前面加了 tag - length 。在序列化 / 反序列化字符串的过程中,字符串拷贝的速度反而决定的真正的速度。


从上图可以看到 encode 字符串的时候,速度基本和 JSON 相差无几。

回到顶部

最后

至此,关于 protocol buffers 的所有,读者应该了然于胸了。

protocol buffers 诞生之初也并不是为了传输数据存在的,只是为了解决服务器多版本协议兼容的问题。实质其实是发明了一个新的跨语言无歧义的 IDL (Interface description language) 。只不过人们后来发现用它来传输数据也不错,才开始用 protocol buffers

想用 protocol buffers 替换 JSON ,可能是考虑到:

  • protocol buffers 相同数据,传输的数据量比 JSON 小,gzip 或者 7zip 压缩以后,网络传输消耗较少。

  • protocol buffers 不是自我描述的,在缺少 .proto 文件以后,有一定的加密性,数据传输过程中都是二进制流,并不是明文。

  • protocol buffers 提供了一套工具,自动化生成代码也非常方便。

  • protocol buffers 具有向后兼容性,改变了数据结构以后,对老的版本没有影响。

  • protocol buffers 原生完美兼容 RPC 调用。

如果很少用到整型数字,浮点型数字,全部都是字符串数据,那么 JSON 和 protocol buffers 性能不会差太多。纯前端之间交互的话,选择 JSON 或者 protocol buffers 差别不是很大。

与后端交互过程中,用到 protocol buffers 比较多,笔者认为选择 protocol buffers 除了性能强以外,完美兼容 RPC 调用也是一个重要因素。

回到顶部

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/448086.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

解决 VUE前端项目报错: Uncaught ReferenceError : initPage is not defined (initPage 方法是有的,依旧报错找不到)

前些天发现了一个巨牛的人工智能学习网站&#xff0c;通俗易懂&#xff0c;风趣幽默&#xff0c;忍不住分享一下给大家。点击跳转到教程。 1. 明明代码中定义了 initPage 这个方法&#xff0c;&#xff0c;却一直报找不到这个方法&#xff1a; Uncaught ReferenceError: init…

掌握新手学车技巧对于新手来说是非常重要的

刚开始学车的时候对于新手来说很多操作不知道从哪里下手&#xff0c;这个时候&#xff0c;如果按照相关的学车技巧来学习的话&#xff0c;对于新手来说是非常有好处的。下面我们就来学习一下让新手们可以快速进入开车状态的学车技巧吧&#xff01;基本上驾校的教练都会教学员把…

iView学习笔记(三):表格搜索,过滤及隐藏列操作

iView学习笔记(三)&#xff1a;表格搜索&#xff0c;过滤及隐藏某列操作 1.后端准备工作 环境说明 python版本&#xff1a;3.6.6 Django版本&#xff1a;1.11.8 数据库&#xff1a;MariaDB 5.5.60 新建Django项目&#xff0c;在项目中新建app&#xff0c;配置好数据库 api_test…

Jenkins自动编译库并上传服务器

Jenkins自动编译库并上传服务器 github地址 首先添加 git 地址&#xff1a; 再添加定时构建&#xff0c;每天夜里构建一次&#xff1a; 执行 shell 脚本进行构建 cd networklayerecho "build json x86" cmake -S . -B cmake-build-release -DCMAKE_BUILD_TYPERele…

解决:The “data“ option should be a function that returns a per-instance value in component definitions

前些天发现了一个巨牛的人工智能学习网站&#xff0c;通俗易懂&#xff0c;风趣幽默&#xff0c;忍不住分享一下给大家。点击跳转到教程。 1. 只是想定义一个变量&#xff0c;方便页面上调用 。 报错&#xff1a; The "data" option should be a function that re…

gdb 调试 TuMediaService

gdb 调试 TuMediaService github地址 起因 首先需要有 armgdb 环境运行 ./armgdb ./TuMediaService 进入 gdb 模式b hi3531_trcod_interface.cc:98 打断点r 运行程序print this->vdec_config_path_ 打印关键值 在这里我们关注的值已经被修改&#xff0c;由于程序中没有刻…

PyQt安装和环境配置

PyQt安装和环境配置 github地址 首先安装Pycharm 新建一个空的 python 工程&#xff0c;找到 setting 安装第三方模块 PyQT5 , 点加号&#xff0c;先安 PyQT5 , 再安装 pyqt5-tools &#xff0c;后面包含 qtdesinger 以上模块都安完&#xff0c;设置扩展工具的参数找到 sett…

HZOJ 大佬(kat)

及其水水水的假期望&#xff08;然而我已经被期望吓怕了……&#xff09;。 数据范围及其沙雕导致丢掉5分…… 因为其实每天的期望是一样的&#xff0c;考虑分开。 f[i][j]表示做k道题&#xff0c;难度最大为j的概率。 则f[i][j](f[i-1][j])*(j-1)*temq[j]*tem;q为前缀和&#…

F12 界面:请求响应内容 Preview 和 Response 不一致、接口返回数据和 jsp 解析到的内容不一致

前些天发现了一个巨牛的人工智能学习网站&#xff0c;通俗易懂&#xff0c;风趣幽默&#xff0c;忍不住分享一下给大家。点击跳转到教程。 1. 情况描述&#xff1a; 我有一个接口只是简单的查询列表数据并返回给前端作一个表格展示。 接口返回的 userId 数据为&#xff1a;…

为什么新手开车起步总是熄火

最近&#xff0c;深圳市民陈小姐年前考完驾照就买了一辆新车&#xff0c;在过完年后上班的第一天&#xff0c;几乎每次等红绿灯的路口起步时汽车都会熄火&#xff0c;导致身后的司机非常不满狂按车喇叭催她“别挡路”&#xff0c;陈小姐自己也急得冒汗。就像陈小姐这样的新手很…

MSCRM日志配置

之前有很多人问我在MSCRM上日志怎么做&#xff0c;具体的如&#xff08;登录日志&#xff0c;操作日志&#xff09;。个人认为操作日志确实比较难做&#xff08;不过我可以给一个思路可以用触发器或者plugin来实现&#xff0c;不过比较麻烦&#xff0c;对系统压力也比较大&…

数据结构——数组

数组 github地址 数组基础 数组最大的有点&#xff1a;快速查询。索引快数组最好应用于 “索引有语义” 的情况但并非所有有语义的索引都适用于数组&#xff08;身份证号&#xff09;数组也可以处理 ”索引没有语义“ 的情况 封装数组类 数组类该具备的功能&#xff1a;增…

十分钟入门 RocketMQ

前些天发现了一个巨牛的人工智能学习网站&#xff0c;通俗易懂&#xff0c;风趣幽默&#xff0c;忍不住分享一下给大家。点击跳转到教程。 本文首先引出消息中间件通常需要解决哪些问题&#xff0c;在解决这些问题当中会遇到什么困难&#xff0c;Apache RocketMQ作为阿里开源的…

使用delegate类型设计自定义事件

在C#编程中&#xff0c;除了Method和Property&#xff0c;任何Class都可以有自己的事件&#xff08;Event&#xff09;。定义和使用自定义事件的步骤如下&#xff1a; &#xff08;1&#xff09;在Class之外定义一个delegate类型&#xff0c;用于确定事件程序的接口 &#xff0…

UTC Time

整个地球分为二十四时区&#xff0c;每个时区都有自己的本地时间。在国际无线电通信场合&#xff0c;为了统一起见&#xff0c;使用一个统一的时间&#xff0c;称为通用协调时(UTC, Universal Time Coordinated)。UTC与格林尼治平均时(GMT, Greenwich Mean Time)一样&#xff0…

解决:Unknown custom element: <myData> - did you register the component correctly? For recursive compon

前些天发现了一个巨牛的人工智能学习网站&#xff0c;通俗易懂&#xff0c;风趣幽默&#xff0c;忍不住分享一下给大家。点击跳转到教程。 1. 引用一个组件报错&#xff1a; Unknown custom element: <myData> - did you register the component correctly?For recursi…

无处不在的container_of

无处不在的container_of linux 内核中定义了一个非常精炼的双向循环链表及它的相关操作。如下所示&#xff1a; struct list_head {struct list_head* next, * prev; };ubuntu 12.04 中这个结构定义在 /usr/src/linux-headers-3.2.0-24-generic/include/linux/types.h 中&…

程序员学习能力提升三要素

摘要&#xff1a;IT技术的发展日新月异&#xff0c;新技术层出不穷&#xff0c;具有良好的学习能力&#xff0c;能及时获取新知识、随时补充和丰富自己&#xff0c;已成为程序员职业发展的核心竞争力。本文中&#xff0c;作者结合多年的学习经验总结出了提高程序员学习能力的三…

AMD Mantle再添新作,引发下代GPU架构猜想

摘要&#xff1a;今年秋天即将发布的《希德梅尔文明&#xff1a;太空》将全面支持AMD Mantle API&#xff0c;如此强大的功能背后离不开强大的CPU、GPU支持。上周AMD爆出了下一代海盗岛R9 300系列&#xff0c;据网友猜测海盗岛家族可能用上速度更快的HBM堆栈式内存。 小伙伴们…

VUE : 双重 for 循环写法、table 解析任意 list 、万能表格组件、解析一维数组、动态生成 table 所有数据

前些天发现了一个巨牛的人工智能学习网站&#xff0c;通俗易懂&#xff0c;风趣幽默&#xff0c;忍不住分享一下给大家。点击跳转到教程。 1.需求&#xff1a; 我想要一个 table 组件能在实际调用时动态生成所有的 tr 、td 。 后端返回的只是一个 list &#xff0c; 前端页…