Using Protostuff and Kryo high-performance serialization framework to achieve RestTemplate serialization components

This article has participated in the “Digitalstar Project” and won a creative gift package to challenge the creative incentive money.

serialization

• Serialization can be simply understood as an object -> byte process, and deserialization is the opposite process. Why serialization? Because network transport only recognizes bytes. So the process of trust depends on serialization.

• Network transmission performance and many other factors, usually support a variety of serialization methods for users to plug and unplug use, some commonly used serialization schemes hessian, Kryo, Protostuff, FST, etc., among which the fastest and best effect is kryo and Protostuff

RedisConfiguration configuration

1. Create a RedisConnectionFactory object

2. Create the RestTemplate object based on the RedisConnectionFactory object.

3. Configure the associated RedisSerializaer components

@Configuration
public class RedisConfiguration {

    @Bean("redisConnectionFactory")
    public RedisConnectionFactory redisConnectionFactory(RedisConfigMapper mapper) {
        List<RedisConfig> redisConfigs = mapper.getRedisConfig();
        List<String> clusterNodes = new ArrayList<>();
        for (RedisConfig rc : redisConfigs) {
            clusterNodes.add(rc.getUrl() + ":" + rc.getPort());
        }
        // Get Redis cluster configuration information
        RedisClusterConfiguration rcf = new RedisClusterConfiguration(clusterNodes);
        return new JedisConnectionFactory(rcf);
    }

    @Bean("redisTemplate")
    public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) throws UnknownHostException {
        RedisTemplate<Object, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(redisConnectionFactory);
        // The serializer used by redis value
        template.setValueSerializer(new XXXRedisSerializer<>());
        // The serializer used by redis key
        template.setKeySerializer(new XXXRedisSerializer<>());
        template.setHashKeySerializer(new XXXRedisSerializer<>());
        template.setHashValueSerializer(new XXXRedisSerializer<>());
        template.afterPropertiesSet();
        returntemplate; }}Copy the code

Kryo serialization implementation

Maven configuration file

<dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-autoconfigure</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
        <dependency>
            <groupId>com.esotericsoftware</groupId>
            <artifactId>kryo</artifactId>
            <version>4.0.1</version>
        </dependency>
        <dependency>
            <groupId>de.javakaffee</groupId>
            <artifactId>kryo-serializers</artifactId>
            <version>0.41</version>
        </dependency>
		<dependency>
        	<groupId>com.esotericsoftware</groupId>
        	<artifactId>kryo-shaded</artifactId>
        	<version>4.0.1</version>
    	</dependency>
    </dependencies>
Copy the code

Because it relies on ASM technology at the bottom, it can be version-shaded with asM-dependencies in frameworks such as Spring (which is well documented), so another dependency is provided to solve this problem: Kryo-shaded

Kryo three ways to read and write

If you know the class bytecode and the object is not empty

  kryo.writeObject(output, classObject);
  RestClass restClass = kryo.readObject(input, RestClass.class);
Copy the code

This is one of the serialization/deserialization methods in Quickstart. However, Kryo provides a second set of read and write methods considering that someObject may be null, which will also result in null returned results.

If you know the class bytecode, and the object may be empty

kryo.writeObjectOrNull(output, classObject);
RestClass someObject = kryo.readObjectOrNull(input, RestClass.class);
Copy the code

But the two methods can’t seem to meet our requirements, in RPC calls, serialization and deserialization distribution in different endpoints, object type, we don’t want to rely on manual specify the parameters, it is best to bytecode deposit information directly to the serialization of results, during deserialization bytecode information on its own. With this in mind, Kryo offers a third way. If the bytecode of the implementation class is unknown and the object may be null.

  kryo.writeClassAndObject(output, object);
Object object = kryo.readClassAndObject(input);
if (object instanceof RestClass) {
}
Copy the code

We sacrifice some space and some performance to store bytecode information

Supported serialization types

The table above shows the types that are supported by default.

Kryo kryo = new Kryo();
kryo.addDefaultSerializer(RestClass.class, RestSerializer.class);
Copy the code

In this way, the serializer can also be extended for a Kryo instance

Kryo support types:

• The enumeration
• Collection, array
• Subclass/polymorphism
• A circular reference
• The inner class
• The generic

Exception problem with Kryo deserialization

• Kryo does not support adding or deleting fields in beans. If you serialize a class using Kryo, store it in Redis, and modify the class, deserialization will fail.

• Another point to note is the support for serialization of some classes created using reflection. Such as the Arrays. AsList (); Creates a List object that causes a serialization exception.

• Deserialization of a class that contains a no-argument constructor is not supported. Attempting to deserialize a class that does not contain a no-argument constructor will get the following exception:

• It is a programming specification that should be followed to ensure that every class has a no-argument constructor, but in practice some third library related classes that do not contain no-argument constructors are a bit of a hassle.

Kryo is thread unsafe

ThreadLocal is maintained to keep them thread-safe.

private static final ThreadLocal<Kryo> kryos = new ThreadLocal<Kryo>() {
    protected Kryo initialValue(a) {
        Kryo kryo = new Kryo();
        // configure kryo instance, customize settings
        return kryo;
    };
};

// Somewhere else, use KryoKryo k = kryos.get(); .Copy the code

Kryo configuration parameters

Each Kryo instance can have two configuration parameters.

• kryo.setRegistrationRequired(false); // Close the registration behavior

Register (someclazz.class), which assigns the class a number starting from 0. However, the biggest problem with Kryo is that it does not guarantee that the same class has the same registration number every time. This depends on the order of registration. This means that different machines may have different numbers before and after the same machine is restarted, which can cause serialization problems, so in distributed projects, registration is usually turned off.

• kryo.setReferences(true); // Support circular references

Circular references, Kryo can turn off support for circular references for high performance. I don’t think turning it off is a good option, however, and for the most part, please keep kryo.setreferences (true).

The Kryo utility class is commonly used

public class KryoSerializer {

    public byte[] serialize(Object obj) {
        Kryo kryo = kryoLocal.get();
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        Output output = new Output(byteArrayOutputStream);/ / < 1 >
        kryo.writeClassAndObject(output, obj);/ / < 2 >
        output.close();
        return byteArrayOutputStream.toByteArray();
    }

    public <T> T deserialize(byte[] bytes) {
        Kryo kryo = kryoLocal.get();
        ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);
        Input input = new Input(byteArrayInputStream);/ / < 1 >
        input.close();
        return (T) kryo.readClassAndObject(input);/ / < 2 >
    }

    private static final ThreadLocal<Kryo> kryoLocal = new ThreadLocal<Kryo>() {/ / < 3 >
        @Override
        protected Kryo initialValue(a) {
            Kryo kryo = new Kryo();
            kryo.setReferences(true);// The default value is true to emphasize action
            kryo.setRegistrationRequired(false);// The default value is false to emphasize the role
            returnkryo; }}; }Copy the code

1. Kryo’s Input and Output receive an InputStream and an OutputStream. Kryo usually converts byte arrays to objects. So commonly used input and output flow to achieve ByteArrayInputStream/ByteArrayOutputStream.

2. The pairing of writeClassAndObject and readClassAndObject is most common in distributed scenarios, where bytecode is stored in the serialization result during serialization so that bytecode information is not passed in during deserialization.

3. Using ThreadLocal to maintain Kryo instances reduces the overhead of instantiating Kryo once for every use and keeps it thread-safe.

KryoRedisSerializer

Data exchange or data persistence, such as using Kryo to serialize objects into byte arrays to send to message queues or to redis, etc.

public class KryoRedisSerializer<T> implements RedisSerializer<T> {

    private static final String DEFAULT_ENCODING = "UTF-8";
 
    // Kryo instance per thread
    private static final ThreadLocal<Kryo> kryoLocal = new ThreadLocal<Kryo>() {
        @Override
        protected Kryo initialValue(a) {
            Kryo kryo = new Kryo();
 
            /** * Do not easily change the configuration here! After the change, the format of the serialization will change and all caches in Redis will have to be cleared at the same time. Otherwise, when those caches come back to deserialize, an error will be reported
            // Support object circular reference (otherwise stack overflow)
            kryo.setReferences(true); // The default value is true. This line is added to remind maintainers not to change this configuration
 
            // Class registration is not mandatory (registration behavior does not guarantee the same class registration number across multiple JVMS; And a large number of classes in the business system are difficult to register.)
            kryo.setRegistrationRequired(false); // The default value is false. This line is added to remind maintainers not to change this configuration
 
            //Fix the NPE bug when deserializing Collections.
            ((Kryo.DefaultInstantiatorStrategy) kryo.getInstantiatorStrategy())
                    .setFallbackInstantiatorStrategy(new StdInstantiatorStrategy());
 
            returnkryo; }};/** * Get the Kryo instance of the current thread **@returnThe Kryo instance of the current thread */
    public static Kryo getInstance(a) {
        return kryoLocal.get();
    }

     @Override
    public byte[] serialize(T t) throws SerializationException {
        byte[] buffer = new byte[2048];
        Output output = new Output(buffer);
        getInstance().writeClassAndObject(output, t);
        return output.toBytes();
    }

    @Override
    public T deserialize(byte[] bytes) throws SerializationException {
        Input input = new Input(bytes);
        @SuppressWarnings("unchecked")
        T t = (T) getInstance().readClassAndObject(input);
        returnt; }}Copy the code

Protostuff serialization implementation

Maven configuration file

<! -- serialize -->
<dependency>
    <groupId>com.dyuproject.protostuff</groupId>
    <artifactId>protostuff-core</artifactId>
    <version>1.1.3</version>
</dependency>
<dependency>
    <groupId>com.dyuproject.protostuff</groupId>
    <artifactId>protostuff-runtime</artifactId>
    <version>1.1.3</version>
</dependency>
Copy the code

Define a ProtoStuffUtil utility class

@Slf4j
public class ProtoStuffUtil {
    /** * serialize the object **@param obj
     * @return* /
    public static <T> byte[] serialize(T obj) {
        if (obj == null) {
            log.error("Failed to serializer, obj is null");
            throw new RuntimeException("Failed to serializer");
        }
 
        @SuppressWarnings("unchecked") Schema<T> schema = (Schema<T>) RuntimeSchema.getSchema(obj.getClass());
        LinkedBuffer buffer = LinkedBuffer.allocate(1024 * 1024);
        byte[] protoStuff;
        try {
            protoStuff = ProtostuffIOUtil.toByteArray(obj, schema, buffer);
        } catch (Exception e) {
            log.error("Failed to serializer, obj:{}", obj, e);
            throw new RuntimeException("Failed to serializer");
        } finally {
            buffer.clear();
        }
        return protoStuff;
    }
 
    /** * deserialize the object **@param paramArrayOfByte
     * @param targetClass
     * @return* /
    public static <T> T deserialize(byte[] paramArrayOfByte, Class<T> targetClass) {
        if (paramArrayOfByte == null || paramArrayOfByte.length == 0) {
            log.error("Failed to deserialize, byte is empty");
            throw new RuntimeException("Failed to deserialize");
        }
 
        T instance;
        try {
            instance = targetClass.newInstance();
        } catch (InstantiationException | IllegalAccessException e) {
            log.error("Failed to deserialize", e);
            throw new RuntimeException("Failed to deserialize");
        }
 
        Schema<T> schema = RuntimeSchema.getSchema(targetClass);
        ProtostuffIOUtil.mergeFrom(paramArrayOfByte, instance, schema);
        return instance;
    }
 
    /** * serialize the list **@param objList
     * @return* /
    public static <T> byte[] serializeList(List<T> objList) {
        if (objList == null || objList.isEmpty()) {
            log.error("Failed to serializer, objList is empty");
            throw new RuntimeException("Failed to serializer");
        }
 
        @SuppressWarnings("unchecked") Schema<T> schema =
                (Schema<T>) RuntimeSchema.getSchema(objList.get(0).getClass());
        LinkedBuffer buffer = LinkedBuffer.allocate(1024 * 1024);
        byte[] protoStuff;
        ByteArrayOutputStream bos = null;
        try {
            bos = new ByteArrayOutputStream();
            ProtostuffIOUtil.writeListTo(bos, objList, schema, buffer);
            protoStuff = bos.toByteArray();
        } catch (Exception e) {
            log.error("Failed to serializer, obj list:{}", objList, e);
            throw new RuntimeException("Failed to serializer");
        } finally {
            buffer.clear();
            try {
                if(bos ! =null) { bos.close(); }}catch(IOException e) { e.printStackTrace(); }}return protoStuff;
    }
 
    /** * deserialize the list **@param paramArrayOfByte
     * @param targetClass
     * @return* /
    public static <T> List<T> deserializeList(byte[] paramArrayOfByte, Class<T> targetClass) {
        if (paramArrayOfByte == null || paramArrayOfByte.length == 0) {
            log.error("Failed to deserialize, byte is empty");
            throw new RuntimeException("Failed to deserialize");
        }
 
        Schema<T> schema = RuntimeSchema.getSchema(targetClass);
        List<T> result;
        try {
            result = ProtostuffIOUtil.parseListFrom(new ByteArrayInputStream(paramArrayOfByte), schema);
        } catch (IOException e) {
            log.error("Failed to deserialize", e);
            throw new RuntimeException("Failed to deserialize");
        }
        returnresult; }}Copy the code

Do not modify the RedisSerializer component or rewrite the operation

Directly customize the RedisClient tool class method, proxy RedisTemplate tool class method

@Component
public class RedisClient {

    private final RedisTemplate<String, String> redisTemplate;
 
    @Autowired
    public RedisClient(RedisTemplate<String, String> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }
 
    /**
     * get cache
     *
     * @param field
     * @param targetClass
     * @param <T>
     * @return* /
    public <T> T get(final String field, Class<T> targetClass) {
        byte[] result = redisTemplate.execute((RedisCallback<byte[]>) connection -> connection.get(field.getBytes()));
        if (result == null) {
            return null;
        return ProtoStuffUtil.deserialize(result, targetClass);
    }
 
    /**
     * put cache
     *
     * @param field
     * @param obj
     * @param <T>
     * @return* /
    public <T> void set(String field, T obj) {
        final byte[] value = ProtoStuffUtil.serialize(obj);
        redisTemplate.execute((RedisCallback<Void>) connection -> {
            connection.set(field.getBytes(), value);
            return null;
        });
    }
 
    /**
     * put cache with expire time
     *
     * @param field
     * @param obj
     * @paramExpireTime Unit: s *@param <T>
     */
    public <T> void setWithExpire(String field, T obj, final long expireTime) {
        final byte[] value = ProtoStuffUtil.serialize(obj);
        redisTemplate.execute((RedisCallback<Void>) connection -> {
            connection.setEx(field.getBytes(), expireTime, value);
            return null;
        });
    }
 
    /**
     * get list cache
     *
     * @param field
     * @param targetClass
     * @param <T>
     * @return* /
    public <T> List<T> getList(final String field, Class<T> targetClass) {
        byte[] result = redisTemplate.execute((RedisCallback<byte[]>) connection -> connection.get(field.getBytes()));
        if (result == null) {
            return null;
        }
 
        return ProtoStuffUtil.deserializeList(result, targetClass);
    }
 
    /**
     * put list cache
     *
     * @param field
     * @param objList
     * @param <T>
     * @return* /
    public <T> void setList(String field, List<T> objList) {
        final byte[] value = ProtoStuffUtil.serializeList(objList);
        redisTemplate.execute((RedisCallback<Void>) connection -> {
            connection.set(field.getBytes(), value);
            return null;
        });
    }
 
    /**
     * put list cache with expire time
     *
     * @param field
     * @param objList
     * @param expireTime
     * @param <T>
     * @return* /
    public <T> void setListWithExpire(String field, List<T> objList, final long expireTime) {
        final byte[] value = ProtoStuffUtil.serializeList(objList);
        redisTemplate.execute((RedisCallback<Void>) connection -> {
            connection.setEx(field.getBytes(), expireTime, value);
            return null;
        });
    }
 
    /**
     * get h cache
     *
     * @param key
     * @param field
     * @param targetClass
     * @param <T>
     * @return* /
    public <T> T hGet(final String key, final String field, Class<T> targetClass) {
        byte[] result = redisTemplate
                .execute((RedisCallback<byte[]>) connection -> connection.hGet(key.getBytes(), field.getBytes()));
        if (result == null) {
            return null;
        }
 
        return ProtoStuffUtil.deserialize(result, targetClass);
    }
 
    /**
     * put hash cache
     *
     * @param key
     * @param field
     * @param obj
     * @param <T>
     * @return* /
    public <T> boolean hSet(String key, String field, T obj) {
        final byte[] value = ProtoStuffUtil.serialize(obj);
        returnredisTemplate.execute( (RedisCallback<Boolean>) connection -> connection.hSet(key.getBytes(), field.getBytes(), value));  }/**
     * put hash cache
     *
     * @param key
     * @param field
     * @param obj
     * @param <T>
     */
    public <T> void hSetWithExpire(String key, String field, T obj, long expireTime) {
        final byte[] value = ProtoStuffUtil.serialize(obj);
        redisTemplate.execute((RedisCallback<Void>) connection -> {
            connection.hSet(key.getBytes(), field.getBytes(), value);
            connection.expire(key.getBytes(), expireTime);
            return null;
        });
    }
 
    /**
     * get list cache
     *
     * @param key
     * @param field
     * @param targetClass
     * @param <T>
     * @return* /
    public <T> List<T> hGetList(final String key, final String field, Class<T> targetClass) {
        byte[] result = redisTemplate
                .execute((RedisCallback<byte[]>) connection -> connection.hGet(key.getBytes(), field.getBytes()));
        if (result == null) {
            return null;
        }
 
        return ProtoStuffUtil.deserializeList(result, targetClass);
    }
 
    /**
     * put list cache
     *
     * @param key
     * @param field
     * @param objList
     * @param <T>
     * @return* /
    public <T> boolean hSetList(String key, String field, List<T> objList) {
        final byte[] value = ProtoStuffUtil.serializeList(objList);
        returnredisTemplate.execute( (RedisCallback<Boolean>) connection -> connection.hSet(key.getBytes(), field.getBytes(), value));  }/**
     * get cache by keys
     *
     * @param key
     * @param fields
     * @param targetClass
     * @param <T>
     * @return* /
    public <T> Map<String, T> hMGet(String key, Collection<String> fields, Class<T> targetClass) {
        List<byte[]> byteFields = fields.stream().map(String::getBytes).collect(Collectors.toList());
        byte[][] queryFields = new byte[byteFields.size()][];
        byteFields.toArray(queryFields);
        List<byte[]> cache = redisTemplate
                .execute((RedisCallback<List<byte[]>>) connection -> connection.hMGet(key.getBytes(), queryFields));
 
        Map<String, T> results = new HashMap<>(16);
        Iterator<String> it = fields.iterator();
        int index = 0;
        while (it.hasNext()) {
            String k = it.next();
            if (cache.get(index) == null) {
                index++;
                continue;
            }
 
            results.put(k, ProtoStuffUtil.deserialize(cache.get(index), targetClass));
            index++;
        }
 
        return results;
    }
 
    /**
     * set cache by keys
     *
     * @param field
     * @param values
     * @param <T>
     */
    public <T> void hMSet(String field, Map<String, T> values) {
        Map<byte[].byte[]> byteValues = new HashMap<>(16);
        for (Map.Entry<String, T> value : values.entrySet()) {
            byteValues.put(value.getKey().getBytes(), ProtoStuffUtil.serialize(value.getValue()));
        }
 
        redisTemplate.execute((RedisCallback<Void>) connection -> {
            connection.hMSet(field.getBytes(), byteValues);
            return null;
        });
    }
 
    /**
     * get caches in hash
     *
     * @param key
     * @param targetClass
     * @param <T>
     * @return* /
    public <T> Map<String, T> hGetAll(String key, Class<T> targetClass) {
        Map<byte[].byte[]> records = redisTemplate
                .execute((RedisCallback<Map<byte[].byte[]>>) connection -> connection.hGetAll(key.getBytes()));
 
        Map<String, T> ret = new HashMap<>(16);
        for (Map.Entry<byte[].byte[]> record : records.entrySet()) {
            T obj = ProtoStuffUtil.deserialize(record.getValue(), targetClass);
            ret.put(new String(record.getKey()), obj);
        }
 
        return ret;
    }
 
    /**
     * list index
     *
     * @param key
     * @param index
     * @param targetClass
     * @param <T>
     * @return* /
    public <T> T lIndex(String key, int index, Class<T> targetClass) {
        byte[] value =
                redisTemplate.execute((RedisCallback<byte[]>) connection -> connection.lIndex(key.getBytes(), index));
        return ProtoStuffUtil.deserialize(value, targetClass);
    }
 
    /**
     * list range
     *
     * @param key
     * @param start
     * @param end
     * @param targetClass
     * @param <T>
     * @return* /
    public <T> List<T> lRange(String key, int start, int end, Class<T> targetClass) {
        List<byte[]> value = redisTemplate
                .execute((RedisCallback<List<byte[]>>) connection -> connection.lRange(key.getBytes(), start, end));
        return value.stream().map(record -> ProtoStuffUtil.deserialize(record, targetClass))
                .collect(Collectors.toList());
    }
 
    /**
     * list left push
     *
     * @param key
     * @param obj
     * @param <T>
     */
    public <T> void lPush(String key, T obj) {
        final byte[] value = ProtoStuffUtil.serialize(obj);
        redisTemplate.execute((RedisCallback<Long>) connection -> connection.lPush(key.getBytes(), value));
    }
 
    /**
     * list left push
     *
     * @param key
     * @param objList
     * @param <T>
     */
    public <T> void lPush(String key, List<T> objList) {
        List<byte[]> byteFields = objList.stream().map(ProtoStuffUtil::serialize).collect(Collectors.toList());
        byte[][] values = new byte[byteFields.size()][];
 
        redisTemplate.execute((RedisCallback<Long>) connection -> connection.lPush(key.getBytes(), values));
    }
 
    /** * Delete key ** exactly@param key
     */
    public void deleteCache(String key) {
        redisTemplate.delete(key);
    }
 
 
    /** ** ** ** ** ** ** ** **@param redisKey
     * @param immutablePair
     */
    public void zAdd(String redisKey, ImmutablePair<String, BigDecimal> immutablePair) {
        final byte[] key = redisKey.getBytes();
        final byte[] value = immutablePair.getLeft().getBytes();
        redisTemplate.execute((RedisCallback<Boolean>) connection -> connection
                .zAdd(key, immutablePair.getRight().doubleValue(), value));
 
    }
 
    /** * get low -> high ranking **@paramRedisKey The category to sort by *@param start
     * @param end
     * @return* /
    public List<ImmutablePair<String, BigDecimal>> zRangeWithScores(String redisKey, int start, int end) {
        Set<RedisZSetCommands.Tuple> items = redisTemplate.execute(
                (RedisCallback<Set<RedisZSetCommands.Tuple>>) connection -> connection
                        .zRangeWithScores(redisKey.getBytes(), start, end));
        return items.stream()
                .map(record -> ImmutablePair.of(new String(record.getValue()), BigDecimal.valueOf(record.getScore())))
                .collect(Collectors.toList());
    }
 
 
    /** * get the highest ranking -> lower ranking **@paramRedisKey The category to sort by *@param start
     * @param end
     * @return* /
    public List<ImmutablePair<String, BigDecimal>> zRevRangeWithScores(String redisKey, int start, int end) {
        Set<RedisZSetCommands.Tuple> items = redisTemplate.execute(
                (RedisCallback<Set<RedisZSetCommands.Tuple>>) connection -> connection
                        .zRevRangeWithScores(redisKey.getBytes(), start, end));
        return items.stream()
                .map(record -> ImmutablePair.of(newString(record.getValue()), BigDecimal.valueOf(record.getScore()))) .collect(Collectors.toList()); }}Copy the code

The resources

www.cnblogs.com/hntyzgn/p/7…