Redis 的 Java 客户端（Jedis、SpringDataRedis、SpringCache、Redisson）基本操作指南

╰半橙微兮° 2023-09-23 17:11 60阅读 0赞

# Jedis #

参考：

*  [Jedic 官方文档][Jedic]
 *  [Jedis 托管在 github 上的源码][Jedis _ github]

Redis 不仅是使用命令来操作，现在基本上主流的语言都有客户端支持，比如 java、C、C\#、C++、php、Node.js、Go 等。在官方网站里列一些 Java 的客户端，有 Jedis、Redisson、Jredis、JDBC-Redis 等，其中官方推荐使用 Jedis 和 Redisson。

在企业中用的最多的就是 Jedis。Jedis 基本上实现了所有的 Redis 命令，并且还支持连接池、集群等高级的用法，而且使用简单，使得在 Java 中使用 Redis 服务将变得非常的简单。

## 依赖、常用API ##

**依赖**

<dependency> 
        <groupId>redis.clients</groupId> 
        <artifactId>jedis</artifactId> 
        <version>2.9.0</version> 
    </dependency>

**jedis 常用API**

// 创建jedis对象，参数host是redis服务器地址，参数port是redis服务端口
    new Jedis(host, port) 
    // 释放资源
    public void close()
    
    // 设置字符串类型的数据
    public String set(String key, String value)
    // 获得字符串类型的数据
    public String get(String key)
    // 删除指定的key
    public Long del(String... keys)
    // 设置哈希类型的数据
    public Long hset(String key, String field, String value)
    // 获得哈希类型的数据
    public String hget(String key, String field)
    // 设置列表类型的数据
    public Long rpush(String key, String... strings)
    public Long lpush(String key, String... strings)
    // 列表左面弹栈
    public String lpop(String key)
    // 列表右面弹栈
    public String rpop(String key)

## jedis 连接池 ##

jedis 连接资源的创建与销毁是很消耗程序性能，所以 jedis 提供了 jedis 的池化技术，jedisPool 在创建时初始化一些连接资源存储到连接池中，使用 jedis 连接资源时不需要创建，而是从连接池中获取一个资源进行 redis 的操作，使用完毕后，不需要销毁该 jedis 连接资源，而是将该资源归还给连接池，供其他请求使用。

**JedisUtils 工具类的封装：**

import redis.clients.jedis.Jedis;
    import redis.clients.jedis.JedisPool;
    import redis.clients.jedis.JedisPoolConfig;
    
    public class JedisUtils {
        
    
        private static JedisPool jedisPool =null;
        
        static {
        
            // 创建jedis连接池的配置对象
            JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
            // 连接池初始化的最大连接数
            jedisPoolConfig.setMaxTotal(40);
            // 最大空闲连接数
            jedisPoolConfig.setMaxIdle(10);
            // 当池内没有可用连接时，最大等待时间
            jedisPoolConfig.setMaxWaitMillis(10000);
            // 创建jedis的连接池
            jedisPool = new JedisPool(jedisPoolConfig, "localhost", 6379);
        }
    
        //获取jedis
        public static Jedis getJedis() {
        
            // 从连接池中获取jedis
            Jedis jedis = jedisPool.getResource();
            return jedis;
        }
    }

# Spring Data Redis #

## 概述、依赖 ##

[官网][Link 1]

Spring Data Redis 是 Spring Data 家族的一部分。 对 Jedis 客户端进行了封装，与 spring 进行了整合。可以非常方便的来实现 redis 的配置和操作。

*  当 Redis 当做数据库或者消息队列来操作时，一般使用 RedisTemplate 工具类来操作
    
    redisTemplate 是 Spring 集成 Redis，操作 redis 的专门工具类
 *  当 Redis 作为缓存使用时，可以将它作为 Spring Cache 的实现，直接通过注解使用
    
    Spring Cache 是 Spring 提供的一整套的缓存解决方案，提供了一整套的接口和代码规范、配置、注解等，它不是具体的缓存实现，具体实现由各自的第三方自己实现。比如 Guava，EhCache，Redis，本地缓存等。

**Spring Boot 整合 Redis 自动配置原理**

![在这里插入图片描述][823bce51588a4b9a8f0321a62455b168.png_pic_center]

**依赖**

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>

## 配置文件及配置类 ##

**单机版yml配置**

spring:
      redis:
        # 连接模式（自定义的配置项）
        model: standalone
        # 配置redis地址（单机模式）
        host: 192.168.85.135
        # 配置redis端口（单机模式）
        port: 6379
        # 库。不配置默认为0
        database: 0
        # 密码。不配置默认无密码
        password: passwd@123
        pool:
          # 连接池初始化的最大连接数
          max-active: 8
          # 最大空闲连接数
          max-idle: 8
          # 最小空闲连接数
          min-idle: 0
          # 当池内没有可用连接时，最大等待时间。-1 为一直等待
          max-wait: -1
        # 集群模式
        cluster:
          nodes: ip1:6379,ip2:6379,ip3:6379
          # 最大重试次数
          max-redirects: 6
        # 哨兵模式
        sentinel:
          nodes: ip1:6379,ip2:6379,ip3:6379
          master: mymaster

**redis 连接配置类**

import com.fasterxml.jackson.annotation.JsonAutoDetect;
    import com.fasterxml.jackson.annotation.PropertyAccessor;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.data.redis.connection.RedisClusterConfiguration;
    import org.springframework.data.redis.connection.RedisNode;
    import org.springframework.data.redis.connection.RedisSentinelConfiguration;
    import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
    import org.springframework.data.redis.connection.jedis.JedisClientConfiguration;
    import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
    import org.springframework.data.redis.core.RedisTemplate;
    import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
    import org.springframework.data.redis.serializer.StringRedisSerializer;
    import redis.clients.jedis.JedisPoolConfig;
    import java.util.List;
    import java.util.Set;
    import java.util.stream.Collectors;
    
    @Configuration
    public class RedisConfig {
        
    
        /**
         * redis 连接模式
         */
        @Value("${spring.redis.model:standalone}")
        private String model;
    
        @Bean
        public JedisPoolConfig jedisPoolConfig(RedisProperties properties){
        
            // 创建jedis连接池的配置对象
            JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
            RedisProperties.Pool pool = properties.getJedis().getPool();
            if (pool == null) pool = new RedisProperties.Pool();
            // 连接池初始化的最大连接数
            jedisPoolConfig.setMaxTotal(pool.getMaxActive());
            // 最大空闲连接数
            jedisPoolConfig.setMaxIdle(pool.getMaxIdle());
            // 最小空闲连接数
            jedisPoolConfig.setMaxIdle(pool.getMinIdle());
            // 当池内没有可用连接时，最大等待时间
            jedisPoolConfig.setMaxWaitMillis(pool.getMaxWait().toMillis());
            return jedisPoolConfig;
        }
        
        /**
         * redis 连接
         */
        @Bean
        public JedisConnectionFactory jedisConnectionFactory(RedisProperties properties, JedisPoolConfig jedisPoolConfig) {
        
            // 设置jedis连接池
            JedisClientConfiguration.JedisClientConfigurationBuilder jedisConfBuilder = JedisClientConfiguration.builder();
            JedisClientConfiguration jedisClientConfiguration = jedisConfBuilder.usePooling().poolConfig(jedisPoolConfig).build();
    
            if ("sentinel".equalsIgnoreCase(model)){
        
                // 哨兵模式连接
                List<String> serverList = properties.getSentinel().getNodes();
                Set<RedisNode> nodes = serverList.stream().map(ipPortStr -> {
        
                    String[] ipPortArr = ipPortStr.split(":");
                    return new RedisNode(ipPortArr[0].trim(), Integer.parseInt(ipPortArr[1]));
                }).collect(Collectors.toSet());
    
                RedisSentinelConfiguration redisConfig = new RedisSentinelConfiguration();
                redisConfig.setSentinels(nodes);
                redisConfig.setDatabase(properties.getDatabase());
                redisConfig.setMaster(properties.getSentinel().getMaster());
                redisConfig.setPassword(properties.getPassword());
                return new JedisConnectionFactory(redisConfig, jedisClientConfiguration);
            } else if ("cluster".equalsIgnoreCase(model)) {
        
                // 集群模式连接
                List<String> serverList = properties.getCluster().getNodes();
                Set<RedisNode> nodes = serverList.stream().map(ipPortStr -> {
        
                    String[] ipPortArr = ipPortStr.split(":");
                    return new RedisNode(ipPortArr[0].trim(), Integer.parseInt(ipPortArr[1]));
                }).collect(Collectors.toSet());
    
                RedisClusterConfiguration redisConfig = new RedisClusterConfiguration();
                redisConfig.setClusterNodes(nodes);
                redisConfig.setMaxRedirects(properties.getCluster().getMaxRedirects());
                redisConfig.setPassword(properties.getPassword());
                return new JedisConnectionFactory(redisConfig, jedisClientConfiguration);
            } else {
        
                // 单节点模式连接
                RedisStandaloneConfiguration redisConfig = new RedisStandaloneConfiguration();
                redisConfig.setHostName(properties.getHost());
                redisConfig.setPort(properties.getPort());
                redisConfig.setDatabase(properties.getDatabase());
                redisConfig.setPassword(properties.getPassword());
                return new JedisConnectionFactory(redisConfig, jedisClientConfiguration);
            }
        }
    
        /**
         * RedisTemplate 配置
         */
        @Bean
        public RedisTemplate<String, Object> redisTemplate(JedisConnectionFactory jedisConnectionFactory) {
        
            // 设置序列化。redisTemplate序列化默认使用的jdkSerializeable，存储二进制字节码，导致key会出现乱码，所以自定义
            Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
            ObjectMapper om = new ObjectMapper();
            om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
            // 指定序列化输入的类型，类必须是非final修饰的，final修饰的类，比如String, Integer等会抛出异常
            om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
            jackson2JsonRedisSerializer.setObjectMapper(om);
    
            // 配置redisTemplate
            RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
            redisTemplate.setConnectionFactory(jedisConnectionFactory);
            StringRedisSerializer stringSerializer = new StringRedisSerializer();
            // key序列化
            redisTemplate.setKeySerializer(stringSerializer);
            // value序列化
            redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
            // Hash key序列化
            redisTemplate.setHashKeySerializer(stringSerializer);
            // Hash value序列化
            redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);
            redisTemplate.afterPropertiesSet();
            return redisTemplate;
        }

## RedisTemplate ##

Spring Data 为 Redis 提供了一个工具类：RedisTemplate。里面封装了对于 Redis 的五种数据结构的各种操作，包括：

*  redisTemplate.opsForValue() ：操作字符串
 *  redisTemplate.opsForHash() ：操作 hash
 *  redisTemplate.opsForList()：操作 list
 *  redisTemplate.opsForSet()：操作 set
 *  redisTemplate.opsForZSet()：操作 zset

一些通用命令，如 del，可以通过 redisTemplate.xx() 来直接调用

### StringRedisTemplate ###

RedisTemplate 在创建时，可以指定其泛型类型：

*  K ：代表 key 的数据类型
 *  V ：代表 value 的数据类型

注意：这里的类型不是 Redis 中存储的数据类型，而是 Java 中的数据类型，RedisTemplate 会自动将 Java 类型转为 Redis 支持的数据类型：字符串、字节、二二进制等等。

不过 RedisTemplate 默认会采用 JDK 自带的序列化（Serialize）来对对象进行转换。生成的数据十分庞大，因此一般都会指定 key 和 value 为 String 类型，这样就由开发者自己把对象序列化为 json 字符串来存储即可。

因大部分情况下，都是使用 key 和 value 都为 String 的 RedisTemplate，故 Spring 默认提供了这样一个实现：

public class StringRedisTemplate extends RedisTemplate<String, String>

### redisService 工具类 ###

可以在代码中直接调用工具类的相关方法

import cn.hutool.core.util.ObjectUtil;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.data.redis.core.*;
    import org.springframework.data.redis.support.atomic.RedisAtomicLong;
    import org.springframework.stereotype.Service;
    import java.io.Serializable;
    import java.util.*;
    import java.util.concurrent.TimeUnit;
    
    @Service
    public class RedisService {
        
        private final Logger logger = LoggerFactory.getLogger(this.getClass());
    
        @Autowired
        private RedisTemplate redisTemplate;
    
        /**
         * 模糊查询key
         */
        public Set<String> listKeys(final String key) {
        
            Set<String> keys = redisTemplate.keys(key);
            return keys;
        }
    
        /**
         * 重命名
         */
        public void rename(final String oldKey, final String newKey) {
        
            redisTemplate.rename(oldKey, newKey);
        }
    
        /**
         * 模糊获取
         */
        public List<Object> listPattern(final String pattern) {
        
            List<Object> result = new ArrayList<>();
            Set<Serializable> keys = redisTemplate.keys(pattern);
            for (Serializable str : keys) {
        
                ValueOperations<Serializable, Object> operations = redisTemplate.opsForValue();
                Object obj = operations.get(str.toString());
                if (!ObjectUtil.isEmpty(obj)) {
        
                    result.add(obj);
                }
            }
            return result;
        }
    
        /**
         * 写入缓存
         */
        public boolean set(final String key, Object value) {
        
            boolean result = false;
            try {
        
                ValueOperations<Serializable, Object> operations = redisTemplate.opsForValue();
                operations.set(key, value);
                result = true;
            } catch (Exception e) {
        
                logger.error("set fail ,key is:" + key, e);
            }
            return result;
        }
    
        /**
         * 批量写入缓存
         */
        public boolean multiSet(Map<String, Object> map) {
        
            boolean result = false;
            try {
        
                ValueOperations<Serializable, Object> operations = redisTemplate.opsForValue();
                operations.multiSet(map);
                result = true;
            } catch (Exception e) {
        
                logger.error("multiSet fail ", e);
            }
            return result;
        }
    
        /**
         * 集合出栈
         */
        public Object leftPop(String key) {
        
            ListOperations list = redisTemplate.opsForList();
            return list.leftPop(key);
        }
    
        public Object llen(final String key) {
        
            final ListOperations list = this.redisTemplate.opsForList();
            return list.size((Object) key);
        }
    
        /**
         * 写入缓存设置时效时间
         */
        public boolean set(final String key, Object value, Long expireTime) {
        
            boolean result = false;
            try {
        
                ValueOperations<Serializable, Object> operations = redisTemplate.opsForValue();
                operations.set(key, value);
                redisTemplate.expire(key, expireTime, TimeUnit.SECONDS);
                result = true;
            } catch (Exception e) {
        
                logger.error("set fail ", e);
            }
            return result;
        }
    
        /**
         * 写入缓存设置时效时间
         */
        public boolean setnx(final String key, Object value, Long expireTime) {
        
            boolean res = false;
            try {
        
                ValueOperations<Serializable, Object> operations = redisTemplate.opsForValue();
                res = operations.setIfAbsent(key, value);
                if (res) {
        
                    redisTemplate.expire(key, expireTime, TimeUnit.SECONDS);
                }
            } catch (Exception e) {
        
                logger.error("setnx fail ", e);
            }
            return res;
        }
    
        /**
         * 缓存设置时效时间
         */
        public void expire(final String key, Long expireTime) {
        
            redisTemplate.expire(key, expireTime, TimeUnit.SECONDS);
        }
    
    
        /**
         * 自增操作
         */
        public long incr(final String key) {
        
            RedisAtomicLong entityIdCounter = new RedisAtomicLong(key, redisTemplate.getConnectionFactory());
            return entityIdCounter.getAndIncrement();
    
        }
    
        /**
         * 批量删除
         */
        public void removeKeys(final List<String> keys) {
        
            if (keys.size() > 0) {
        
                redisTemplate.delete(keys);
            }
        }
    
        /**
         * 批量删除key
         */
        public void removePattern(final String pattern) {
        
            Set<Serializable> keys = redisTemplate.keys(pattern);
            if (keys.size() > 0) {
        
                redisTemplate.delete(keys);
            }
        }
    
        /**
         * 删除对应的value
         */
        public void remove(final String key) {
        
            if (exists(key)) {
        
                redisTemplate.delete(key);
            }
        }
    
        /**
         * 判断缓存中是否有对应的value
         */
        public boolean exists(final String key) {
        
            return redisTemplate.hasKey(key);
        }
    
        /**
         * 判断缓存中是否有对应的value(模糊匹配)
         */
        public boolean existsPattern(final String pattern) {
        
            if (redisTemplate.keys(pattern).size() > 0) {
        
                return true;
            } else {
        
                return false;
            }
        }
    
        /**
         * 读取缓存
         */
        public Object get(final String key) {
        
            ValueOperations<Serializable, Object> operations = redisTemplate.opsForValue();
            return operations.get(key);
        }
    
        /**
         * 哈希 添加
         */
        public void hmSet(String key, Object hashKey, Object value) {
        
            HashOperations<String, Object, Object> hash = redisTemplate.opsForHash();
            hash.put(key, hashKey, value);
        }
    
        /**
         * 哈希 添加
         */
        public Boolean hmSet(String key, Object hashKey, Object value, Long expireTime, TimeUnit timeUnit) {
        
            HashOperations<String, Object, Object> hash = redisTemplate.opsForHash();
            hash.put(key, hashKey, value);
            return redisTemplate.expire(key, expireTime, timeUnit);
        }
    
        /**
         * 哈希获取数据
         */
        public Object hmGet(String key, Object hashKey) {
        
            HashOperations<String, Object, Object> hash = redisTemplate.opsForHash();
            return hash.get(key, hashKey);
        }
    
        /**
         * 哈希获取所有数据
         */
        public Object hmGetValues(String key) {
        
            HashOperations<String, Object, Object> hash = redisTemplate.opsForHash();
            return hash.values(key);
        }
    
        /**
         * 哈希获取所有键值
         */
        public Object hmGetKeys(String key) {
        
            HashOperations<String, Object, Object> hash = redisTemplate.opsForHash();
            return hash.keys(key);
        }
    
        /**
         * 哈希获取所有键值对
         */
        public Object hmGetMap(String key) {
        
            HashOperations<String, Object, Object> hash = redisTemplate.opsForHash();
            return hash.entries(key);
        }
    
        /**
         * 哈希 删除域
         */
        public Long hdel(String key, Object hashKey) {
        
            HashOperations<String, Object, Object> hash = redisTemplate.opsForHash();
            return hash.delete(key, hashKey);
        }
    
        /**
         * 列表添加
         */
        public void rPush(String k, Object v) {
        
            ListOperations<String, Object> list = redisTemplate.opsForList();
            list.rightPush(k, v);
        }
    
        /**
         * 列表删除
         */
        public void listRemove(String k, Object v) {
        
            ListOperations<String, Object> list = redisTemplate.opsForList();
            list.remove(k, 1, v);
        }
    
        public void rPushAll(String k, Collection var2) {
        
            ListOperations<String, Object> list = redisTemplate.opsForList();
            list.rightPushAll(k, var2);
        }
    
        /**
         * 列表获取
         */
        public Object lRange(String k, long begin, long end) {
        
            ListOperations<String, Object> list = redisTemplate.opsForList();
            return list.range(k, begin, end);
        }
    
        /**
         * 集合添加
         */
        public void add(String key, Object value) {
        
            SetOperations<String, Object> set = redisTemplate.opsForSet();
            set.add(key, value);
        }
    
        /**
         * 判断元素是否在集合中
         */
        public Boolean isMember(String key, Object value) {
        
            SetOperations<String, Object> set = redisTemplate.opsForSet();
            return set.isMember(key, value);
        }
    
        /**
         * 集合获取
         */
        public Set<Object> setMembers(String key) {
        
            SetOperations<String, Object> set = redisTemplate.opsForSet();
            return set.members(key);
        }
    
        /**
         * 有序集合添加
         */
        public void zAdd(String key, Object value, double scoure) {
        
            ZSetOperations<String, Object> zset = redisTemplate.opsForZSet();
            zset.add(key, value, scoure);
        }
    
        /**
         * 有序集合获取
         */
        public Set<Object> rangeByScore(String key, double scoure, double scoure1) {
        
            ZSetOperations<String, Object> zset = redisTemplate.opsForZSet();
            return zset.rangeByScore(key, scoure, scoure1);
        }
    
        /**
         * 有序集合根据区间删除
         */
        public void removeRangeByScore(String key, double scoure, double scoure1) {
        
            ZSetOperations<String, Object> zset = redisTemplate.opsForZSet();
            zset.removeRangeByScore(key, scoure, scoure1);
        }
    
        /**
         * 列表添加
         */
        public void lPush(String k, Object v) {
        
            ListOperations<String, Object> list = redisTemplate.opsForList();
            list.rightPush(k, v);
        }
    
        /**
         * 获取当前key的超时时间
         */
        public Long getExpireTime(final String key) {
        
            return redisTemplate.opsForValue().getOperations().getExpire(key, TimeUnit.SECONDS);
        }
    
        public Long extendExpireTime(final String key, Long extendTime) {
        
            Long curTime = redisTemplate.opsForValue().getOperations().getExpire(key, TimeUnit.SECONDS);
            long total = curTime.longValue() + extendTime;
            redisTemplate.expire(key, total, TimeUnit.SECONDS);
            return total;
        }
    
        public Set getKeys(String k) {
        
            return redisTemplate.keys(k);
        }
    
    }

# Spring Cache #

## 概述 ##

Spring 从 3.1 开始定义了 org.springframework.cache.Cache 和 org.springframework.cache.CacheManager 接口来统一不同的缓存技术；并支持使用 JCache（JSR-107）注解简化开发。

*  Cache（缓存）接口为缓存的组件规范定义，包含缓存的各种操作集合
    
    Cache 接口下 Spring 提供了各种 xxxCache 的实现；如 RedisCache ，EhCacheCache ，ConcurrentMapCache 等
 *  CacheManager（缓存管理器）管理各种缓存（Cache）组件，负责对缓存的增删改查
    
    CacheManager 的缓存的介质可配置，如：ConcurrentMap/EhCache/Redis等
    
    当没有加入EhCache 或者 Redis 依赖时默认采用concurrentMap实现的缓存，是存在内存中，重启服务器则清空缓存

**Spring Cache 的原理**

基于 Proxy / AspectJ 动态代理技术的 AOP 思想（面向切面编程）。

每次调用需要缓存功能的方法时，Spring 会检查指定参数的指定的目标方法是否已经被调用过；如果有就直接从缓存中获取方法调用后的结果，如果没有就调用方法并缓存结果后返回给用户。下次调用直接从缓存中获取。

## 配置文件及配置类 ##

**yaml 配置文件属性**

spring:
      cache:
        #cache-names:    # 可以自动配置
        #type: redis     # 可以自动配置
        redis:
          time-to-live: 3600000  # 指定存活时间。单位毫秒，缺省默认为 -1 （永不过时）
          key-prefix: CACHE_     # key前缀，缺省默认使用缓存的名称（@Cacheable注解的value参数值）作为前缀
          use-key-prefix: true   # 是否使用前缀，默认为true，指定为false时不使用任何key前缀
          cache-null-values: true  # 是否缓存空值。默认为true。Spring Cache 对缓存穿透问题的解决方案

**Spring cache 配置类**

一般仅自定义 CacheManager 和 KeyGenerator 即可，其他的自定义属于高阶使用

import com.fasterxml.jackson.annotation.JsonAutoDetect;
    import com.fasterxml.jackson.annotation.PropertyAccessor;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.boot.autoconfigure.cache.CacheProperties;
    import org.springframework.boot.context.properties.EnableConfigurationProperties;
    import org.springframework.cache.Cache;
    import org.springframework.cache.CacheManager;
    import org.springframework.cache.annotation.CachingConfigurerSupport;
    import org.springframework.cache.annotation.EnableCaching;
    import org.springframework.cache.interceptor.CacheErrorHandler;
    import org.springframework.cache.interceptor.CacheResolver;
    import org.springframework.cache.interceptor.KeyGenerator;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.data.redis.cache.RedisCacheConfiguration;
    import org.springframework.data.redis.cache.RedisCacheManager;
    import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
    import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
    import org.springframework.data.redis.serializer.RedisSerializationContext;
    import org.springframework.data.redis.serializer.StringRedisSerializer;
    import java.lang.reflect.Method;
    import java.util.ArrayList;
    import java.util.List;
    
    @Configuration
    @EnableCaching
    @EnableConfigurationProperties(CacheProperties.class)
    @Slf4j
    public class RedisCachingConfig extends CachingConfigurerSupport {
        
    
        /**
         * 自定义缓存管理器
         */
        @Bean
        public RedisCacheManager redisCacheManager(JedisConnectionFactory jedisConnectionFactory, CacheProperties cacheProperties) {
        
    
            Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
            ObjectMapper om = new ObjectMapper();
            om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
            om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
            jackson2JsonRedisSerializer.setObjectMapper(om);
    
            // 配置序列化（解决乱码的问题，因为默认使用JDK的序列化机制，转换为二进制数据）
            RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig()
                // 设置Key的序列化方式
                .serializeKeysWith(
                    RedisSerializationContext.SerializationPair.fromSerializer(new StringRedisSerializer()))
                // 设置值的序列化方式
                .serializeValuesWith(
                    RedisSerializationContext.SerializationPair.fromSerializer(jackson2JsonRedisSerializer))
                .disableCachingNullValues();
    
            // 注：若使用自定义的 RedisCacheConfiguration，则不会自动从配置文件中取出来配置，需要手动注册配置文件中所有的配置项
            CacheProperties.Redis redisProperties = cacheProperties.getRedis();
            if (redisProperties.getTimeToLive() != null) config = config.entryTtl(redisProperties.getTimeToLive());
            if (redisProperties.getKeyPrefix() != null) config = config.prefixKeysWith(redisProperties.getKeyPrefix());
            if (!redisProperties.isCacheNullValues()) config = config.disableCachingNullValues();
            if (!redisProperties.isUseKeyPrefix()) config = config.disableKeyPrefix();
    
            return RedisCacheManager
                .builder(jedisConnectionFactory)
                .cacheDefaults(config)
                .build();
        }
    
        /**
         * 自定义 key 生成器
         */
        @Bean
        @Override
        public KeyGenerator keyGenerator() {
        
            return new KeyGenerator() {
        
                @Override
                public Object generate(Object target, Method method, Object... params) {
        
                    StringBuilder sb = new StringBuilder();
                    sb.append(target.getClass().getName());
                    sb.append("$").append(method.getName());
                    for (int i = 0; i < params.length; i++) {
        
                        if (i == 0) {
        
                            sb.append("[").append(params[i].toString());
                        } else {
        
                            sb.append(",").append(params[i].toString());
                        }
                    }
                    sb.append("]");
                    return sb.toString();
                }
            };
        }
    
        /**
         * 自定义错误处理器
         */
        @Override
        @Bean
        public CacheErrorHandler errorHandler() {
        
            // 当缓存读写异常时，忽略异常
            return new CacheErrorHandler(){
        
                @Override
                public void handleCacheGetError(RuntimeException e, Cache cache, Object o) {
        
                    log.error(e.getMessage(), e);
                }
                @Override
                public void handleCachePutError(RuntimeException e, Cache cache, Object o, Object o1) {
        
                    log.error(e.getMessage(), e);
                }
                @Override
                public void handleCacheEvictError(RuntimeException e, Cache cache, Object o) {
        
                    log.error(e.getMessage(), e);
                }
                @Override
                public void handleCacheClearError(RuntimeException e, Cache cache) {
        
                    log.error(e.getMessage(), e);
                }
            };
        }
    
        /**
         * 自定义缓存解析器
         */
        @Override
        @Bean
        public CacheResolver cacheResolver() {
        
            // 通过Guava实现的自定义堆内存缓存管理器
    //        CacheManager guavaCacheManager = new GuavaCacheManager();
            CacheManager redisCacheManager = this.cacheManager();
            List<CacheManager> list = new ArrayList<>();
            // 优先读取堆内存缓存
    //        list.add(concurrentMapCacheManager);
            // 堆内存缓存读取不到该key时再读取redis缓存
            list.add(redisCacheManager);
            return new CustomCacheResolver(list);
        }
    }

**自定义缓存解析器类**

public class CustomCacheResolver implements CacheResolver, InitializingBean {
        
    
        @Nullable
        private List<CacheManager> cacheManagerList;
    
        public CustomCacheResolver(){
        }
        public CustomCacheResolver(List<CacheManager> cacheManagerList){
        
            this.cacheManagerList = cacheManagerList;
        }
    
        public void setCacheManagerList(@Nullable List<CacheManager> cacheManagerList) {
        
            this.cacheManagerList = cacheManagerList;
        }
        public List<CacheManager> getCacheManagerList() {
        
            return cacheManagerList;
        }
    
        @Override
        public void afterPropertiesSet()  {
        
            Assert.notNull(this.cacheManagerList, "CacheManager is required");
        }
    
        @Override
        public Collection<? extends Cache> resolveCaches(CacheOperationInvocationContext<?> context) {
        
            Collection<String> cacheNames = context.getOperation().getCacheNames();
            if (cacheNames == null) {
        
                return Collections.emptyList();
            }
            Collection<Cache> result = new ArrayList<>();
            for(CacheManager cacheManager : getCacheManagerList()){
        
                for (String cacheName : cacheNames) {
        
                    Cache cache = cacheManager.getCache(cacheName);
                    if (cache == null) {
        
                        throw new IllegalArgumentException("Cannot find cache named '" +
                                cacheName + "' for " + context.getOperation());
                    }
                    result.add(cache);
                }
            }
            return result;
        }
    }

## CachingConfigurerSupport 说明 ##

支持自定义缓存的读写机制：

*  cacheManager（缓存管理器）
    
    默认情况，SpringBoot 会使用 SimpleCacheConfiguration 缓存配置类。然后创建一个 ConcurrentMapCacheManager 缓存管理器，可以获取 ConcurrentMap 来作为缓存组件使用。
    
    引入 redis 的 starter 后，RedisCacheConfiguration 缓存配置类就会生效，会创建一个 RedisCacheManager
    
     *  默认创建的 RedisCacheManager 在操作 redis 的时候 RedisTemplate<Object, Object>
     *  RedisTemplate<Object, Object> 是默认使用 JDK的序列化机制
     *  想要保存为 JSON 格式就可以自定义 CacheManager
     *  注：从执行时间上来看，JdkSerializationRedisSerializer 是最高效的（毕竟是JDK原生的），但是序列化的结果字符串是最长的。 JSON 由于其数据格式的紧凑性，序列化的长度是最小的，时间比前者要多一些。而 OxmSerialiabler 在时间上看是最长的（当时和使用具体的 Marshaller 有关）。故推荐使用 JacksonJsonRedisSerializer 作为 POJO 的序列器。
 *  keyGenerator（key 生成器）
    
    当 cache 相关注解未指定时，默认自动使用 SimpleKeyGenerator（将方法的所有参数值进行组合）生成 key，若不同方法指定相同的缓存分区，且参数值相同，SimpleKeyGenerator 自动生成的 key 就相同了，可以自定义 keyGenerator 避免发生这种情况
 *  errorHandler（错误处理器）
    
    当 redis 连接出现异常时，调用标注了 cache 相关注解的方法会抛出异常影响到正常的业务流程，可以自定义 errorHandler 处理缓存读写的异常
    
    如果缓存**读**发生了异常：
    
     *  缓存错误处理器可以采用忽略异常，从而继续从数据库读取数据，对业务没有影响
     *  但是如果请求量很大就会出现缓存雪崩的问题，大量的查询请求发送到数据库导致数据库负载过大而阻塞甚至宕机
     *  建议使用多层缓存兜底
    
    如果缓存**写**发生了异常，就可能导致数据库的数据和缓存的数据不一致的问题：
    
     *  为了解决该问题，需要继续扩展 CacheErrorHandler 的 handleCachePutError 和 handleCacheEvictError 方法
     *  思路就是将 redis 写操作失败的 key 保存下来，通过重试任务删除这些 key 对应的缓存解决数据库数据与缓存数据不一致的问题
 *  cacheResolver（缓存解析器）
    
    可以通过自定义 CacheResolver 实现动态选择 CacheManager
    
    可以使用多种缓存机制：优先从堆内存读取缓存，堆内存缓存不存在时再从 redis 读取缓存，redis 缓存不存在时最后从数据库读取数据，并将读取到的数据依次写到 redis 和堆内存中。
    
    通过自定义 CacheResolver 开发者可以实现更多的自定义功能，例如热点缓存自动升降级的场景：
    
     *  项目大多数情况下只使用 redis 做缓存，当某些场景下个别数据成为了热数据，通过例如 storm 实时统计出热数据后，项目将这些热数据缓存到堆内存，缓解网络和 redis 的负载压力。
     *  这种场景完全可以通过自定义 CacheResolver 来实现，storm 实时统计出热数据，自定义的 CacheResolver 在调用resolveCaches 选择 CacheManager 前，先判断此次读写的缓存 key 是否是热数据。如果是热数据则使用堆内存的CacheManager，否则使用redis的CacheManager。

## Spring Cache 中的主要注解 ##

*  **@EnableCaching** ：开启基于注解的缓存功能，在配置类上标注 @EnableCaching 注解（不需要重复配 Redis）
 *  **@Cacheable** ：缓存数据或者获取缓存数据，，一般用在**查询方法**上
 *  **@CachePut** ：修改缓存数据。保证方法被调用，又希望结果被缓存。一般用在**新增方法**上
 *  **@CacheEvict** ： 清空缓存。一般用在**更新或者删除方法**上
 *  **@CacheConfig** ：统一配置 @Cacheable 中的 value 值，主要标注在类上，也可标注在方法上
 *  **@Caching** ：组合多个 Cache 注解

### @Cacheable：缓存数据|获取缓存 ###

缓存数据或者获取缓存数据，一般用在**查询方法**上。

标注的方法第一次被调用时，根据方法**对其返回结果进行缓存**（注意：保存的数据是 return 返回的数据），下次请求时，如果**缓存存在**，则直接读取缓存数据返回；如果**缓存不存在**，则执行方法，并把返回的结果存入缓存中。

缓存的数据默认使用 JDK 序列化机制（将数据转换为二进制），默认过期时间 TTL -1（永不过去）。

主要参数：

*  **value / cacheNames** 属性：指定缓存的名称（缓存的前缀/分区/缓存空间，按照业务类型分）
    
    必填，至少指定一个；也可以用 @CacheConfig 替代
    
    示例：
    
        @Cacheable(value="testcache")
         @Cacheable(value={
              "testcache1","testcache2"}
 *  \*\*key \*\*属性：缓存的键后缀
    
    为空时默认使用 `SimpleKeyGenerator`（将方法的所有参数值进行组合）生成 key，如果指定要按照 SpEL 表达式编写
    
    注：完整的缓存键值对格式为：value属性值::key属性值=方法返回值
    
     *  `value属性值::` 为缓存的前缀，可通过配置文件 spring.cache.redis.key-prefix 属性指定
    
    示例：
    
        @Cacheable(value="testcache", key="#id")
    
    SimpleKeyGenerator 源码（了解）：
    
        public SimpleKey(Object... elements) {
              
                Assert.notNull(elements, "Elements must not be null");
                this.params = new Object[elements.length];
                System.arraycopy(elements, 0, this.params, 0, elements.length);
                this.hashCode = Arrays.deepHashCode(this.params);
            }
 *  **condition** 属性：缓存的条件
    
    可以为空，使用 SpEL 编写，返回 true 或者 false
    
    只有为 true 才进行缓存/清除缓存，在调用方法之前之后都能判断
    
    示例：
    
        @Cacheable(value="testcache", condition="#id.length()>2")
 *  **unless** 属性：否决缓存的条件
    
    条件为 true 不缓存，false 才缓存
    
    只在方法执行之后判断，此时可以拿到返回值 result 进行判断
    
    示例：
    
        @Cacheable(value="testcache", condition="#result == null")
 *  **sync** 属性：是否使用异步模式
    
    即执行方法时是否加锁。默认为 false
    
    Sping Cache 对 **缓存击穿**（大量并发进来同时查询一个正好过期的数据）问题的解决方案
 *  **keyGenerator** 属性：key的生成器
    
    可以指定 key 的组件id，与 key 属性只能二选一使用
 *  **cacheManager** 属性：指定缓存管理器
 *  **cacheResolver** 属性：指定获取解析器

### @CacheEvict：清除缓存数据 ###

使用该注解标志的方法，会**清除指定的缓存**。一般用在**更新或者删除方法**上（即更新数据库数据后马上清除缓存，删除数据库数据后也马上清除缓存，以便下次查询重新获取缓存）。

根据对应的 value 和 key 删除缓存，value 和 key 必须相同才会删除（注：value + key 组合成 redis 的键）；若没有指定 key 值且 allEntries=false 时，则 key 值默认取入参值删除缓存，若没有入参则不清除缓存。

主要参数：value，key，condition，allEntries，beforeInvocation

*  **allEntries** 属性：是否清空所有缓存内容
    
    缺省为 false，如果指定为 true，则方法调用后将立即清空所有缓存
    
    示例：
    
        @CachEvict(value="testcache", allEntries=true)
 *  **beforeInvocation** 属性：是否在方法执行前就清除缓存数据
    
    缺省为 false，缺省清空下，如果方法执行抛出异常，则不会清除缓存
    
    如果指定为 true，则在方法还没有执行的时候就清除缓存
    
    示例：
    
        @CachEvict(value="testcache", beforeInvocation=true)
    
    注：作用只有一个，就是先清除缓存再执行方法

### @CachePut：新增或更新缓存 ###

使用该注解标志的方法，每次都会执行，并将**结果存入指定的缓存中**，一般用在**新增方法**上

先根据 value 和 key 查询缓存，如果存在则修改；不存在则新增。

主要参数：value，key，condition

**注意：保存的数据是 return 返回的数据**

### @CacheConfig：统一配置 value 值 ###

统一配置 @Cacheable 注解中的 value 值，主要标注在类上，也可标注在方法上

如果 @Cacheable 注解中没有 value 值则用 @CacheConfig 中的值；如果 @Cacheable 注解中有 value 值则以@Cacheable 中的 value 值为准（就近原则）。

### @Caching：组合多个注解 ###

组合注解，可以组合多个注解

@Caching(put = {
        
    	@CachePut(value = "user", key = "#user.id"),
    	@CachePut(value = "user", key = "#user.username"),
    	@CachePut(value = "user", key = "#user.email")
    })
    public User save(User user) {
        }

## Cache SpEL 表达式 ##

Cache SpEL 表达式语法

<table> 
 <thead> 
  <tr> 
   <th>名称</th> 
   <th>位置</th> 
   <th>描述</th> 
   <th>示例</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>methodName</td> 
   <td>root object</td> 
   <td>当前被调用的方法名</td> 
   <td>#root.methodName</td> 
  </tr> 
  <tr> 
   <td>method</td> 
   <td>root object</td> 
   <td>当前被调用的方法</td> 
   <td>#root.method.name</td> 
  </tr> 
  <tr> 
   <td>target</td> 
   <td>root object</td> 
   <td>当前被调用的目标对象</td> 
   <td>#root.target</td> 
  </tr> 
  <tr> 
   <td>targetClass</td> 
   <td>root object</td> 
   <td>当前被调用的目标对象类</td> 
   <td>#root.targetClass</td> 
  </tr> 
  <tr> 
   <td>args</td> 
   <td>root object</td> 
   <td>当前被调用的方法的参数列表</td> 
   <td>#root.args[0]</td> 
  </tr> 
  <tr> 
   <td>caches</td> 
   <td>root object</td> 
   <td>当前方法调用使用的缓存列表（如@Cacheable(value={“cache1”,“cache2”})，则有两个cache）</td> 
   <td>#root.caches[0].name</td> 
  </tr> 
  <tr> 
   <td>argument name</td> 
   <td>evaluation context</td> 
   <td>方法参数的名字，可以直接 #参数名，也可以使用 #p0或#a0 的形式，0代表参数的索引</td> 
   <td>#iban、#a0、#p0</td> 
  </tr> 
  <tr> 
   <td>result</td> 
   <td>evaluation context</td> 
   <td>方法执行后的返回值（仅当方法执行之后的判断有效，如’unless’，'cache put’的表达式，'cache evict’的表达式beforeInvocation=false）</td> 
   <td>#result</td> 
  </tr> 
 </tbody> 
</table>

**SpEL 运算符**

<table> 
 <thead> 
  <tr> 
   <th>类型</th> 
   <th>运算符</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>关系运算符</td> 
   <td>&lt; ，&gt; ， &lt;= ，&gt;=，==，!=,lt，gt，le，ge，eq，ne</td> 
  </tr> 
  <tr> 
   <td>算数运算符</td> 
   <td>+，-，*，/ ，%，^</td> 
  </tr> 
  <tr> 
   <td>逻辑运算符</td> 
   <td>&amp;&amp;，</td> 
  </tr> 
  <tr> 
   <td>条件运算符</td> 
   <td>? : (ternary),? : (elvis)</td> 
  </tr> 
  <tr> 
   <td>正则表达式</td> 
   <td>matches</td> 
  </tr> 
  <tr> 
   <td>其他类型</td> 
   <td>?. ，?[…] ，![…] ，<sup><a rel="nofollow">1</a></sup>，$[…]</td> 
  </tr> 
 </tbody> 
</table>

# Redisson #

## 概述 ##

某些场景下，可能需要实现分布式的不同类型锁，比如：公平锁、互斥锁、可重入锁、读写锁、红锁（redLock）等等。实现起来比较麻烦。开源框架 Redisson 实现了上述的这些锁功能，而且还有很多其它的强大功能。

Redisson 是一个在 Redis 的基础上实现的 Java 驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的 Java 常用对象，还提供了许多分布式服务。其中包括(`BitSet`, `Set`, `Multimap`, `SortedSet`, `Map`, `List`, `Queue`, `BlockingQueue`, `Deque`, `BlockingDeque`, `Semaphore`, `Lock`, `AtomicLong`, `CountDownLatch`, `Publish / Subscribe`, `Bloom filter`, `Remote service`, `Spring cache`, `Executor service`, `Live Object service`, `Scheduler service`) 等，Redisson提供了使用 Redis 的最简单和最便捷的方法。Redisson的宗旨是促进使用者对 Redis 的关注分离（Separation of Concern），从而让使用者能够将精力更集中地放在处理业务逻辑上。

*  [官网地址][Link 2]
 *  [GitHub地址][GitHub]

## 依赖及客户端配置 ##

**依赖**

<dependency>
        <groupId>org.redisson</groupId>
        <artifactId>redisson</artifactId>
        <version>3.10.6</version>
    </dependency>

**配置 Redisson 客户端：**

import org.redisson.Redisson;
    import org.redisson.api.RedissonClient;
    import org.redisson.config.Config;
    import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class RedisConfig {
        
        @Bean
        public RedissonClient redissonClient(RedisProperties prop) {
        
            Config config = new Config();
            config.useSingleServer().setAddress("redis://" + prop.getHost() + ":" + prop.getPort());
            return Redisson.create(config);
        }
    }

注意：这里读取了一个名为 RedisProperties 的属性，因为引入了SpringDataRedis，Spring已经自动加载了 RedisProperties，并且读取了配置文件中的 Redis 信息。

## 常用 API ##

**RedissonClient 接口**

// 创建锁对象，并指定锁的名称
    RLock getLock(String name)

**RLock 接口**

获取锁 方法

// 获取锁，`waitTime`默认0s，即获取锁失败不重试，`leaseTime`默认30s
    boolean tryLock()
    // 获取锁，设置锁等待时间`waitTime`、时间单位`unit`。释放时间`leaseTime`默认的30s
    boolean tryLock(long waitTime, TimeUnit unit)
    // 获取锁，设置锁等待时间`waitTime`、释放时间`leaseTime`，时间单位`unit`。
    boolean tryLock(long waitTime, long leaseTime, TimeUnit unit)
      // 如果获取锁失败后，会在`waitTime`减去获取锁用时的剩余时间段内继续尝试获取锁，如果依然获取失败，则认为获取锁失败；
      // 获取锁后，如果超过`leaseTime`未释放，为避免死锁会自动释放。
    
    // 释放锁
    void unlock()

# Redis 分布式锁 #

## Redis 分布式锁原理 ##

分布式锁的关键是**多进程共享的内存标记**，因此只要在 Redis 中放置一个这样的标记就可以了。

在实现分布式锁时，注意需要实现下列目标：

*  多进程可见：多进程可见，否则就无法实现分布式效果
    
     *  redis 本身就是多服务共享的，不用过多关注
 *  避免死锁：死锁的情况有很多，要考虑各种异常导致死锁的情况，保证锁可以被释放
    
     *  服务宕机后的锁释放问题：设置锁时最好设置锁的有效期，如果服务宕机，有效期到时自动删除锁
        
            > set lock 001 nx ex 20
            OK
            > get lock
            001
            > ttl lock
            10
            > set lock 001 nx ex 20
            null
 *  排它：同一时刻，只能有一个进程获得锁
    
     *  可以利用 Redis 的 setnx 命令（ set when not exits）来实现。当多次执行 setnx 命令时，只有第一次执行的才会成功并返回1，其它情况返回0
        
            > setnx lock 001
            1
            > get lock
            001
            > setnx lock 001
            0
        
        定义一个固定的 key，多个进程都执行 setnx，设置这个 key 的值，返回 1 的服务获取锁，返回 0 则没有获取
 *  高可用：避免锁服务宕机或处理好宕机的补救措施
    
     *  利用Redis的主从、哨兵、集群，保证高可用

## 分布式不可重入锁 ##

### 流程 ###

按照上面所述的理论，分布式锁的流程大概如下：

![在这里插入图片描述][b3e9b818db314ff2a515ccbbb396e647.png_pic_center]

基本流程：

*  1、通过set命令设置锁
 *  2、判断返回结果是否是OK
    
     *  1）Nil，获取失败，结束或重试（自旋锁）
     *  2）OK，获取锁成功
        
         *  执行业务
         *  释放锁
 *  3、异常情况，服务宕机。超时时间EX结束，会自动释放锁

**注意：释放锁时需要判断锁的value释放跟自己存进去的一致**

不然下面的场景下会出现释放锁的问题：

1.  三个进程：A和B和C，在执行任务，并争抢锁，此时A获取了锁，并设置自动过期时间为10s
2.  A开始执行业务，因为某种原因，业务阻塞，耗时超过了10秒，此时锁自动释放了
3.  B恰好此时开始尝试获取锁，因为锁已经自动释放，成功获取锁
4.  A此时业务执行完毕，执行释放锁逻辑（删除key），于是B的锁被释放了，而B其实还在执行业务
5.  此时进程C尝试获取锁，也成功了，因为A把B的锁删除了。
    
    B 和 C 同时获取了锁，违反了排它性！

### 代码实现 ###

**定义一个锁接口：**

public interface RedisLock {
        
        boolean lock(long releaseTime);
        void unlock();
    }

**定义一个锁工具：**

import org.springframework.data.redis.core.StringRedisTemplate;
    import java.util.UUID;
    import java.util.concurrent.TimeUnit;
    
    public class SimpleRedisLock implements RedisLock{
        
    
        private StringRedisTemplate redisTemplate;
        /**
         * 设定好锁对应的 key
         */
        private String key;
        /**
         * 存入的线程信息的前缀，防止与其它JVM中线程信息冲突
         */
        private final String ID_PREFIX = UUID.randomUUID().toString();
    
        public SimpleRedisLock(StringRedisTemplate redisTemplate, String key) {
        
            this.redisTemplate = redisTemplate;
            this.key = key;
        }
    
        public boolean lock(long releaseTime) {
        
            // 获取线程信息作为值，方便判断是否是自己的锁
            String value = ID_PREFIX + Thread.currentThread().getId();
            // 尝试获取锁
            Boolean boo = redisTemplate.opsForValue().setIfAbsent(key, value, releaseTime, TimeUnit.SECONDS);
            // 判断结果
            return boo != null && boo;
        }
    
        public void unlock(){
        
            // 获取线程信息作为值，方便判断是否是自己的锁
            String value = ID_PREFIX + Thread.currentThread().getId();
            // 获取现在的锁的值
            String val = redisTemplate.opsForValue().get(key);
            // 判断是否是自己
            if(value.equals(val)) {
        
                // 删除key即可释放锁
                redisTemplate.delete(key);
            }
        }
    }

**在定时任务中使用锁：**

import com.test.task.utils.SimpleRedisLock;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.data.redis.core.StringRedisTemplate;
    import org.springframework.scheduling.annotation.Scheduled;
    import org.springframework.stereotype.Component;
    
    @Slf4j
    @Component
    public class HelloJob {
        
    
        @Autowired
        private StringRedisTemplate redisTemplate;
    
        @Scheduled(cron = "0/10 * * * * ?")
        public void hello() {
        
            // 创建锁对象
            RedisLock lock = new SimpleRedisLock(redisTemplate, "lock");
            // 获取锁,设置自动失效时间为50s
            boolean isLock = lock.lock(50);
            // 判断是否获取锁
            if (!isLock) {
        
                // 获取失败
                log.info("获取锁失败，停止定时任务");
                return;
            }
            try {
        
                // 执行业务
                log.info("获取锁成功，执行定时任务。");
                // 模拟任务耗时
                Thread.sleep(500);
            } catch (InterruptedException e) {
        
                log.error("任务执行异常", e);
            } finally {
        
                // 释放锁
                lock.unlock();
            }
        }
    }

## 分布式可重入锁 ##

### 可重入锁概述 ###

可重入锁，也叫做递归锁，指的是在同一线程内，外层函数获得锁之后，内层递归函数仍然可以获取到该锁。换一种说法：**同一个线程再次进入同步代码时，可以使用自己已获取到的锁。**

可重入锁可以避免因同一线程中多次获取锁而导致死锁发生。

如何实现可重入锁：**在锁已经被使用时，判断这个锁是否是自己的，如果是则再次获取**。

可以在set锁的值时，**存入获取锁的线程的信息**，这样下次再来时，就能知道当前持有锁的是不是自己，如果是就允许再次获取锁。

要注意，因为锁的获取是**可重入**的，因此必须记录重入的次数，这样不至于在释放锁时一下就释放掉，而是逐层释放。

因此，不能再使用简单的key-value结构，这里推荐使用hash结构：

*  key：lock
 *  hashKey：线程信息
 *  hashValue：重入次数，默认1

释放锁时，每次都把**重入次数减一**，减到 0 说明多次获取锁的逻辑都执行完毕，才可以删除key，释放锁

### 流程图 ###

这里重点是获取锁的流程：

![在这里插入图片描述][afe88b6bb2e243a28c2cacc68cd47890.png_pic_center]

下面假设锁的 key 为 “lock”，hashKey 是当前线程的 id：“threadId”，锁自动释放时间假设为 20 s

**获取锁的步骤：**

1.  判断 lock 是否存在 `EXISTS lock`
    
    存在，说明锁已被获取，接下来判断是不是自己的锁
    
    判断当前线程 id 作为 hashKey 是否存在：`HEXISTS lock threadId`
    
     *  不存在，说明锁已被获取，且不是自己获取的，锁获取失败，end
     *  存在，说明是自己获取的锁，重入次数+1：`HINCRBY lock threadId 1`，去到步骤3
2.  不存在，说明可以获取锁，`HSET key threadId 1`
3.  设置锁自动释放时间，`EXPIRE lock 20`

**释放锁的步骤：**

1.  判断当前线程 id 作为 hashKey 是否存在：`HEXISTS lock threadId`
    
     *  不存在，说明锁已经失效，不用管了
     *  存在，说明锁还在，重入次数减1：`HINCRBY lock threadId -1`，获取新的重入次数
2.  判断重入次数是否为 0：
    
     *  为 0，说明锁全部释放，删除key：`DEL lock`
     *  大于 0，说明锁还在使用，重置有效时间：`EXPIRE lock 20`

上述流程有一个最大的问题，就是有大量的判断，这样在多线程运行时，会有线程安全问题，除非能保证**执行命令的原子性**。

常见分布式可重入锁实现：

*  Redisson 分布式锁
 *  执行 lua 脚本。lua 脚本中可以定义多条语句，语句执行具备原子性。

## Redisson 分布式锁 ##

代码示例

import lombok.extern.slf4j.Slf4j;
    import org.redisson.api.RLock;
    import org.redisson.api.RedissonClient;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.scheduling.annotation.Scheduled;
    
    @Slf4j
    @Component
    public class RedsssionJob {
        
    
        @Autowired
        private RedissonClient redissonClient;
    
        @Scheduled(cron = "0/10 * * * * ?")
        public void hello() {
        
            // 创建锁对象，并制定锁的名称
            RLock lock = redissonClient.getLock("taskLock");
            // 获取锁，自动失效时间默认为50s
            boolean isLock = lock.tryLock();
            // 判断是否获取锁
            if (!isLock) {
        
                // 获取失败
                log.info("获取锁失败，停止定时任务");
                return;
            }
            try {
        
                // 执行业务
                log.info("获取锁成功，执行定时任务。");
                // 模拟任务耗时
                Thread.sleep(500);
            } catch (InterruptedException e) {
        
                log.error("任务执行异常", e);
            } finally {
        
                // 释放锁
                lock.unlock();
                log.info("任务执行完毕，释放锁");
            }
        }
    }

## Lua 脚本分布式锁（了解） ##

Lua 脚本介绍详见拓展之 Redis 的 Lua 脚本

### 分布式锁 Lua 脚本编写 ###

假设有3个参数：

*  KEYS\[1\]：就是锁的 key
 *  ARGV\[1\]：就是线程 id 信息
 *  ARGV\[2\]：锁过期时长

获取锁：

if (redis.call('EXISTS', KEYS[1]) == 0) then
        redis.call('HSET', KEYS[1], ARGV[1], 1);
        redis.call('EXPIRE', KEYS[1], ARGV[2]);
        return 1;
    end;
    if (redis.call('HEXISTS', KEYS[1], ARGV[1]) == 1) then
        redis.call('HINCRBY', KEYS[1], ARGV[1], 1);
        redis.call('EXPIRE', KEYS[1], ARGV[2]);
        return 1;
    end;
    return 0;

释放锁：

if (redis.call('HEXISTS', KEYS[1], ARGV[1]) == 0) then
        return nil;
    end;
    local count = redis.call('HINCRBY', KEYS[1], ARGV[1], -1);
    if (count > 0) then
        redis.call('EXPIRE', KEYS[1], ARGV[2]);
        return nil;
    else
        redis.call('DEL', KEYS[1]);
        return nil;
    end;

### Java 执行 Lua 脚本 ###

RedisTemplate 中提供了一个方法，用来执行 Lua 脚本：

public <T> T execute(RedisScript<T> script, List<K> keys, Object... args)

参数：

*  RedisScript script ：封装了 Lua 脚本的对象
 *  List keys ：脚本中的 key 的值
 *  Object … args ：脚本中的参数的值

把脚本封装到 RedisScript 对象中，有两种方式来构建 RedisScript 对象：

*  方式1：自定义 RedisScript 的实现类 DefaultRedisScript 的对象（常用）
    
        // 场景脚本对象
        DefaultRedisScript<Long> script = new DefaultRedisScript<Long>();
        // 设置脚本数据源，从 classpath 读取
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("lock.lua")));
        // 设置返回值类型
        script.setResultType(Long.class);
    
    可以把脚本文件写到 classpath 下的某个位置，然后通过加载这个文件来获取脚本内容，并设置给 DefaultRedisScript 实例
 *  方式2：通过 RedisScript 中的静态方法（需要把脚本内容写到代码中，作为参数传递，不够优雅）
    
        static <T> RedisScript<T> of(String script)
        static <T> RedisScript<T> of(String script, Class<T> resultType)
    
    参数：
    
     *  String script ：Lua 脚本
     *  Class resultType ：返回值类型

### 可重入分布式锁的实现 ###

1.  在 classpath 中编写两个 Lua 脚本文件
2.  定义一个类（ReentrantRedisLock）实现 RedisLock 接口
    
    基本逻辑：利用静态代码块来加载脚本并初始化，实现 RedisLock 接口的 lock 和 unlock 方法
    
        public class ReentrantRedisLock implements RedisLock {
              
        
            private StringRedisTemplate redisTemplate;
            /**
             * 设定好锁对应的 key
             */
            private String key;
        
            /**
             * 存入的线程信息的前缀，防止与其它JVM中线程信息冲突
             */
            private final String ID_PREFIX = UUID.randomUUID().toString();
        
            public ReentrantRedisLock(StringRedisTemplate redisTemplate, String key) {
              
                this.redisTemplate = redisTemplate;
                this.key = key;
            }
        
            private static final DefaultRedisScript<Long> LOCK_SCRIPT;
            private static final DefaultRedisScript<Object> UNLOCK_SCRIPT;
            
            static {
              
                // 加载释放锁的脚本
                LOCK_SCRIPT = new DefaultRedisScript<>();
                LOCK_SCRIPT.setScriptSource(new ResourceScriptSource(new ClassPathResource("lock.lua")));
                LOCK_SCRIPT.setResultType(Long.class);
        
                // 加载释放锁的脚本
                UNLOCK_SCRIPT = new DefaultRedisScript<>();
                UNLOCK_SCRIPT.setScriptSource(new ResourceScriptSource(new ClassPathResource("unlock.lua")));
            }
            
            // 锁释放时间
            private String releaseTime;
        
            @Override
            public boolean lock(long releaseTime) {
              
                // 记录释放时间
                this.releaseTime = String.valueOf(releaseTime);
                // 执行脚本
                Long result = redisTemplate.execute(
                        LOCK_SCRIPT,
                        Collections.singletonList(key),
                        ID_PREFIX + Thread.currentThread().getId(), this.releaseTime);
                // 判断结果
                return result != null && result.intValue() == 1;
            }
        
            @Override
            public void unlock() {
              
                // 执行脚本
                redisTemplate.execute(
                        UNLOCK_SCRIPT,
                        Collections.singletonList(key),
                        ID_PREFIX + Thread.currentThread().getId(), this.releaseTime);
            }
        }
3.  新建一个定时任务，测试重入锁：
    
        import com.leyou.task.utils.RedisLock;
        import com.leyou.task.utils.ReentrantRedisLock;
        import lombok.extern.slf4j.Slf4j;
        import org.springframework.beans.factory.annotation.Autowired;
        import org.springframework.data.redis.core.StringRedisTemplate;
        import org.springframework.scheduling.annotation.Scheduled;
        import org.springframework.stereotype.Component;
        
        @Slf4j
        @Component
        public class ReentrantJob {
              
        
            @Autowired
            private StringRedisTemplate redisTemplate;
        
            private int max = 2;
        
            @Scheduled(cron = "0/10 * * * * ?")
            public void hello() {
              
                // 创建锁对象
                RedisLock lock = new ReentrantRedisLock(redisTemplate, "lock");
                // 执行任务
                runTaskWithLock(lock, 1);
            }
        
            private void runTaskWithLock(RedisLock lock, int count) {
              
                // 获取锁,设置自动失效时间为50s
                boolean isLock = lock.lock(50);
                // 判断是否获取锁
                if (!isLock) {
              
                    // 获取失败
                    log.info("{}层 获取锁失败，停止定时任务", count);
                    return;
                }
                try {
              
                    // 执行业务
                    log.info("{}层 获取锁成功，执行定时任务。", count);
                    Thread.sleep(500);
                    if(count < max){
              
                        runTaskWithLock(lock, count + 1);
                    }
                } catch (InterruptedException e) {
              
                    log.error("{}层 任务执行失败", count, e);
                } finally {
              
                    // 释放锁
                    lock.unlock();
                    log.info("{}层 任务执行完毕，释放锁", count);
                }
            }
        }

# 拓展 #

## 缓存穿透、雪崩、击穿 ##

*  **缓存穿透**
    
    是指查询一个一定不存在的数据（缓存和数据库中均不存在该数据），每次都会去数据库查询，高并发时可能会导致数据库挂掉或者发生 io 阻塞。
    
    原因为：一般是首次请求不命中时才查询数据库进行缓存，并且出于容错考虑，如果从数据库查不到数据则不写入缓存，这将导致不存在的数据每次请求都都查询数据库，失去了缓存的意义。
    
    若有人利用不存在的 key 频繁攻击应用，这就是漏洞。
    
    **解决方案：**
    
     *  方案1：采用布隆过滤器，将所有可能存在的数据哈希到一个足够大的bitMap中，一个一定不存在的数据会被这个bitMap拦截掉，从而避免了对底层存储系统的查询压力。
     *  方案2：如果一个查询返回的数据为空（不管是数据不存在，还是系统故障），则存入一个空字符进行缓存，并设置一个较短的过期时间，最长不超过五分钟。
 *  **缓存雪崩**
    
    是指在设置缓存时采用了相同的过期时间，导致缓存在某一时刻同时失效，请求全部转发到DB，DB瞬时压力过重导致雪崩。
    
    缓存失效时的雪崩效应对底层系统的冲击非常可怕。
    
    **解决方案：**
    
     *  方案1：用加锁或者队列的方式保证缓存的单线程（进程）写，从而避免失效时大量的并发请求落到底层存储系统上
     *  方案2：将缓存失效时间分散开，比如可以在原有的失效时间基础上增加一个随机值，比如1-5分钟随机，这样每一个缓存的过期时间的重复率就会降低，就很难引发集体失效的事件。
 *  **缓存击穿**
    
    一个设置了过期时间的 key 在失效时被超高并发地访问（非常"热点"的数据），缓存过期，数据又还没有重新加载到缓存中，并发压力瞬间被转移到数据库，这时大并发的请求可能会瞬间导致数据库挂掉或者发生 io 阻塞。
    
    缓存击穿和缓存雪崩的区别在于缓存击穿针对某一 key 缓存失效，缓存雪崩则是很多 key 同时失效。
    
    **解决方案：**
    
     *  使用互斥锁（mutex key）
        
        常用 mutex。简单地来说，就是在缓存失效的时候（判断拿出来的值为空），不是立即去 load db，而是先使用缓存工具的某些带成功操作返回值的操作（比如 Redis 的 SETNX 或者 Memcache 的 ADD）去 set 一个 mutex key，当操作返回成功时，再进行 load db 的操作并回设缓存；否则，就重试整个 get 缓存的方法。

## Redis 的 Lua 脚本 ##

### 介绍 ###

实现 Redis 的原子操作有多种方式，比如 Redis 事务，但是相比而言，使用 Redis 的 Lua 脚本更加优秀，具有不可替代的好处：

*  原子性：redis 会将整个脚本作为一个整体执行，不会被其他命令插入。
 *  复用：客户端发送的脚本会永久存在 redis 中，以后可以重复使用，而且各个 Redis 客户端可以共用。
 *  高效：Lua 脚本解析后会形成缓存，不用每次执行都解析。
 *  减少网络开销：Lua 脚本缓存后，可以形成 SHA 值，作为缓存的 key，以后调用可以直接根据 SHA 值来调用脚本，不用每次发送完整脚本，较少网络占用和时延

### Redis 脚本命令 ###

常用命令：

*  **直接执行一段脚本**：EVAL script numkeys key \[key …\] arg \[arg …\]
    
    参数：
    
     *  script：脚本内容，或者脚本地址
     *  numkeys：脚本中用到的 key 的数量，接下来的 numkeys 个参数会作为 key 参数，剩下的作为 arg 参数
     *  key：作为 key 的参数，会被存入脚本环境中的 KEYS 数组，角标从 1 开始
     *  arg：其它参数，会被存入脚本环境中的 ARGV 数组，角标从 1 开始
    
    示例：
    
        > eval "return 'hello world!'" 0
        hello world!
    
    其中：
    
     *  “return ‘hello world!’” ：就是脚本的内容，直接返回字符串，没有别的命令
     *  0 ：就是说没有用 key 参数，直接返回
 *  **将一段脚本编译并缓存起来，生成并返回一个SHA1值作为脚本字典的key**：SCRIPT LOAD script
    
    参数：
    
     *  script：脚本内容，或者脚本地址
    
    示例：
    
        > script load "return 'hello world!'"
        ada0bc9efe2392bdcc0083f7f8deaca2da7f32ec
    
    其中：
    
     *  返回的 `ada0bc9efe2392bdcc0083f7f8deaca2da7f32ec` 就是脚本缓存后得到的 sha1 值
        
        在脚本字典中，每一个这样的 sha1值，对应一段解析好的脚本
 *  **通过脚本的 sha1 值执行一段脚本**：EVALSHA sha1 numkeys key \[key …\] arg \[arg …\]
    
    与 EVAL 类似，区别是通过脚本的 sha1 值，去脚本缓存中查找，然后执行
    
    参数：
    
     *  sha1：就是脚本对应的 sha1 值
    
    示例：
    
        > evalsha ada0bc9efe2392bdcc0083f7f8deaca2da7f32ec 0
        hello world!

### Lua基本语法 ###

Lua 脚本遵循 Lua 的基本语法，几个常用的：

*  \*\*调用 redis 命令的两个函数：\*\*redis.call() 和 redis.pcall()
    
    区别在于 call 执行过程中出现错误会直接返回错误；pcall 则在遇到错误后，会继续向下执行。基本语法类似：
    
        redis.call("命令名称", 参数1， 参数2 ...)
    
    示例：
    
        eval "return redis.call('set', KEYS[1], ARGV[1])" 1 name Jack
    
    其中：
    
     *  ‘set’：就是执行set 命令
     *  KEYS\[1\]：从脚本环境中KEYS数组里取第一个key参数
     *  ARGV\[1\]：从脚本环境中ARGV数组里取第一个arg参数
     *  1：声明 key 只有一个，接下来的第一个参数作为 key 参数
     *  name：key 参数，会被存入到 KEYS 数组
     *  Jack：arg 参数，会被存入 ARGV 数组
 *  **条件判断语法**：if (条件语句) then …; else …; end;
    
    **变量接收语法**：local 变量名 = 变量值;
    
    示例：
    
        local val = redis.call('get', KEYS[1]);
        if (val > ARGV[1]) then 
            return 1; 
        else 
        	return 0; 
        end;
    
    基本逻辑：获取指定 key 的值，判断是否大于指定参数，如果大于则返回 1，否则返回 0
    
    示例：
    
        > set num 321
        OK
        > script load "local val = redis.call('get', KEYS[1]); if (val > ARGV[1]) then return 1; else return 0; end;"
        ad4bc448c3c264aeaa475a0407683c35bf1bc7af
        > evalsha ad4bc448c3c264aeaa475a0407683c35bf1bc7af 1 num 400
        0
    
    其中：
    
    1.  num 一开始是 321
    2.  保存脚本
    3.  然后执行并传递 num，400。判断 num 的值是否大于 400
    4.  结果返回 0

--------------------

1.  … ↩︎

[Jedic]: http://xetorthio.github.io/jedis/
[Jedis _ github]: https://github.com/xetorthio/jedis
[Link 1]: http://projects.spring.io/spring-data-redis/
[823bce51588a4b9a8f0321a62455b168.png_pic_center]: https://img-blog.csdnimg.cn/823bce51588a4b9a8f0321a62455b168.png#pic_center
[Link 2]: https://redisson.org/
[GitHub]: https://github.com/redisson/redisson
[b3e9b818db314ff2a515ccbbb396e647.png_pic_center]: https://img-blog.csdnimg.cn/b3e9b818db314ff2a515ccbbb396e647.png#pic_center
[afe88b6bb2e243a28c2cacc68cd47890.png_pic_center]: https://img-blog.csdnimg.cn/afe88b6bb2e243a28c2cacc68cd47890.png#pic_center