-
Redis 的 Java 客户端（Jedis、SpringDataRedis、SpringCache、Redisson）基本操作指南
Jedis
参考：
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.clientsgroupId> <artifactId>jedisartifactId> <version>2.9.0version> dependency>- 1
- 2
- 3
- 4
- 5
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)- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
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; } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
Spring Data Redis
概述、依赖
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 自动配置原理
依赖
<dependency> <groupId>org.springframework.bootgroupId> <artifactId>spring-boot-starter-data-redisartifactId> dependency>- 1
- 2
- 3
- 4
配置文件及配置类
单机版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- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
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; }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
- 115
- 116
- 117
- 118
- 119
- 120
- 121
- 122
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>- 1
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); } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
- 115
- 116
- 117
- 118
- 119
- 120
- 121
- 122
- 123
- 124
- 125
- 126
- 127
- 128
- 129
- 130
- 131
- 132
- 133
- 134
- 135
- 136
- 137
- 138
- 139
- 140
- 141
- 142
- 143
- 144
- 145
- 146
- 147
- 148
- 149
- 150
- 151
- 152
- 153
- 154
- 155
- 156
- 157
- 158
- 159
- 160
- 161
- 162
- 163
- 164
- 165
- 166
- 167
- 168
- 169
- 170
- 171
- 172
- 173
- 174
- 175
- 176
- 177
- 178
- 179
- 180
- 181
- 182
- 183
- 184
- 185
- 186
- 187
- 188
- 189
- 190
- 191
- 192
- 193
- 194
- 195
- 196
- 197
- 198
- 199
- 200
- 201
- 202
- 203
- 204
- 205
- 206
- 207
- 208
- 209
- 210
- 211
- 212
- 213
- 214
- 215
- 216
- 217
- 218
- 219
- 220
- 221
- 222
- 223
- 224
- 225
- 226
- 227
- 228
- 229
- 230
- 231
- 232
- 233
- 234
- 235
- 236
- 237
- 238
- 239
- 240
- 241
- 242
- 243
- 244
- 245
- 246
- 247
- 248
- 249
- 250
- 251
- 252
- 253
- 254
- 255
- 256
- 257
- 258
- 259
- 260
- 261
- 262
- 263
- 264
- 265
- 266
- 267
- 268
- 269
- 270
- 271
- 272
- 273
- 274
- 275
- 276
- 277
- 278
- 279
- 280
- 281
- 282
- 283
- 284
- 285
- 286
- 287
- 288
- 289
- 290
- 291
- 292
- 293
- 294
- 295
- 296
- 297
- 298
- 299
- 300
- 301
- 302
- 303
- 304
- 305
- 306
- 307
- 308
- 309
- 310
- 311
- 312
- 313
- 314
- 315
- 316
- 317
- 318
- 319
- 320
- 321
- 322
- 323
- 324
- 325
- 326
- 327
- 328
- 329
- 330
- 331
- 332
- 333
- 334
- 335
- 336
- 337
- 338
- 339
- 340
- 341
- 342
- 343
- 344
- 345
- 346
- 347
- 348
- 349
- 350
- 351
- 352
- 353
- 354
- 355
- 356
- 357
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 对缓存穿透问题的解决方案- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
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); } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
- 115
- 116
- 117
- 118
- 119
- 120
- 121
- 122
- 123
- 124
- 125
- 126
- 127
- 128
- 129
- 130
- 131
- 132
- 133
- 134
- 135
自定义缓存解析器类
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; } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
CachingConfigurerSupport 说明
支持自定义缓存的读写机制：
-
cacheManager（缓存管理器）
默认情况，SpringBoot 会使用 SimpleCacheConfiguration 缓存配置类。然后创建一个 ConcurrentMapCacheManager 缓存管理器，可以获取 ConcurrentMap 来作为缓存组件使用。
引入 redis 的 starter 后，RedisCacheConfiguration 缓存配置类就会生效，会创建一个 RedisCacheManager
- 默认创建的 RedisCacheManager 在操作 redis 的时候 RedisTemplate
- RedisTemplate
- 想要保存为 JSON 格式就可以自定义 CacheManager
- 注：从执行时间上来看，JdkSerializationRedisSerializer 是最高效的（毕竟是JDK原生的），但是序列化的结果字符串是最长的。 JSON 由于其数据格式的紧凑性，序列化的长度是最小的，时间比前者要多一些。而 OxmSerialiabler 在时间上看是最长的（当时和使用具体的 Marshaller 有关）。故推荐使用 JacksonJsonRedisSerializer 作为 POJO 的序列器。
- 默认创建的 RedisCacheManager 在操作 redis 的时候 RedisTemplate
-
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"}- 1
- 2
-
**key **属性：缓存的键后缀
为空时默认使用
SimpleKeyGenerator（将方法的所有参数值进行组合）生成 key，如果指定要按照 SpEL 表达式编写注：完整的缓存键值对格式为：value属性值::key属性值=方法返回值
value属性值::为缓存的前缀，可通过配置文件 spring.cache.redis.key-prefix 属性指定
示例：
@Cacheable(value="testcache", key="#id")- 1
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); }- 1
- 2
- 3
- 4
- 5
- 6
-
condition 属性：缓存的条件
可以为空，使用 SpEL 编写，返回 true 或者 false
只有为 true 才进行缓存/清除缓存，在调用方法之前之后都能判断
示例：
@Cacheable(value="testcache", condition="#id.length()>2")- 1
-
unless 属性：否决缓存的条件
条件为 true 不缓存，false 才缓存
只在方法执行之后判断，此时可以拿到返回值 result 进行判断
示例：
@Cacheable(value="testcache", condition="#result == null")- 1
-
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)- 1
-
beforeInvocation 属性：是否在方法执行前就清除缓存数据
缺省为 false，缺省清空下，如果方法执行抛出异常，则不会清除缓存
如果指定为 true，则在方法还没有执行的时候就清除缓存
示例：
@CachEvict(value="testcache", beforeInvocation=true)- 1
注：作用只有一个，就是先清除缓存再执行方法
@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) {}- 1
- 2
- 3
- 4
- 5
- 6
Cache SpEL 表达式
Cache SpEL 表达式语法
名称 位置 描述 示例 methodName root object 当前被调用的方法名 #root.methodName method root object 当前被调用的方法 #root.method.name target root object 当前被调用的目标对象 #root.target targetClass root object 当前被调用的目标对象类 #root.targetClass args root object 当前被调用的方法的参数列表 #root.args[0] caches root object 当前方法调用使用的缓存列表（如@Cacheable(value={“cache1”,“cache2”})，则有两个cache） #root.caches[0].name argument name evaluation context 方法参数的名字，可以直接 #参数名，也可以使用 #p0或#a0 的形式，0代表参数的索引 #iban、#a0、#p0 result evaluation context 方法执行后的返回值（仅当方法执行之后的判断有效，如’unless’，'cache put’的表达式，'cache evict’的表达式beforeInvocation=false） #result SpEL 运算符
类型 运算符 关系运算符 < ，> ， <= ，>=，==，!=,lt，gt，le，ge，eq，ne 算数运算符 +，-，*，/ ，%，^ 逻辑运算符 &&， 条件运算符 ? : (ternary),? : (elvis) 正则表达式 matches 其他类型 ?. ，?[…] ，![…] ，1，$[…]
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），从而让使用者能够将精力更集中地放在处理业务逻辑上。
依赖及客户端配置
依赖
<dependency> <groupId>org.redissongroupId> <artifactId>redissonartifactId> <version>3.10.6version> dependency>- 1
- 2
- 3
- 4
- 5
配置 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); } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
注意：这里读取了一个名为 RedisProperties 的属性，因为引入了SpringDataRedis，Spring已经自动加载了 RedisProperties，并且读取了配置文件中的 Redis 信息。
常用 API
RedissonClient 接口
// 创建锁对象，并指定锁的名称 RLock getLock(String name)- 1
- 2
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()- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
Redis 分布式锁
Redis 分布式锁原理
分布式锁的关键是多进程共享的内存标记，因此只要在 Redis 中放置一个这样的标记就可以了。
在实现分布式锁时，注意需要实现下列目标：
-
多进程可见：多进程可见，否则就无法实现分布式效果
- redis 本身就是多服务共享的，不用过多关注
-
避免死锁：死锁的情况有很多，要考虑各种异常导致死锁的情况，保证锁可以被释放
-
服务宕机后的锁释放问题：设置锁时最好设置锁的有效期，如果服务宕机，有效期到时自动删除锁
> set lock 001 nx ex 20 OK > get lock 001 > ttl lock 10 > set lock 001 nx ex 20 null- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
-
-
排它：同一时刻，只能有一个进程获得锁
-
可以利用 Redis 的 setnx 命令（ set when not exits）来实现。当多次执行 setnx 命令时，只有第一次执行的才会成功并返回1，其它情况返回0
> setnx lock 001 1 > get lock 001 > setnx lock 001 0- 1
- 2
- 3
- 4
- 5
- 6
定义一个固定的 key，多个进程都执行 setnx，设置这个 key 的值，返回 1 的服务获取锁，返回 0 则没有获取
-
-
高可用：避免锁服务宕机或处理好宕机的补救措施
- 利用Redis的主从、哨兵、集群，保证高可用
分布式不可重入锁
流程
按照上面所述的理论，分布式锁的流程大概如下：
基本流程：
- 1、通过set命令设置锁
- 2、判断返回结果是否是OK
- 1）Nil，获取失败，结束或重试（自旋锁）
- 2）OK，获取锁成功
- 执行业务
- 释放锁
- 3、异常情况，服务宕机。超时时间EX结束，会自动释放锁
注意：释放锁时需要判断锁的value释放跟自己存进去的一致
不然下面的场景下会出现释放锁的问题：
-
三个进程：A和B和C，在执行任务，并争抢锁，此时A获取了锁，并设置自动过期时间为10s
-
A开始执行业务，因为某种原因，业务阻塞，耗时超过了10秒，此时锁自动释放了
-
B恰好此时开始尝试获取锁，因为锁已经自动释放，成功获取锁
-
A此时业务执行完毕，执行释放锁逻辑（删除key），于是B的锁被释放了，而B其实还在执行业务
-
此时进程C尝试获取锁，也成功了，因为A把B的锁删除了。
B 和 C 同时获取了锁，违反了排它性！
代码实现
定义一个锁接口：
public interface RedisLock { boolean lock(long releaseTime); void unlock(); }- 1
- 2
- 3
- 4
定义一个锁工具：
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); } } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
在定时任务中使用锁：
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(); } } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
分布式可重入锁
可重入锁概述
可重入锁，也叫做递归锁，指的是在同一线程内，外层函数获得锁之后，内层递归函数仍然可以获取到该锁。换一种说法：同一个线程再次进入同步代码时，可以使用自己已获取到的锁。
可重入锁可以避免因同一线程中多次获取锁而导致死锁发生。
如何实现可重入锁：在锁已经被使用时，判断这个锁是否是自己的，如果是则再次获取。
可以在set锁的值时，存入获取锁的线程的信息，这样下次再来时，就能知道当前持有锁的是不是自己，如果是就允许再次获取锁。
要注意，因为锁的获取是可重入的，因此必须记录重入的次数，这样不至于在释放锁时一下就释放掉，而是逐层释放。
因此，不能再使用简单的key-value结构，这里推荐使用hash结构：
- key：lock
- hashKey：线程信息
- hashValue：重入次数，默认1
释放锁时，每次都把重入次数减一，减到 0 说明多次获取锁的逻辑都执行完毕，才可以删除key，释放锁
流程图
这里重点是获取锁的流程：
下面假设锁的 key 为 “lock”，hashKey 是当前线程的 id：“threadId”，锁自动释放时间假设为 20 s
获取锁的步骤：
-
判断 lock 是否存在
EXISTS lock存在，说明锁已被获取，接下来判断是不是自己的锁
判断当前线程 id 作为 hashKey 是否存在：
HEXISTS lock threadId- 不存在，说明锁已被获取，且不是自己获取的，锁获取失败，end
- 存在，说明是自己获取的锁，重入次数+1：
HINCRBY lock threadId 1，去到步骤3
-
不存在，说明可以获取锁，
HSET key threadId 1 -
设置锁自动释放时间，
EXPIRE lock 20
释放锁的步骤：
- 判断当前线程 id 作为 hashKey 是否存在：
HEXISTS lock threadId- 不存在，说明锁已经失效，不用管了
- 存在，说明锁还在，重入次数减1：
HINCRBY lock threadId -1，获取新的重入次数
- 判断重入次数是否为 0：
- 为 0，说明锁全部释放，删除key：
DEL lock - 大于 0，说明锁还在使用，重置有效时间：
EXPIRE lock 20
- 为 0，说明锁全部释放，删除key：
上述流程有一个最大的问题，就是有大量的判断，这样在多线程运行时，会有线程安全问题，除非能保证执行命令的原子性。
常见分布式可重入锁实现：
- 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("任务执行完毕，释放锁"); } } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
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;- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
释放锁：
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;- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
Java 执行 Lua 脚本
RedisTemplate 中提供了一个方法，用来执行 Lua 脚本：
public <T> T execute(RedisScript<T> script, List<K> keys, Object... args)- 1
参数：
- 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);- 1
- 2
- 3
- 4
- 5
- 6
可以把脚本文件写到 classpath 下的某个位置，然后通过加载这个文件来获取脚本内容，并设置给 DefaultRedisScript 实例
-
方式2：通过 RedisScript 中的静态方法（需要把脚本内容写到代码中，作为参数传递，不够优雅）
static <T> RedisScript<T> of(String script) static <T> RedisScript<T> of(String script, Class<T> resultType)- 1
- 2
参数：
- String script ：Lua 脚本
- Class resultType ：返回值类型
可重入分布式锁的实现
-
在 classpath 中编写两个 Lua 脚本文件
-
定义一个类（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); } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
-
新建一个定时任务，测试重入锁：
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); } } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
拓展
缓存穿透、雪崩、击穿
-
缓存穿透
是指查询一个一定不存在的数据（缓存和数据库中均不存在该数据），每次都会去数据库查询，高并发时可能会导致数据库挂掉或者发生 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!- 1
- 2
其中：
- “return ‘hello world!’” ：就是脚本的内容，直接返回字符串，没有别的命令
- 0 ：就是说没有用 key 参数，直接返回
-
将一段脚本编译并缓存起来，生成并返回一个SHA1值作为脚本字典的key：SCRIPT LOAD script
参数：
- script：脚本内容，或者脚本地址
示例：
> script load "return 'hello world!'" ada0bc9efe2392bdcc0083f7f8deaca2da7f32ec- 1
- 2
其中：
-
返回的
ada0bc9efe2392bdcc0083f7f8deaca2da7f32ec就是脚本缓存后得到的 sha1 值在脚本字典中，每一个这样的 sha1值，对应一段解析好的脚本
-
通过脚本的 sha1 值执行一段脚本：EVALSHA sha1 numkeys key [key …] arg [arg …]
与 EVAL 类似，区别是通过脚本的 sha1 值，去脚本缓存中查找，然后执行
参数：
- sha1：就是脚本对应的 sha1 值
示例：
> evalsha ada0bc9efe2392bdcc0083f7f8deaca2da7f32ec 0 hello world!- 1
- 2
Lua基本语法
Lua 脚本遵循 Lua 的基本语法，几个常用的：
-
**调用 redis 命令的两个函数：**redis.call() 和 redis.pcall()
区别在于 call 执行过程中出现错误会直接返回错误；pcall 则在遇到错误后，会继续向下执行。基本语法类似：
redis.call("命令名称", 参数1， 参数2 ...)- 1
示例：
eval "return redis.call('set', KEYS[1], ARGV[1])" 1 name Jack- 1
其中：
- ‘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;- 1
- 2
- 3
- 4
- 5
- 6
基本逻辑：获取指定 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
- 2
- 3
- 4
- 5
- 6
其中：
- num 一开始是 321
- 保存脚本
- 然后执行并传递 num，400。判断 num 的值是否大于 400
- 结果返回 0
… ↩︎
-
相关阅读:
10分钟掌握Python缓存
推荐一款.Net Core开发的后台管理系统YiShaAdmin
测试到底是个啥
FATAL EXCEPTION: OkHttp Dispatcher
男孩姓卜取什么名字好听
微信小程序 slot 不显示
基于Java实现的免疫算法-克隆选择算法
JavaScript中this关键字实践
【Torch】torch.load( )系列语句解读解读,易学易用
【C++】基础篇
- 原文地址：https://blog.csdn.net/footless_bird/article/details/128136127
