related redis/lettuce#1254

JacksonRedisCodec.java

import com.fasterxml.jackson.annotation.*;
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.databind.*;
import com.fasterxml.jackson.databind.jsontype.TypeResolverBuilder;
import io.lettuce.core.codec.RedisCodec;
import io.lettuce.core.codec.Utf8StringCodec;

import javax.xml.datatype.XMLGregorianCalendar;
import java.io.IOException;
import java.nio.ByteBuffer;

public class JacksonRedisCodec<V> implements RedisCodec<String, V> {

    private static final byte[] EMPTY = new byte[0];

    private final RedisCodec<String, String> keyCodec = new Utf8StringCodec();

    private final ObjectMapper mapObjectMapper;

    public JacksonRedisCodec() {
        this.mapObjectMapper = new ObjectMapper();
        init(this.mapObjectMapper);
        initTypeInclusion(this.mapObjectMapper);
    }

    private void init(ObjectMapper objectMapper) {
        objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
        objectMapper.setVisibility(objectMapper.getSerializationConfig()
                .getDefaultVisibilityChecker()
                .withFieldVisibility(JsonAutoDetect.Visibility.ANY)
                .withGetterVisibility(JsonAutoDetect.Visibility.NONE)
                .withSetterVisibility(JsonAutoDetect.Visibility.NONE)
                .withCreatorVisibility(JsonAutoDetect.Visibility.NONE));
        objectMapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
        objectMapper.enable(JsonGenerator.Feature.WRITE_BIGDECIMAL_AS_PLAIN);
        objectMapper.disable(SerializationFeature.FAIL_ON_EMPTY_BEANS);
        objectMapper.enable(MapperFeature.SORT_PROPERTIES_ALPHABETICALLY);
        objectMapper.addMixIn(Throwable.class, ThrowableMixIn.class);
    }

    private void initTypeInclusion(ObjectMapper mapObjectMapper) {
        TypeResolverBuilder<?> mapTyper = new ObjectMapper.DefaultTypeResolverBuilder(ObjectMapper.DefaultTyping.NON_FINAL) {
            @Override
            public boolean useForType(JavaType t) {
                switch (_appliesFor) {
                    case NON_CONCRETE_AND_ARRAYS:
                        while (t.isArrayType()) {
                            t = t.getContentType();
                        }
                        // fall through
                    case OBJECT_AND_NON_CONCRETE:
                        return (t.getRawClass() == Object.class) || !t.isConcrete();
                    case NON_FINAL:
                        while (t.isArrayType()) {
                            t = t.getContentType();
                        }
                        // to fix problem with wrong long to int conversion
                        if (t.getRawClass() == Long.class) {
                            return true;
                        }
                        if (t.getRawClass() == XMLGregorianCalendar.class) {
                            return false;
                        }
                        return !t.isFinal(); // includes Object.class
                    default:
                        // case JAVA_LANG_OBJECT:
                        return t.getRawClass() == Object.class;
                }
            }
        };
        mapTyper.init(JsonTypeInfo.Id.CLASS, null);
        mapTyper.inclusion(JsonTypeInfo.As.PROPERTY);
        mapObjectMapper.setDefaultTyping(mapTyper);

        // warm up codec
        try {
            byte[] s = mapObjectMapper.writeValueAsBytes(1);
            mapObjectMapper.readValue(s, Object.class);
        } catch (IOException e) {
            throw new IllegalStateException(e);
        }
    }

    @Override
    public String decodeKey(ByteBuffer bytes) {
        return keyCodec.decodeKey(bytes);
    }

    @Override
    public V decodeValue(ByteBuffer bytes) {
        if (!bytes.hasRemaining()) {
            return null;
        }
        try {
            return (V) mapObjectMapper.readValue(bytes.array(), Object.class);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public ByteBuffer encodeKey(String key) {
        return keyCodec.encodeKey(key);
    }

    @Override
    public ByteBuffer encodeValue(V value) {
        if (value == null) {
            return ByteBuffer.wrap(EMPTY);
        }
        try {
            byte[] bytes = mapObjectMapper.writeValueAsBytes(value);
            return ByteBuffer.wrap(bytes);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @JsonIdentityInfo(generator= ObjectIdGenerators.IntSequenceGenerator.class, property="@id")
    @JsonAutoDetect(fieldVisibility = JsonAutoDetect.Visibility.ANY,
            getterVisibility = JsonAutoDetect.Visibility.PUBLIC_ONLY,
            setterVisibility = JsonAutoDetect.Visibility.NONE,
            isGetterVisibility = JsonAutoDetect.Visibility.NONE)
    public static class ThrowableMixIn {

    }
}

Lz4RedisCodec.java

import io.lettuce.core.codec.RedisCodec;
import net.jpountz.lz4.LZ4Compressor;
import net.jpountz.lz4.LZ4Factory;
import net.jpountz.lz4.LZ4FastDecompressor;

import java.nio.ByteBuffer;

public class Lz4RedisCodec<V> implements RedisCodec<String, V> {

    private static final int DECOMPRESSION_HEADER_SIZE = Integer.SIZE / 8;

    private final LZ4Factory factory = LZ4Factory.fastestInstance();

    private RedisCodec<String, V> innerCodec;    // an instance of JacksonRedisCodec

    public Lz4RedisCodec(RedisCodec<String, V> innerCodec) {
        this.innerCodec = innerCodec;
    }

    @Override
    public String decodeKey(ByteBuffer bytes) {
        return innerCodec.decodeKey(bytes);
    }

    @Override
    public V decodeValue(ByteBuffer src) {
        if (!src.hasRemaining()) {
            return null;
        }
        int decompressSize = src.getInt();
        LZ4FastDecompressor decompressor = factory.fastDecompressor();
        ByteBuffer dest = ByteBuffer.allocate(decompressSize);
        dest.mark();
        decompressor.decompress(src, dest);
        dest.reset();
        return innerCodec.decodeValue(dest);
    }

    @Override
    public ByteBuffer encodeKey(String key) {
        return innerCodec.encodeKey(key);
    }

    @Override
    public ByteBuffer encodeValue(V value) {
        ByteBuffer src = innerCodec.encodeValue(value);
        int length = src.remaining();
        LZ4Compressor compressor = factory.fastCompressor();
        ByteBuffer dest = ByteBuffer.allocate(compressor.maxCompressedLength(length) + DECOMPRESSION_HEADER_SIZE);
        dest.mark();
        dest.putInt(length);
        compressor.compress(src, dest);
        dest.limit(dest.position());
        dest.reset();
        return dest;
    }
}

你好，看到你提的issue了，请问代码哪一行会导致编码异常 ？谢谢

ByteBuffer encodeValue(V value);

若value中包含Collection类型字段，因为是使用的异步api例如setAsync，如果调用方在调用后立即对value中collection字段进行修改，可能会造成ConcurrentModificationException。

: )