Redis 工具类与 Redis 布隆过滤器

Redis 工具类

1. 核心依赖

<!--redis-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>31.1-jre</version>
</dependency>
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>fastjson</artifactId>
    <version>1.2.31</version>
</dependency>

2. 序列化

public class FastJsonRedisSerializer<T> implements RedisSerializer<T> {
 
    public static final Charset DEFAULT_CHARSET = Charset.forName("UTF-8");
 
    private Class<T> clazz;
 
    static {
        ParserConfig.getGlobalInstance().setAutoTypeSupport(true);
    }
 
    public FastJsonRedisSerializer(Class<T> clazz) {
        super();
        this.clazz = clazz;
    }
 
    @Override
    public byte[] serialize(T t) throws SerializationException {
        if (t == null) {
            return new byte[0];
        }
        return JSON.toJSONString(t, SerializerFeature.WriteClassName).getBytes(DEFAULT_CHARSET);
    }
 
    @Override
    public T deserialize(byte[] bytes) throws SerializationException {
        if (bytes == null || bytes.length <= 0) {
            return null;
        }
        String str = new String(bytes, DEFAULT_CHARSET);
        return JSON.parseObject(str, clazz);
    }

    protected JavaType getJavaType(Class<?> clazz)
    {
        return TypeFactory.defaultInstance().constructType(clazz);
    }
}

@Configuration
public class RedisSerializeConfig {

    @Bean
    @SuppressWarnings(value = { "unchecked", "rawtypes" })
    public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
        RedisTemplate<Object, Object> template = new RedisTemplate<Object, Object>();
        template.setConnectionFactory(connectionFactory);
        FastJsonRedisSerializer serializer = new FastJsonRedisSerializer(Object.class);
        // 使用StringRedisSerializer来序列化和反序列化redis的key值
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(serializer);
        // Hash的key也采用StringRedisSerializer的序列化方式
        template.setHashKeySerializer(new StringRedisSerializer());
        template.setHashValueSerializer(serializer);
        template.afterPropertiesSet();
        return template;
    }
}

3. 布隆过滤器

/**
 * 算法过程：
 * 1. 首先需要k个hash函数，每个函数可以把key散列成为1个整数
 * 2. 初始化时，需要一个长度为n比特的数组，每个比特位初始化为0
 * 3. 某个key加入集合时，用k个hash函数计算出k个散列值，并把数组中对应的比特位置为1
 * 4. 判断某个key是否在集合时，用k个hash函数计算出k个散列值，并查询数组中对应的比特位，如果所有的比特位都是1，认为在集合中。
 **/
@Component
public class BloomFilterHelper<T> {

    private int numHashFunctions;

    private int bitSize;

    private Funnel<T> funnel;

    private static final int NUM_BITS = (int) 1e4;
    private static final double RATE = 0.03;//不存在误判为存在的概率

    private static void funnel(@Nullable Object o, PrimitiveSink primitiveSink) {
        primitiveSink.putBytes(o.toString().getBytes());
    }

    public BloomFilterHelper() {
        this((Funnel) BloomFilterHelper::funnel, NUM_BITS, RATE);
    }
    public BloomFilterHelper(Funnel<T> funnel, int expectedInsertions, double fpp) {
        this.funnel = funnel;
        // 计算bit数组长度
        bitSize = optimalNumOfBits(expectedInsertions, fpp);
        // 计算hash方法执行次数
        numHashFunctions = optimalNumOfHashFunctions(expectedInsertions, bitSize);
    }

    public int[] getHashOffset() {
        return new int[numHashFunctions];
    }

    public int[] murmurHashOffset(T value) {
        int[] offset = new int[numHashFunctions];

        long hash64 = Hashing.murmur3_128().hashObject(value, funnel).asLong();
        int hash1 = (int) hash64;
        int hash2 = (int) (hash64 >>> 32);
        for (int i = 1; i <= numHashFunctions; i++) {
            int nextHash = hash1 + i * hash2;
            if (nextHash < 0) {
                nextHash = ~nextHash;
            }
            offset[i - 1] = nextHash % bitSize;
        }
        return offset;
    }

    /**
     * 计算bit数组长度
     */
    private int optimalNumOfBits(long n, double p) {
        if (p == 0) {
            // 设定最小期望长度
            p = Double.MIN_VALUE;
        }
        return (int) (-n * Math.log(p) / (Math.log(2) * Math.log(2)));
    }

    /**
     * 计算hash方法执行次数
     */
    private int optimalNumOfHashFunctions(long n, long m) {
        return Math.max(1, (int) Math.round((double) m / n * Math.log(2)));
    }
}

@Component
@RequiredArgsConstructor
public class RedisBloomFilter {

    private final RedisTemplate redisTemplate;

    private final BloomFilterHelper bloomFilterHelper;

    public void init(String bloomFilterName) {
        int[] offset = bloomFilterHelper.getHashOffset();
        for (int i : offset) {
            redisTemplate.opsForValue().setBit(bloomFilterName, i, true);
        }
    }

    /**
     * 根据给定的布隆过滤器添加值
     */
    public <T> void add(String bloomFilterName, T value) {
        int[] offset = bloomFilterHelper.murmurHashOffset(value);
        for (int i : offset) {
            redisTemplate.opsForValue().setBit(bloomFilterName, i, true);
        }
    }

    /**
     * 根据给定的布隆过滤器判断值是否存在
     */
    public <T> boolean contains(String bloomFilterName, T value) {
        int[] offset = bloomFilterHelper.murmurHashOffset(value);
        for (int i : offset) {
            if (!redisTemplate.opsForValue().getBit(bloomFilterName, i)) {
                return false;
            }
        }
        return true;
    }

}

4. Redis工具类

@Component
@RequiredArgsConstructor
@Slf4j
@SuppressWarnings(value = { "unchecked", "rawtypes" })
public class RedisCache {

    private final RedisTemplate redisTemplate;

    private final RedisBloomFilter redisBloomFilter;

    /**
     * 设置有效时间
     *
     * @param key Redis键
     * @param timeout 超时时间
     * @return true=设置成功；false=设置失败
     */
    public Boolean expire(final String key, final long timeout, final TimeUnit timeUnit) {
        log.info("为 Redis 的键值设置超时时间\t[{}]-[{}  {}]", key, timeout, timeUnit.name());
        return redisTemplate.expire(key, timeout, timeUnit);
    }

    /**
     * 原子设置过期时间
     * @param key
     * @param value
     * @param timeout
     */
    public <T> void execute(final String key, final T value, final long timeout, final TimeUnit timeUnit) {
        log.info("尝试存入 Redis\t[{}]-[{}]，超时时间:[{}  {}]", key, value, timeout, timeUnit.name());
        redisTemplate.execute(new SessionCallback() {
            @Override
            public Object execute(RedisOperations redisOperations) throws DataAccessException {
                redisOperations.multi();
                redisOperations.opsForValue().set(key, value);
                redisOperations.expire(key, timeout, timeUnit);
                return redisOperations.exec();
            }
        });
    }

    /**
     *  获得对象的剩余存活时间
     * @param key 键
     * @return 剩余存活时间
     */
    public long getKeyTTL(final String key, final TimeUnit timeUnit) {
        int ttl = Math.toIntExact(redisTemplate.opsForValue().getOperations().getExpire(key));
        String message = null;
        switch (ttl) {
            case -1:
                message = "没有设置过期时间";
                break;
            case -2:
                message = "key不存在";
                break;
            default:
                message = ttl + "  " + TimeUnit.SECONDS.name();
                break;
        }
        log.info("查询 Redis key[{}] 剩余存活时间:{}", key, message);
        return TimeUnit.SECONDS.convert(ttl, timeUnit);
    }

    /**
     * 缓存基本的对象，Integer、String、实体类等
     *
     * @param key 缓存的键值
     * @param value 缓存的值
     */
    public <T> void setCacheObject(final String key, final T value) {
        log.info("存入 Redis\t[{}]-[{}]", key, value);
        redisTemplate.opsForValue().set(key, value);
    }

    /**
     * 缓存基本的对象，Integer、String、实体类等
     *
     * @param key 缓存的键值
     * @param value 缓存的值
     * @param timout 超时时间
     */
    public <T> void setCacheObject(final String key, final T value, final long timout, final TimeUnit timeUnit) {
        log.info("存入 Redis\t[{}]-[{}]，超时时间:[{}  {}]", key, value, timout, timeUnit.name());
        redisTemplate.opsForValue().set(key, value, timout, timeUnit);
    }

    /**
     * 获取键值
     * @param key 键
     * @return 键对应的值，并封装成 Optional 对象
     * @param <T>
     */
    public <T> Optional<T> getCacheObject(final String key) {
        T value = (T) redisTemplate.opsForValue().get(key);
        log.info("查询 Redis\t[{}]-[{}]", key, value);
        return Optional.ofNullable(value);
    }

    /**
     * 让指定 Redis 键值进行自减
     * @param key 键
     * @return 自减后的值
     */
    public long decrementCacheNumber(final String key) {
        long number = redisTemplate.opsForValue().decrement(key);
        log.info("Redis key[{}] 自减后：{}", key, number);
        return number;
    }

    /**
     * 让指定 Redis 键值进行自增
     * @param key 键
     * @return 自增后的值
     */
    public long incrementCacheNumber(final String key) {
        long number = redisTemplate.opsForValue().increment(key);
        log.info("Redis key[{}] 自增后：{}", key, number);
        return number;
    }

    /**
     * 初始化布隆过滤器
     * @param bloomFilterName
     */
    public void initBloomFilter(final String bloomFilterName) {
        log.info("初始化布隆过滤器[{}]", bloomFilterName);
        redisTemplate.execute(new SessionCallback() {
            @Override
            public Object execute(RedisOperations redisOperations) throws DataAccessException {
                redisOperations.multi();
                redisBloomFilter.init(bloomFilterName);
                return redisOperations.exec();
            }
        });
    }

    /**
     * 初始化布隆过滤器
     * @param bloomFilterName
     * @param timeout
     * @param timeUnit
     */
    public void initBloomFilter(final String bloomFilterName, final long timeout, final TimeUnit timeUnit) {
        redisTemplate.execute(new SessionCallback() {
            @Override
            public Object execute(RedisOperations redisOperations) throws DataAccessException {
                redisOperations.multi();
                redisBloomFilter.init(bloomFilterName);
                expire(bloomFilterName, timeout, timeUnit);
                return redisOperations.exec();
            }
        });
    }

    /**
     * 加入布隆过滤器
     * @param bloomFilterName 隆过滤器的名字
     * @param key key 键
     */
    public <T> void addToBloomFilter(final String bloomFilterName, final T key) {
        log.info("加入布隆过滤器[{}]\tkey[{}]", bloomFilterName, key);
        redisTemplate.execute(new SessionCallback() {
            @Override
            public Object execute(RedisOperations redisOperations) throws DataAccessException {
                redisOperations.multi();
                redisBloomFilter.add(bloomFilterName, key);
                return redisOperations.exec();
            }
        });
    }

    /**
     * 布隆过滤器是否存在该键值
     * @param bloomFilterName 布隆过滤器的名字
     * @param key 键
     * @return 键是否存在
     */
    public <T> boolean containsInBloomFilter(final String bloomFilterName, final T key) {
        boolean flag = redisBloomFilter.contains(bloomFilterName, key);
        log.info("key[{}]\t是否存在于布隆过滤器[{}]:\t{}", key, bloomFilterName, flag);
        return flag;
    }

    /**
     * 缓存Map
     *
     * @param key
     * @param data
     */
    public <K, T> void setCacheMap(final String key, final Map<K, T> data) {
        if (Objects.nonNull(data)) {
            log.info("Map 存入 Redis\t[{}]-[{}]", key, data);
            redisTemplate.opsForHash().putAll(key, data);
        }
    }

    /**
     * 缓存Map
     *
     * @param key
     * @param data
     */
    public <K, T> void setCacheMap(final String key, final Map<K, T> data, long timeout, final TimeUnit timeUnit) {
        if (Objects.nonNull(data)) {
            Map<String, T> map = new HashMap<>();
            data.entrySet().stream().parallel().forEach(entry -> {
                map.put(entry.getKey().toString(), entry.getValue());
            });
            log.info("尝试存入 Redis\t[{}]-[{}] 超时时间:[{}  {}]", key, map, timeout, timeUnit.name());
            redisTemplate.execute(new SessionCallback() {
                @Override
                public Object execute(RedisOperations redisOperations) throws DataAccessException {
                    redisOperations.multi();
                    redisTemplate.opsForHash().putAll(key, map);
                    expire(key, timeout, timeUnit);
                    return redisOperations.exec();
                }
            });
        }
    }

    /**
     * 获得缓存的Map
     *
     * @param key
     * @return
     */
    public <K, T> Optional<Map<K, T>> getCacheMap(final String key) {
        Map<K, T> data = redisTemplate.opsForHash().entries(key);
        data = data.size() == 0 ? null: data;
        log.info("获取 Redis 中的 Map 缓存\t[{}]-[{}]", key, data);
        return Optional.ofNullable(data);
    }

    /**
     * 往Hash中存入数据
     *
     * @param key Redis键
     * @param hashKey Hash键
     * @param value 值
     */
    public <K, T> void setCacheMapValue(final String key, final K hashKey, final T value) {
        log.info("存入 Redis 的某个 Map\t[{}.{}]-[{}]", key, hashKey, value);
        redisTemplate.opsForHash().put(key, hashKey.toString(), value);
    }

    /**
     * 获取Hash中的数据
     *
     * @param key Redis键
     * @param hashKey Hash键
     * @return Hash中的对象
     */
    public <K, T> Optional<T> getCacheMapValue(final String key, final K hashKey) {
        T value = (T) redisTemplate.opsForHash().get(key, hashKey.toString());
        log.info("获取 Redis 中的 Map 的键值\t[{}.{}]-[{}]", key, hashKey, value);
        return Optional.ofNullable(value);
    }

    /**
     * 删除Hash中的数据
     *
     * @param key
     * @param hashKey
     */
    public <K> void delCacheMapValue(final String key, final K hashKey) {
        log.info("删除 Redis 中的 Map 的键值\tkey[{}.{}]", key, hashKey);
        redisTemplate.opsForHash().delete(key, hashKey.toString());
    }

    /**
     * 让指定 HashMap 的键值进行自减
     * @param key HashMap的名字
     * @param hashKey HashMap的一个键
     * @return 自减后的值
     */
    public <K> long decrementCacheMapNumber(final String key, final K hashKey) {
        long number = redisTemplate.opsForHash().increment(key, hashKey.toString(), -1);
        log.info("Redis key[{}.{}] 自减后：{}", key, hashKey, number);
        return number;
    }

    /**
     * 让指定 HashMap 的键值进行自增
     * @param key HashMap的名字
     * @param hashKey HashMap的一个键
     * @return 自增后的值
     */
    public <K> long incrementCacheMapNumber(final String key, final K hashKey) {
        long number = redisTemplate.opsForHash().increment(key, hashKey.toString(), +1);
        log.info("Redis key[{}.{}] 自增后：{}", key, hashKey, number);
        return number;
    }

    /**
     * 删除单个对象
     * @param key
     */
    public boolean deleteObject(final String key) {
        log.info("删除 Redis 的键值\tkey[{}]", key);
        return redisTemplate.delete(key);
    }

}

5. 查询Redis与Redis设置缓存的技巧

伪代码：

redisCache.getCacheObject(redisKey)
    .orElseGet(() -> {
        /* balabala ....*/
        // 可能查询有误，所以这里也可能没法获取到data（为null），也就可能没必要设置缓存
        redisCache.setCacheObject(redisKey, data, ttl, unit); 
        /* balabala ....*/
        return data;
    }));