网易云商·七鱼智能客服自适应 ProtoStuff 数据库缓存实践

ゞ浴缸里的玫瑰 2023-09-27 23:21 0阅读 0赞

![format_png][]

## 需求背景 ##

目前，网易云商·七鱼智能客服数据库缓存使用了 spring-data-redis 框架，并由自研的缓存组件进行管理。该组件使用 Jackson 框架对缓存数据进行序列化和反序列化，并将其以明文 JSON 的形式存储在 Redis 中。

这种方式存在两个问题：

*  **速度慢，CPU占用高**

在应用服务中，读写缓存数据时需要进行字符串的反序列化和序列化操作，即将对象转换为 JSON 格式再转换为字节数组，但是使用 Jackson 序列化方式的性能并不是最优的。此外，在线上服务分析中发现，对于缓存命中率较高的应用，在并发稍微高一点的情况下，Jackson 序列化会占用较多的 CPU 资源。

*  **存储空间大，资源浪费**

对于 Redis 集群来说，JSON 数据占用的存储空间较大，会浪费 Redis 存储资源。

在对同类序列化框架进行调研后，我们决定使用 ProtoStuff 代替 Jackson 框架。本文将简要介绍 ProtoStuff 的存储原理，并讨论在替换过程中遇到的一些问题。

## 关于 ProtoStuff ##

### 什么是 ProtoStuff？ ###

ProtoStuff 是一种基于 Google Protocol Buffers（protobuf）协议的序列化和反序列化库，它可以将 Java 对象序列化为二进制数据并进行网络传输或存储，也可以将二进制数据反序列化为 Java 对象。与其他序列化库相比，ProtoStuff 具有更高的性能和更小的序列化大小，因为它使用了基于标记的二进制编码格式，同时避免了 Java 序列化的一些缺点，例如序列化后的数据过大和序列化性能较慢等问题。因此，ProtoStuff 被广泛应用于高性能的分布式系统和大规模数据存储系统中。

Protostuff 的序列化编码算法与 Protobuf 基本相同，都采用基于 Varint 编码的变长序列化方式，以实现对编码后的字节数组的压缩。此外，Protostuff 还引入了 LinkedBuffer 这种数据结构，通过链表的方式将不连续内存组合起来，从而实现数据的动态扩张，提高存储效率。

> Varint 编码是一种可变长度的整数编码方式，用于压缩数字数据，使其更加紧凑。它使用 1 个或多个字节来表示一个整数，其中每个字节的高位都用于指示下一个字节是否属于同一个数。较小的数字使用较少的字节编码，而较大的数字则需要更多的字节编码。这种编码方式被广泛应用于网络传输和存储领域。

### LinkedBuffer ###

简单看一下 LinkedBuffer 的源码：

arduino

复制代码

`public final class LinkedBuffer{ /** * The minimum buffer size for a {@link LinkedBuffer}. */ public static final int MIN_BUFFER_SIZE = 256; /** * The default buffer size for a {@link LinkedBuffer}. */ public static final int DEFAULT_BUFFER_SIZE = 512; final byte[] buffer; final int start; int offset; LinkedBuffer next; }`

byte\[\] buffer 是用来存储序列化过程中的字节数组的，默认的大小是 512，最低可以设置成 256。LinkedBuffer next 指向的是下一个节点。start 是开始位置，offset 是偏移量。

链表大概长这样，这样就可以把几块连续的内存块给链接到一起了。

![format_png 1][]

### **Schema 接口** ###

除了 LinkedBuffer 这个类，还有一个关键的接口：Schema，这是一个类似于数据库 DDL 结构的东西，它定义了序列化对象的类的结构信息，有哪些字段，字段的顺序是怎么样的，怎样序列化，怎样反序列化。

在使用的时候一般用的是 RuntimeSchema 这个实现类。

ini

复制代码

`public final class RuntimeSchema<T> implements Schema<T>, FieldMap<T> { private final FieldMap<T> fieldMap; public static <T> RuntimeSchema<T> createFrom(Class<T> typeClass, Set<String> exclusions, IdStrategy strategy) { // 省略部分代码 final Map<String, java.lang.reflect.Field> fieldMap = findInstanceFields(typeClass); final ArrayList<Field<T>> fields = new ArrayList<Field<T>>(fieldMap.size()); int i = 0; boolean annotated = false; for (java.lang.reflect.Field f : fieldMap.values()) { if (!exclusions.contains(f.getName())) { if (f.getAnnotation(Deprecated.class) != null) { i++; continue; } final Tag tag = f.getAnnotation(Tag.class); final int fieldMapping; final String name; if (tag == null) { // 省略部分代码 fieldMapping = ++i; name = f.getName(); } else { // 省略部分代码 annotated = true; fieldMapping = tag.value(); // 省略部分代码 name = tag.alias().isEmpty() ? f.getName() : tag.alias(); } final Field<T> field = RuntimeFieldFactory.getFieldFactory(f.getType(), strategy).create(fieldMapping, name, f, strategy); fields.add(field); } } return new RuntimeSchema<T>(typeClass, fields, RuntimeEnv.newInstantiator(typeClass)); } static void fill(Map<String, java.lang.reflect.Field> fieldMap, Class<?> typeClass) { if (Object.class != typeClass.getSuperclass()) fill(fieldMap, typeClass.getSuperclass()); for (java.lang.reflect.Field f : typeClass.getDeclaredFields()) { int mod = f.getModifiers(); if (!Modifier.isStatic(mod) && !Modifier.isTransient(mod) && f.getAnnotation(Exclude.class) == null) fieldMap.put(f.getName(), f); } } @Override public List<Field<T>> getFields() { return fieldMap.getFields(); } @Override public final void writeTo(Output output, T message) throws IOException { for (Field<T> f : getFields()) f.writeTo(output, message); } }`

根据 fill 方法的实现，我们可以得知 fieldMap 是通过调用当前类及其父类的 getDeclaredFields 方法所获取的所有字段。接着，在 createFrom 方法中，我们遍历所有字段，获取每个字段的序列化序号 fieldMapping。在序列化过程中，我们调用 writeTo 方法，将每个字段按照 fieldMapping 的顺序写入字节数组中。

众所周知，Java 的 getDeclaredFields 方法返回的字段数组不是按照特定的顺序排列的。字段的顺序取决于具体的 JVM 实现以及编译器等因素。因此，在不使用 Tag 注解的时候，**序列化的字段顺序是不固定的**。如果在原有的字段中间随意插入一个字段，或者在合并代码的时候调换了字段的顺序，反序列化的数据不仅会错乱，很大概率还会报错。

![format_png 2][]

在 ProtoStuff 的官方文档里，**推荐使用 @Tag 注解来显式的声明字段序列化的顺序**。Tag 注解对于小项目或者固定不会变的类对象确实是挺好用的，但是对于老项目序列化框架迁移来说，多个代码仓库超过 400 个对象需要加 Tag 注解，代码改动量和影响范围将会非常庞大。而且一旦有字段加了 Tag 注解，那么后续添加的所有字段都需要添加注解，并且需要保证新增字段的顺序是递增的，会有一定的维护成本和风险。

## **自适应 ProtoStuff 的改造方案** ##

为了减少序列化框架迁移过程的代码改动范围和风险，降低后期编码维护成本，我们需要一个可以在序列化与反序列化时自动适配字段的改造方案。

### **主要思路** ###

#### **序列化** ####

*  将 getDeclaredFields 方法获取到的当前类及其父类所有的字段，根据字段名称进行排序。
 *  遍历排序后的字段列表，将字段转换成 ProtoStuff 需要的 Field 列表，再调用 RuntimeSchema 的构造方法新建一个对象。通过 RuntimeSchema 对象完成序列化操作，生成字节数组。
 *  由于 ProtoStuff 的编码是 T-L-V 格式的，只存了对象字段的下标和具体的值，没有存完整的类路径，而且 spring-data-redis 反序列化的时候不知道目标对象的类型，因此还需要一个包装类来存储额外的信息。
 *  对统一包装对象进行序列化，返回生成的字节数组。
 *  将缓存对象的类结构信息缓存到 Redis 中，以便反序列化时使用。

![format_png 3][]

为了提供序列化的效率，还可以将 RuntimeSchema 对象缓存到本地。

#### **反序列化** ####

*  将字节数组反序列化成通用的包装类。
 *  从包装类中获取到源数据的类路径，版本号，字段哈希值。先判断源数据类是否是集合或者基本数据类型，如果是基本数据类型，直接返回 source 字段的内容。如果是集合类，判断本地版本号是否与包装类获取到的版本号一致，一致的时候返回 source 字段的内容。
 *  源数据类型既不是集合也不是基本数据类型，获取本地对象的版本号，如果本地对象版本号大于缓存版本号，则将缓存数据淘汰掉。
 *  如果本地对象的版本号和缓存中的版本号一致，就直接使用本地类进行转换，获取到 RuntimeSchema 进行反序列化。
 *  如果本地对象的版本号小于缓存中的版本号，则需要根据类路径 + 缓存版本号 从 Redis 中获取到对应的类结构信息，将本地的字段进行重新排序，获取到和缓存数据对应的字段顺序值，再生成相应的 RuntimeSchema 进行反序列化。

## **代码实现** ##

ProtoStuff 的入门使用是很简单的，只需要引入 ProtoStuff 的依赖，然后在需要使用序列化的类字段上加上 Tag 注解即可使用。也可以不使用注解，ProtoStuff 会根据字段顺序来确定缓存中的顺序。

### **增加 Maven 依赖** ###

xml

复制代码

` <dependency> <groupId>io.protostuff</groupId> <artifactId>protostuff-core</artifactId> <version>1.7.4</version> </dependency> <dependency> <groupId>io.protostuff</groupId> <artifactId>protostuff-runtime</artifactId> <version>1.7.4</version> </dependency>`

### **统一包装类** ###

java

复制代码

`public class ProtoStuffWrapper implements Serializable { private static final long serialVersionUID = 6310017353904821602L; // 版本号 @Tag(1) private int version; // 包装类型的完整路径名 @Tag(2) private String className; // 包装对象序列化后的字节数组 @Tag(3) private byte[] data; // 是否是没有包装的类型 @Tag(4) private boolean noWrapperObject; // 用于存储集合对象 @Tag(5) private Object source; // 类字段hash @Tag(6) private int classHash; // 省略 get set 和 构造方法 }`

对于基本数据类型和一些 Java 的基础对象，以及集合，Map 类对象，会直接将数据放在 source 中。

### **重写序列化方法** ###

实现 org.springframework.data.redis.serializer.RedisSerializer 接口， 重写序列化方法。

#### **流程图** ####

![format_png 4][]

#### **代码** ####

ini

复制代码

`public class ProtostuffRedisSerializer implements RedisSerializer<Object> { private static final Map<String, ProtoSchema> SCHEMA_CACHE = new ConcurrentHashMap<>(200); private static final Map<String, Schema> REMOTE_CLASS_SCHEMA_CACHE = new ConcurrentHashMap<>(200); private static final Delegate<Timestamp> TIMESTAMP_DELEGATE = new TimestampDelegate(); private static final DefaultIdStrategy ID_STRATEGY = (DefaultIdStrategy) RuntimeEnv.ID_STRATEGY; private static final ThreadLocal<LinkedBuffer> BUFFER = ThreadLocal.withInitial(() -> LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE)); private static final Schema WRAPPER_SCHEMA = RuntimeSchema.getSchema(ProtoStuffWrapper.class); private static final int SECONDS_OF_THIRTY_DAYS = 30 * 60 * 60 * 24; private static final long MILLISECOND_OF_THIRTY_DAYS = SECONDS_OF_THIRTY_DAYS * 1000L; private final StringRedisTemplate redisTemplate; static { ID_STRATEGY.registerDelegate(TIMESTAMP_DELEGATE); } public ProtostuffRedisSerializer(StringRedisTemplate redisTemplate) { this.redisTemplate = redisTemplate; } @Override public byte[] serialize(Object o) throws SerializationException { if (source == null) { return EMPTY_ARRAY; } LinkedBuffer buffer = BUFFER.get(); byte[] data = new byte[0]; try { String className = getClassName(source); Class<?> typeClass = source.getClass(); Object serializeObj; if (isBasicType(source, className) || isArrayType(typeClass)) { int classVersion = 0; if (isArrayType(typeClass)) { classVersion = readVersion(source); } serializeObj = new ProtoStuffWrapper(className, classVersion, source); } else { ProtoSchema protoSchema = getCachedProtoSchema(className, source); try { data = ProtostuffIOUtil.toByteArray(source, protoSchema.getSchema(), buffer); } finally { buffer.clear(); } serializeObj = new ProtoStuffWrapper(className, data, protoSchema); } data = ProtostuffIOUtil.toByteArray(serializeObj, WRAPPER_SCHEMA, buffer); } catch (Exception e) { logger.error("protostuff serialize fail", e); } finally { buffer.clear(); } return data; } @Override public Object deserialize(byte[] bytes) throws SerializationException { return deserialize(source, Object.class); } }`

从上面的 deserialize 方法的定义中可以看到，入参只有一个字节数组，出参是一个 Object，没有 Class 类的参数，因此必须要有一个统一的包装类来保存目标类的定义信息。

#### **Timestamp 序列化代理** ####

对于 Timestamp 类型的字段需要自己写一个序列化代理去处理，不然会有解析失败的问题。

java

复制代码

`public class TimestampDelegate implements Delegate<Timestamp> { @Override public WireFormat.FieldType getFieldType() { return WireFormat.FieldType.FIXED64; } @Override public Timestamp readFrom(Input input) throws IOException { return new Timestamp(input.readFixed64()); } @Override public void writeTo(Output output, int number, Timestamp timestamp, boolean repeated) throws IOException { output.writeFixed64(number, timestamp.getTime(), repeated); } @Override public void transfer(Pipe pipe, Input input, Output output, int number, boolean repeated) throws IOException { output.writeFixed64(number, input.readFixed64(), repeated); } @Override public Class<?> typeClass() { return Timestamp.class; } }`

#### **ProtoSchema** ####

本地缓存对象，用来缓存序列化对象的 RuntimeSchema 和类的相关信息。

arduino

复制代码

`public class ProtoSchema { // 版本号 private int version; // 类字段hash private int hash; // 序列化对象的RuntimeSchema private Schema schema; // 本地缓存生效开始时间 private long createTime; // 省略 get set 和 构造方法 }`

#### **getCachedProtoSchema** ####

获取序列化对象的 RuntimeSchema 和类的相关信息。本地缓存中存在则直接使用缓存中的数据，不存在时，解析类对象，根据排序后的字段构建 RuntimeSchema 来进行序列化。

ini

复制代码

`private ProtoSchema getCachedProtoSchema(String className, Object source) { ProtoSchema protoSchema = SCHEMA_CACHE.get(className); if (protoSchema != null) { if (protoSchema.getVersion() == 0) { // 基本类型包装类直接返回 return protoSchema; } if (System.currentTimeMillis() - protoSchema.getCreateTime() < MILLISECOND_OF_THIRTY_DAYS) { // 本地缓存在有效期内直接返回，不在有效期的重新加载类结构信息 return protoSchema; } } Class<?> typeClass = source.getClass(); List<Field<?>> fields = new ArrayList<>(); LinkedHashMap<String, java.lang.reflect.Field> fieldMap = new LinkedHashMap<>(); fill(fieldMap, typeClass); java.lang.reflect.Field[] declaredFields = fieldMap.values().toArray(new java.lang.reflect.Field[0]); // 按字段名进行排序 Arrays.sort(declaredFields, Comparator.comparing(java.lang.reflect.Field::getName)); int length = declaredFields.length; List<ProtoFieldDescription> fieldDescriptionList = new ArrayList<>(length); java.lang.reflect.Field f; Class<?> type; io.protostuff.runtime.Field<?> field; ProtoFieldDescription d; int index = 0; for (java.lang.reflect.Field declaredField : declaredFields) { f = declaredField; type = f.getType(); d = new ProtoFieldDescription(f.getName(), ++index, type.getCanonicalName()); fieldDescriptionList.add(d); field = RuntimeFieldFactory.getFieldFactory(type, ID_STRATEGY).create(d.getIndex(), d.getFieldName(), f, ID_STRATEGY); fields.add(field); } RuntimeSchema schema = new RuntimeSchema(typeClass, fields, RuntimeEnv.newInstantiator(typeClass)); String[] fieldNames = fieldDescriptionList.stream().map(ProtoFieldDescription::getFieldName).toArray(String[]::new); protoSchema = new ProtoSchema(readVersion(source), Arrays.hashCode(fieldNames), schema); // 本地缓存ProtoStuffSchema SCHEMA_CACHE.putIfAbsent(className, protoSchema); // 缓存类结构信息到Redis cacheFieldDescription(getCacheKey(className, protoSchema.getVersion()), JSON.toJSONString(fieldDescriptionList)); return protoSchema; } static void fill(Map<String, java.lang.reflect.Field> fieldMap, Class<?> typeClass) { if (Object.class != typeClass.getSuperclass()) { fill(fieldMap, typeClass.getSuperclass()); } for (java.lang.reflect.Field f : typeClass.getDeclaredFields()) { int mod = f.getModifiers(); if (!Modifier.isStatic(mod) && !Modifier.isTransient(mod) && f.getAnnotation(Exclude.class) == null) { fieldMap.put(f.getName(), f); }`

将 ProtoStuffSchema 缓存在本地，可以避免每次都重复解析类的结构，优化性能。本地缓存增加了有效期，可以保存 Redis 中的类结构信息和本地缓存中的一致，从而避免出现 Redis 中的数据过期导致老版本应用没法读取到对应版本类结构信息的情况。

RuntimeSchema(java.lang.Class, java.util.Collection<io.protostuff.runtime.field>, io.protostuff.runtime.RuntimeEnv.Instantiator) 这个构造方法是自适应的关键，正是因为有了这个构造方法，我们才能自己构建字段的顺序。

### **重写反序列化方法** ###

#### **流程图** ####

![format_png 5][]

首先，需要对字节数组进行解析，以得到相应的统一包装类。随后，需要根据缓存版本号和本地类版本号进行比较，以确定是否需要使用缓存中的数据。

生成版本号的逻辑是：基础版本号加上类的字段数量。如果版本号相同，我们还需要检查类的字段哈希值，然后根据字段哈希值获取排序后的字段名的哈希值。

#### **代码** ####

scss

复制代码

`public <T> T deserialize(byte[] source, Class<T> type) throws SerializationException { if (isEmpty(source)) { return null; } try { ProtoStuffWrapper wrapper = new ProtoStuffWrapper(); ProtostuffIOUtil.mergeFrom(source, wrapper, WRAPPER_SCHEMA); int cacheVersion = wrapper.getVersion(); if (wrapper.isNoWrapperObject()) { // 集合数组,基本类型包装类 缓存对象，缓存与本地版本不一致，直接淘汰掉 if (cacheVersion == 0 || cacheVersion == inferVersion(wrapper.getSource())) { return (T) wrapper.getSource(); } return null; } String className = wrapper.getClassName(); if (StringUtils.isNotEmpty(className)) { Class<?> typeClass = Class.forName(className); ProtoSchema protoSchema = getProtoSchema(className, typeClass); int localVersion = protoSchema.getVersion(); if (cacheVersion >= localVersion) { Schema cachedSchema = getCachedSchema(wrapper, typeClass, protoSchema); if (cachedSchema != null) { Object newMessage = cachedSchema.newMessage(); ProtostuffIOUtil.mergeFrom(wrapper.getData(), newMessage, cachedSchema); return (T) newMessage; } } } } catch (Exception e) { // 缓存，本地结构不一致, 打印一个错误日志 } return null; } private ProtoSchema getProtoSchema(String className, Class<?> typeClass) throws InstantiationException, IllegalAccessException { ProtoSchema protoSchema = SCHEMA_CACHE.get(className); if (protoSchema != null) { return protoSchema; } return getCachedProtoSchema(className, typeClass.newInstance()); } private Schema getCachedSchema(ProtoStuffWrapper wrapper, Class<?> typeClass, ProtoSchema protoSchema) { if (wrapper.getVersion() == protoSchema.getVersion()) { if (protoSchema.getHash() == wrapper.getClassHash()) { return protoSchema.getSchema(); } else { // 缓存，本地结构不一致, 打印一个错误日志 // logger.error("警告，本地与缓存中的版本号一致，但是字段顺序不一致，应用存在异常。请重新部署, className:{}", wrapper.getClassName()); } } // 缓存中为新版本，本地为老版本 return getSchemaFromCache(typeClass, wrapper); }`

#### **getCachedSchema** ####

本地版本为老版本，缓存版本为新版本时，反序列化的时候需要先获取到 Redis 中新版本的类描述信息。为了避免重复请求 Redis，类描述信息也会在本地缓存一份数据。

ini

复制代码

`private <T> Schema<T> getSchemaFromCache(Class<?> typeClass, ProtoStuffWrapper wrapper) { String cacheKey = getCacheKey(wrapper.getClassName(), wrapper.getVersion()); Schema schema = REMOTE_CLASS_SCHEMA_CACHE.get(cacheKey); if (schema != null) { return schema; } Map<String, ProtoFieldDescription> fieldDescriptionMap = getProtoFieldDescriptionMap(cacheKey); if (MapUtils.isEmpty(fieldDescriptionMap)) { return null; } java.lang.reflect.Field[] declaredFields = typeClass.getDeclaredFields(); final ArrayList<io.protostuff.runtime.Field<T>> fields = new ArrayList<>(declaredFields.length); ProtoFieldDescription d; for (java.lang.reflect.Field field : declaredFields) { d = fieldDescriptionMap.get(field.getName()); if (d != null) { Class<?> type = field.getType(); if (Objects.equals(d.getType(), type.getCanonicalName())) { // 字段类型一致 io.protostuff.runtime.Field<T> pField = RuntimeFieldFactory.getFieldFactory(type, ID_STRATEGY).create(d.getIndex(), d.getFieldName(), field, ID_STRATEGY); fields.add(pField); } } } schema = new RuntimeSchema(typeClass, fields, RuntimeEnv.newInstantiator(typeClass)); REMOTE_CLASS_SCHEMA_CACHE.putIfAbsent(cacheKey, schema); return schema; } private Map<String, ProtoFieldDescription> getProtoFieldDescriptionMap(String key) { String cache = getStringFromRedis(key); if (StringUtils.isEmpty(cache)) { return new ConcurrentHashMap<>(); } List<ProtoFieldDescription> fieldDescriptionList = JSON.parseArray(cache, ProtoFieldDescription.class); if (fieldDescriptionList == null) { return new ConcurrentHashMap<>(); } return fieldDescriptionList.stream().collect(Collectors.toMap(ProtoFieldDescription::getFieldName, Function.identity(), (a, b) -> b)); }`

## **总结** ##

ProtoStuff 是一个非常优秀的 Java 序列化框架，具有高效性、空间占用小、易用性和可扩展性等优点。

本方案在设计之初，考虑到数据库缓存序列化框架作为缓存组件的一部分，需要更多地为使用的业务方考虑。因此，改造方案花费了大量精力将框架做成自适应的。此举的目的在于，让接入方在使用过程中无需担心新增字段可能会引发的反序列化顺序问题，也无需额外维护 Tag 标签的顺序，更不需要对历史代码进行兼容改造。只要简单的升级一下依赖的二方包，就可以实现组件的升级。

[format_png]: https://img-blog.csdnimg.cn/img_convert/fc834a179fa78ce0d8724a6d19268c53.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/8fc3ca63a30bf708253b2621c4a3087f.webp?x-oss-process=image/format,png
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/fdb260c374577a5baed274a21b27144f.webp?x-oss-process=image/format,png
[format_png 3]: https://img-blog.csdnimg.cn/img_convert/bbce113148bf587a0ce0b79563e33981.webp?x-oss-process=image/format,png
[format_png 4]: https://img-blog.csdnimg.cn/img_convert/fbad98c87685dc9dafee41fafaca3966.webp?x-oss-process=image/format,png
[format_png 5]: https://img-blog.csdnimg.cn/img_convert/fe7e213e3ffee33f0c93a56eb260884e.webp?x-oss-process=image/format,png