1.缓存Redis实战操作记录
文章目录
缓存解决并发问题
1.如何解决高并发
1.1 高并发详情页处理
- 加服务器做集群，使用
nginx做负载均衡 - 使用缓存，减少数据库的IO操作
- 限流、熔断、降级
1.2 缓存使用的思路
- 连接缓存
- 查询缓存
- 如果缓存中没有，查询mysql
- mysql查询结果存入redis
1.3 数据存储策略
企业中的存储策略（核心就是，设计key）
数据对象名：数据对象id：对象属性
User:123:passpword
User:123:username
Sku:108:info
1.4 思考:缓存问题
- 缓存穿透
- 缓存击穿
- 缓存雪崩
- 缓存数据和数据库同步的问题
- 缓存
key的过期时间 - 使用redis实现分布式锁
- 分布式锁另一种 redisson框架
2.springboot整合redis
2.1 数据类型
Redis支持五种数据类型：string（字符串），hash（哈希），list（列表），set（集合）及zset(sorted set：有序集合)。
2.2 缓存redis整合
spring工厂式整合方式.
.缓存Redis实战操作记录"
【设计模式：工厂模式】
2.2.1 加入依赖
<!-- https://mvnrepository.com/artifact/redis.clients/jedis -->
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>2.9.0</version>
</dependency>2.2.2 写一个reids的工具类(用来将redis的池初始化到spring容器中)
public class RedisUtil {
private JedisPool jedisPool;
public void initPool(String host,int port ,int database){
JedisPoolConfig poolConfig = new JedisPoolConfig();
poolConfig.setMaxTotal(200);
poolConfig.setMaxIdle(30);
poolConfig.setBlockWhenExhausted(true);
poolConfig.setMaxWaitMillis(10*1000);
poolConfig.setTestOnBorrow(true);
jedisPool=new JedisPool(poolConfig,host,port,20*1000);
}
public Jedis getJedis(){
Jedis jedis = jedisPool.getResource();
return jedis;
}
}2.2.3 写一个spring整合redis的配置类
将redis的链接池创建到spring的容器中
@Configuration
public class RedisConfig {
//读取配置文件中的redis的ip地址
@Value("${spring.redis.host:disabled}")
private String host = “192.168.222.20”;
@Value("${spring.redis.port:0}")
private int port = “6179”;
@Value("${spring.redis.database:0}")
private int database;
@Bean
public RedisUtil getRedisUtil(){
if(host.equals("disabled")){
return null;
}
RedisUtil redisUtil=new RedisUtil();
redisUtil.initPool(host,port,database);
return redisUtil;
}
}@ComponentScan(value = {"com.atguigu.gmall.config"})
//记得要注入到spring容器中2.3.4 补充:
springboot整合redis
1.导入jar包依赖
<!-- springboot整合redis -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>2.加入redis配置
#redis
# Redis数据库索引（默认为0）
spring.redis.database=0
# Redis服务器地址
spring.redis.host=192.168.3.159
# Redis服务器连接端口
spring.redis.port=6379
# Redis服务器连接密码（默认为空）
spring.redis.password=
# 连接池最大连接数（使用负值表示没有限制）
spring.redis.pool.max-active=2000
# 连接池最大阻塞等待时间（使用负值表示没有限制）
spring.redis.pool.max-wait=6000
# 连接池中的最大空闲连接
spring.redis.pool.max-idle=300
# 连接池中的最小空闲连接
spring.redis.pool.min-idle=100
# 连接超时时间（毫秒）
spring.redis.timeout=300003.RedisUtils
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
//TODO 不用写Redisconfig配置类
@Component
public class RedisUtils {
@Autowired
private RedisTemplate<String, Object> redisTemplate;
/* * 指定缓存失效时间 * @param key 键 * @param time 时间(秒) * @return */
public boolean expire(String key, long time) {
try {
if (time > 0) {
redisTemplate.expire(key, time, TimeUnit.SECONDS);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/** * 根据key 获取过期时间 * * @param key 键 不能为null * @return 时间(秒) 返回0代表为永久有效 */
public long getExpire(String key) {
return redisTemplate.getExpire(key, TimeUnit.SECONDS);
}
/** * 53 * 判断key是否存在 * 54 * * @param key 键 * 55 * @return true 存在 false不存在 * 56 */
public boolean hasKey(String key) {
try {
return redisTemplate.hasKey(key);
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/** * 删除缓存 * * @param key 可以传一个值 或多个 */
@SuppressWarnings("unchecked")
public void del(String... key) {
if (key != null && key.length > 0) {
if (key.length == 1) {
redisTemplate.delete(key[0]);
} else {
redisTemplate.delete(CollectionUtils.arrayToList(key));
}
}
}
//============================String=============================
/** * 普通缓存获取 * * @param key 键 * @return 值 */
public Object get(String key) {
return key == null ? null : redisTemplate.opsForValue().get(key);
}
/** * 普通缓存放入 * * @param key 键 * @param value 值 * @return true成功 false失败 */
public boolean set(String key, Object value) {
try {
redisTemplate.opsForValue().set(key, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/** * 普通缓存放入并设置时间 * * @param key 键 * @param value 值 * @param time 时间(秒) time要大于0 如果time小于等于0 将设置无限期 * @return true成功 false 失败 */
public boolean set(String key, Object value, long time) {
try {
if (time > 0) {
redisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS);
} else {
set(key, value);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/** * 递增 * * @param key 键 * @param delta 要增加几(大于0) * @return */
public long incr(String key, long delta) {
if (delta < 0) {
throw new RuntimeException("递增因子必须大于0");
}
return redisTemplate.opsForValue().increment(key, delta);
}
/** * 递减 * * @param key 键 * @param delta 要减少几(小于0) * @return */
public long decr(String key, long delta) {
if (delta < 0) {
throw new RuntimeException("递减因子必须大于0");
}
return redisTemplate.opsForValue().increment(key, -delta);
}
// ================================Map=================================
/** * HashGet * * @param key 键 不能为null * @param item 项 不能为null * @return 值 */
public Object hget(String key, String item) {
return redisTemplate.opsForHash().get(key, item);
}
/** * 获取hashKey对应的所有键值 * * @param key 键 * @return 对应的多个键值 */
public Map<Object, Object> hmget(String key) {
return redisTemplate.opsForHash().entries(key);
}
/** * HashSet * * @param key 键 * @param map 对应多个键值 * @return true 成功 false 失败 */
public boolean hmset(String key, Map<String, Object> map) {
try {
redisTemplate.opsForHash().putAll(key, map);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/** * HashSet 并设置时间 * * @param key 键 * @param map 对应多个键值 * @param time 时间(秒) * @return true成功 false失败 */
public boolean hmset(String key, Map<String, Object> map, long time) {
try {
redisTemplate.opsForHash().putAll(key, map);
if (time > 0) {
expire(key, time);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/** * 向一张hash表中放入数据,如果不存在将创建 * * @param key 键 * @param item 项 * @param value 值 * @return true 成功 false失败 */
public boolean hset(String key, String item, Object value) {
try {
redisTemplate.opsForHash().put(key, item, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/** * 向一张hash表中放入数据,如果不存在将创建 * * @param key 键 * @param item 项 * @param value 值 * @param time 时间(秒) 注意:如果已存在的hash表有时间,这里将会替换原有的时间 * @return true 成功 false失败 */
public boolean hset(String key, String item, Object value, long time) {
try {
redisTemplate.opsForHash().put(key, item, value);
if (time > 0) {
expire(key, time);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/** * 删除hash表中的值 * * @param key 键 不能为null * @param item 项 可以使多个 不能为null */
public void hdel(String key, Object... item) {
redisTemplate.opsForHash().delete(key, item);
}
/** * 判断hash表中是否有该项的值 * * @param key 键 不能为null * @param item 项 不能为null * @return true 存在 false不存在 */
public boolean hHasKey(String key, String item) {
return redisTemplate.opsForHash().hasKey(key, item);
}
/** * hash递增 如果不存在,就会创建一个 并把新增后的值返回 * * @param key 键 * @param item 项 * @param by 要增加几(大于0) * @return */
public double hincr(String key, String item, double by) {
return redisTemplate.opsForHash().increment(key, item, by);
}
/** * hash递减 * * @param key 键 * @param item 项 * @param by 要减少记(小于0) * @return */
public double hdecr(String key, String item, double by) {
return redisTemplate.opsForHash().increment(key, item, -by);
}
// ============================set=============================
/** * 根据key获取Set中的所有值 * * @param key 键 * @return */
public Set<Object> sGet(String key) {
try {
return redisTemplate.opsForSet().members(key);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
/** * 根据value从一个set中查询,是否存在 * * @param key 键 * @param value 值 * @return true 存在 false不存在 */
public boolean sHasKey(String key, Object value) {
try {
return redisTemplate.opsForSet().isMember(key, value);
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/** * 将数据放入set缓存 * * @param key 键 * @param values 值 可以是多个 * @return 成功个数 */
public long sSet(String key, Object... values) {
try {
return redisTemplate.opsForSet().add(key, values);
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
/** * 将set数据放入缓存 * * @param key 键 * @param time 时间(秒) * @param values 值 可以是多个 * @return 成功个数 */
public long sSetAndTime(String key, long time, Object... values) {
try {
Long count = redisTemplate.opsForSet().add(key, values);
if (time > 0)
expire(key, time);
return count;
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
/** * 获取set缓存的长度 * * @param key 键 * @return */
public long sGetSetSize(String key) {
try {
return redisTemplate.opsForSet().size(key);
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
/** * 移除值为value的 * * @param key 键 * @param values 值 可以是多个 * @return 移除的个数 */
public long setRemove(String key, Object... values) {
try {
Long count = redisTemplate.opsForSet().remove(key, values);
return count;
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
// ===============================list=================================
/** * 获取list缓存的内容 * * @param key 键 * @param start 开始 * @param end 结束 0 到 -1代表所有值 * @return */
public List<Object> lGet(String key, long start, long end) {
try {
return redisTemplate.opsForList().range(key, start, end);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
/** * 获取list缓存的长度 * * @param key 键 * @return */
public long lGetListSize(String key) {
try {
return redisTemplate.opsForList().size(key);
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
/** * 通过索引 获取list中的值 * * @param key 键 * @param index 索引 index>=0时， 0 表头，1 第二个元素，依次类推；index<0时，-1，表尾，-2倒数第二个元素，依次类推 * @return */
public Object lGetIndex(String key, long index) {
try {
return redisTemplate.opsForList().index(key, index);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
/** * 将list放入缓存 * * @param key 键 * @param value 值 * @return */
public boolean lSet(String key, Object value) {
try {
redisTemplate.opsForList().rightPush(key, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/** * 将list放入缓存 * * @param key 键 * @param value 值 * @param time 时间(秒) * @return */
public boolean lSet(String key, Object value, long time) {
try {
redisTemplate.opsForList().rightPush(key, value);
if (time > 0)
expire(key, time);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/** * 将list放入缓存 * * @param key 键 * @param value 值 * @return */
public boolean lSet(String key, List<Object> value) {
try {
redisTemplate.opsForList().rightPushAll(key, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/** * 将list放入缓存 * * @param key 键 * @param value 值 * @param time 时间(秒) * @return */
public boolean lSet(String key, List<Object> value, long time) {
try {
redisTemplate.opsForList().rightPushAll(key, value);
if (time > 0)
expire(key, time);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/** * 根据索引修改list中的某条数据 * * @param key 键 * @param index 索引 * @param value 值 * @return */
public boolean lUpdateIndex(String key, long index, Object value) {
try {
redisTemplate.opsForList().set(key, index, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/** * 移除N个值为value * * @param key 键 * @param count 移除多少个 * @param value 值 * @return 移除的个数 */
public long lRemove(String key, long count, Object value) {
try {
Long remove = redisTemplate.opsForList().remove(key, count, value);
return remove;
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
}3、缓存问题
3.1 缓存穿透
是利用redis和mysql机制(redis缓存一旦不存在，就访问mysql),直接绕过缓存访问mysql，而直到的db请求压力.一般在代码中防止该现象发生。
- 穿透:利用不存在的key,绕过redis直接攻击db —-将空值存入到redis
@Autowired
RedisUtil redisUtil;
@Override
public PmsSkuInfo findSkuInfo(String skuId) {
//TODO SkuServiceImpl : 记得处理数据同步问题。
PmsSkuInfo pmsSkuInfo = null;
//连接缓存
Jedis jedis = redisUtil.getJedis();
//查询缓存
String skuKey = "sku:"+skuId+":info"; //TODO 写在常量类中，或者配置文件中
String skuJson = jedis.get(skuKey);
if(StringUtils.isNotBlank(skuJson)){
pmsSkuInfo = JSON.parseObject(skuJson,PmsSkuInfo.class);
return pmsSkuInfo;
}
//mysql查询结果存入redis中 --------------------
String lockKey = "sku:"+skuId+":lock";
//设置分布式锁 如果不加过期时间会出现死锁现象
String ok = jedis.set(lockKey, "1", "nx", "px", 10*1000/*拿到锁的线程有10s过期时间*/);
if(StringUtils.isNotBlank(ok)&&ok.equals("OK")){
//设置成功，有权在10秒的过期时间内访问数据
pmsSkuInfo = getSkuInfoFromDb(skuId);
if(pmsSkuInfo !=null){
jedis.set(skuKey,JSON.toJSONString(pmsSkuInfo));
}else{
//数据库中不存在该sku
//为了防止缓存穿透，将null 或者空字符串设置给redis
jedis.setex(skuKey,60*3,JSON.toJSONString(""));
}
//TODO 从数据库查询到了数据 和没有查询到数据 之后需要释放锁释放锁
jedis.del(lockKey);
}else{
//设置失败 自旋:该线程在睡眠几秒后，重写尝试访问。
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
//TODO 注意这里的return ,如果不加return,程序会重新开启一个线程，和当前线程没有关系。
//return 一种结束方法 另一种返回
return findSkuInfo(skuId);
}
//--------------------------------------------
jedis.close();
return pmsSkuInfo;
}3.2 缓存击穿
是某一个热点key在高并发访问的情况下，突然失效，导致大量的并发打进mysql数据库的情况
解决:Redis分布式锁。
- 击穿:热点key失效，大量的并发打进mysql数据库。—-分布式锁
3.3 缓存雪崩
缓存时采用了相同的过期时间，导致缓存在某一时刻同时失效，导致的db崩溃
解决：设置不同的缓存失效时间
- 雪崩：很多key集体失效，导致数据库负载过重宕机。—-设置不同的缓存失效时间
3.4 区别
共同点：缓存失效,直接打进mysql。
不同的:
- 穿透:利用不存在的key,绕过redis直接攻击db —-将空值存入到redis
- 击穿:热点key失效。—-分布式锁
- 雪崩：很多key集体失效，导致数据库负载过重宕机。—-设置不同的缓存失效时间
缓存失效，使用分布式锁，限制对mysql请求。
4.分布式锁
- 第一种分布式锁:Redis自带一个分布式锁,set ex nx
- 第二种分布式锁:redisson 一个redis的带有juc的lock功能的客户端的实现(既有jedis的功能，又有juc的锁功能)
核心:节点唯一,zk分布式锁也是这个原理
4.1 Redis分布式锁
setnx命令,只在键不存在时， 才对键进行设置操作
setnx —-上锁，那么什么时候释放锁了。
set sku:107:info 1 px 1000 nx
sku:107:info表示key
1表示value
px:设置过期时间，单位为毫秒 , 1000ms
删除key:del sku:107:info
value 使用一个唯一值。
.缓存Redis实战操作记录"
@Autowired
RedisUtil redisUtil;
@Override
public PmsSkuInfo findSkuInfo(String skuId) {
//TODO SkuServiceImpl : 记得处理数据同步问题。
PmsSkuInfo pmsSkuInfo = null;
//连接缓存
Jedis jedis = redisUtil.getJedis();
//查询缓存
String skuKey = "sku:"+skuId+":info"; //TODO 写在常量类中，或者配置文件中
String skuJson = jedis.get(skuKey);
if(StringUtils.isNotBlank(skuJson)){
pmsSkuInfo = JSON.parseObject(skuJson,PmsSkuInfo.class);
return pmsSkuInfo;
}
String token = UUID.randomUUID().toString();// 为了防止删除锁删除错误。
//mysql查询结果存入redis中 --------------------
String lockKey = "sku:"+skuId+":lock";
//设置分布式锁 如果不加过期时间会出现死锁现象
String ok = jedis.set(lockKey, token, "nx", "px", 10*1000/*拿到锁的线程有10s过期时间*/);
if(StringUtils.isNotBlank(ok)&&ok.equals("OK")){
//设置成功，有权在10秒的过期时间内访问数据
pmsSkuInfo = getSkuInfoFromDb(skuId);
if(pmsSkuInfo !=null){
jedis.set(skuKey,JSON.toJSONString(pmsSkuInfo));
}else{
//数据库中不存在该sku
//为了防止缓存穿透，将null 或者空字符串设置给redis
jedis.setex(skuKey,60*3,JSON.toJSONString(""));
}
//TODO 从数据库查询到了数据 和没有查询到数据 之后需要释放锁释放锁;
String lockToken = jedis.get(lockKey);
if(StringUtils.isNotBlank(lockToken)&&lockToken.equals(token)){
//TODO 使用锁的value 唯一值，是为了在删除分布式锁的时候是删除的是自己的锁。
//TODO 使用lua 脚本，在查询到key的同时删除该key，防止高并发下的意外的发生
String script ="if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
jedis.eval(script, Collections.singletonList("lockKey"),Collections.singletonList(token));
jedis.del(lockKey); //TODO 执行到132行的时候，锁过期了。 在get那个token 和删除锁有一个时间差。----解决方案lua脚本
}
}else{
//设置失败 自旋:该线程在睡眠几秒后，重写尝试访问。
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
//TODO 注意这里的return ,如果不加return,程序会重新开启一个线程，和当前线程没有关系。
//return 一种结束方法 另一种返回
return findSkuInfo(skuId);
}
//--------------------------------------------
jedis.close();
return pmsSkuInfo;
}
.缓存Redis实战操作记录"
4.1.1 问题
lua脚本.lockKey
String script ="if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
jedis.eval(script, Collections.singletonList("lockKey"),Collections.singletonList(token));
.缓存Redis实战操作记录"
4.2 分布式锁框架
https://github.com/redisson/redisson/wiki/1.-Overview
Redisson是一个在Redis的基础上实现的java控制redis各种各样的工具的集合，包括分布式锁和多线程。
4.2.1 整合
- 加入依赖
<!-- https://mvnrepository.com/artifact/org.redisson/redisson -->
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.10.5</version>
</dependency>- 加入配置
spring.redis.host=192.168.159.130
spring.redis.port=6379- 加入配置类
@Configuration
public class GmallRedissonConfig {
@Value("${spring.redis.host}")
private String host;
@Value("${spring.redis.port}")
private String port;
@Bean
public RedissonClient redissonClient(){
Config config = new Config();
config.useSingleServer().setAddress("redis://"+host+":"+port);
RedissonClient redisson = Redisson.create(config);
return redisson;
}
}4.2.2 redisson锁
可重入锁ReentrantLock
基于Redis的Redisson分布式可重入锁RLock Java对象实现了java.util.concurrent.locks.Lock接口。同时还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。
Redisson实现了juc的lock锁，并且可以在分布式的redis环境下使用
redisson默认给的是重入锁。
.缓存Redis实战操作记录"
/** * 测试Redisson 的Demo */
@Controller
public class TestRedissonDemo {
@Autowired
RedisUtil redisUtil;
@Autowired
RedissonClient redissonClient;
@RequestMapping("/testRedisson")
@ResponseBody
public String testRedisson(){
RLock lock = redissonClient.getLock("anyLock");//anyLock 锁的名称
lock.lock();
lock.unlock();
return "success";
}
}另外Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了。
// 加锁以后10秒钟自动解锁
// 无需调用unlock方法手动解锁
lock.lock(10, TimeUnit.SECONDS);
// 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
if (res) {
try {
...
} finally {
lock.unlock();
}
}发布者：全栈程序员栈长，转载请注明出处：https://javaforall.cn/100750.html原文链接：https://javaforall.cn
本文参与 腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。
原始发表：2021年7月1，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读
目录
相关产品与服务
云数据库 Redis
腾讯云数据库 Redis（TencentDB for Redis）是腾讯云打造的兼容 Redis 协议的缓存和存储服务。丰富的数据结构能帮助您完成不同类型的业务场景开发。支持主从热备，提供自动容灾切换、数据备份、故障迁移、实例监控、在线扩容、数据回档等全套的数据库服务。
腾讯云开发者
