0
点赞
收藏
分享

微信扫一扫

听GPT 讲Rust源代码--src/tools(7)

小月亮06 2023-12-04 阅读 45

前言:

处理Redis常见问题与提高Redis缓存性能

一、Redis作为缓存常见问题及其处理方案

1)缓存穿透

根源:请求不断的查询一个不存在的key,缓存层和存储层都不会命中。

解决方案   

  • 对接口参数进行校验、防止出现恶意攻击;
  • 查询不到值时,将value设置成一个标记为加入缓存中,下次再查询就返回一个标记数而不必经过数据库,例如查询id为5的商品,不存在则返回一个-9999,然后在做逻辑判断,但是需要设置一个较短的缓存有效时间,防止以后key对应的value有数据的时候仍然返回空造成错误。
  • 使用bitmap类型定义一个可以访问的白名单,id作为偏移量。
  • 采用布隆过滤器

2)缓存击穿

根源:缓存击穿是指对于一些设置了过期时间的key,这些key可能在某些时间被超高并发访问,是一种’热点‘数据,然后在这个数据被访问前正好key失效了,那么对这个key的查询会全部转到数据库上,造成数据库压力增大导致卡顿崩溃的现象。

解决方案:

  • 设置热点数据永不过期;
  • 加锁,大量并发只让一个人去查,其他人等待,直到以后释放锁,其他人读取到锁先查缓存。

3)缓存雪崩

根源:大量的热数据key同时过期,过期之后涌入大量请求,导致请求直接访问数据库,骤增数据库压力。

解决方案:

  • 设置热点数据永不过期;
  • 将缓存过期时间设置成某一段时间内的随机数,这样就不会同时过期;
  • 分布式处理缓存,将缓存存在不同的地方
  • 依赖隔离组件为后端限流熔断并降级。比如使用Sentinel或Hystrix限流降级组件

4)缓存与数据不一致问题

1、双写不一致情况(修改数据更新缓存)

线程1先写入了数据库,这时候准备更新缓存,但是因为某原因导致出现延迟,此时线程二快速将新数据写入数据库,并且成功更新了缓存,完事之后线程1恢复了速度开始更新缓存,就导致了线程2是最后写入数据的,但是缓存的内容还是旧值,从而达到双写不一致的错误场景

2、读写并发不一致(修改数据删除缓存)

线程一先写入数据10,并删除了缓存,之后线程三读取数据,发现缓存为空,于是去查询数据库,而此时查询数据库的时间较长,与此同时线程二写入数据6,又删除了缓存,在这之后线程三也读成功更新了缓存,造成了数据库的结果是6而缓存的结果是10这种错误情况

解决方案:

1、对于并发几率很小的数据(如个人维度的订单数据、用户数据等),这种几乎不用考虑这个问题,很少会发生缓存不一致,可以给缓存数据加上过期时间,每隔一段时间触发读的主动更新即可。

2、就算并发很高,如果业务上能容忍短时间的缓存数据不一致(如商品名称,商品分类菜单等),缓存加上过期时间依然可以解决大部分业务对于缓存的要求。

3、如果不能容忍缓存数据不一致,可以通过加分布式读写锁保证并发读写或写写的时候按顺序排好队,读读的时候相当于无锁。

4、也可以用阿里开源的canal通过监听数据库的binlog日志及时的去修改缓存,但是引入了新的中间件,增加了系统的复杂度。

0

二、针对不同热度的数据采用不同的处理方式

1)热点数据

处理方案:

1、缓存永不过期

2、缓存读延期功能

当命中缓存的时候,设置key的过期时间为默认时间,相当于时间设满,设置过期时间所需要的时间是非常非常少的,对性能的影响也是微乎其微。对于热数据的获取可以实现无线续期的效果

2)冷门数据

处理方案:
针对冷门数据最好不进行缓存,避免内存浪费以及无意义的缓存在过期

基础缓存代码分析

源码与图示

很基础的Redis工具类

@Component
public class RedisUtil {

    @Autowired
    private RedisTemplate redisTemplate;

    public void set(String key, Object value) {
        redisTemplate.opsForValue().set(key, value);
    }

    public void set(String key, Object value, long timeout, TimeUnit unit) {
        redisTemplate.opsForValue().set(key, value, timeout, unit);
    }

    public boolean setIfAbsent(String key, Object value, long timeout, TimeUnit unit) {
        return redisTemplate.opsForValue().setIfAbsent(key, value, timeout, unit);
    }

    public <T> T get(String key, Class<?> T) {
        return (T) redisTemplate
                .opsForValue().get(key);
    }

    public String get(String key) {
        return (String) redisTemplate
                .opsForValue().get(key);
    }

    public Long decr(String key) {
        return redisTemplate
                .opsForValue().decrement(key);
    }

    public Long decr(String key, long delta) {
        return redisTemplate
                .opsForValue().decrement(key, delta);
    }

    public Long incr(String key) {
        return redisTemplate
                .opsForValue().increment(key);
    }

    public Long incr(String key, long delta) {
        return redisTemplate
                .opsForValue().increment(key, delta);
    }

    public void expire(String key, long time, TimeUnit unit) {
        redisTemplate.expire(key, time, unit);
    }

}

代码:

@Service
public class ProductService {

    @Autowired
    private ProductDao productDao;

    @Autowired
    private RedisUtil redisUtil;

    @Autowired
    private Redisson redisson;

    public static final Integer PRODUCT_CACHE_TIMEOUT = 60 * 60 * 24;
    public static final String EMPTY_CACHE = "{}";
    public static final String LOCK_PRODUCT_HOT_CACHE_PREFIX = "lock:product:hot_cache:";
    public static final String LOCK_PRODUCT_UPDATE_PREFIX = "lock:product:update:";
    //新增数据
    @Transactional
    public Product create(Product product) {
        Product productResult = productDao.create(product);
        redisUtil.set(RedisKeyPrefixConst.PRODUCT_CACHE + productResult.getId(), JSON.toJSONString(productResult),
                genProductCacheTimeout(), TimeUnit.SECONDS);
        return productResult;
    }
    //修改数据
    @Transactional
    public Product update(Product product) {
        Product productResult = null;
        RReadWriteLock readWriteLock = redisson.getReadWriteLock(LOCK_PRODUCT_UPDATE_PREFIX + product.getId());
        RLock writeLock = readWriteLock.writeLock();
        writeLock.lock();
        try {
            productResult = productDao.update(product);
            redisUtil.set(RedisKeyPrefixConst.PRODUCT_CACHE + productResult.getId(), JSON.toJSONString(productResult),
                    genProductCacheTimeout(), TimeUnit.SECONDS);
        } finally {
            writeLock.unlock();
        }
        return productResult;
    }
    //读数据方法
    public Product get(Long productId) throws InterruptedException {
        Product product = null;
        String productCacheKey = RedisKeyPrefixConst.PRODUCT_CACHE + productId;
        //读取缓存中的数据,具体方法实现看源码 getProductFromCache
        product = getProductFromCache(productCacheKey);
        if (product != null) {
            //此处需要和前端进行约定,如果对象的ID为空,则需要提示商品不存在
            return product;
        }
        //DCL 如果存在很高的并发量,导致竞争锁耗时过程可以采用定时阻塞的型式
        //需要精确预估执行完后面代码所需要的时候,然后将该值设置为过期时间,时间一过线程就可以继续执行
        RLock hotCacheLock = redisson.getLock(LOCK_PRODUCT_HOT_CACHE_PREFIX + productId);
        hotCacheLock.lock();
        try {
            //再次尝试从缓存中获取数据,避免其他线程已经读取过db而这边线程又重复读取
            product = getProductFromCache(productCacheKey);
            if (product != null) {
                return product;
            }
                //从数据库中读取数据
                product = productDao.get(productId);
                //读取到的数据不为空,则将数据存入redis中。
                if (product != null) {
                    redisUtil.set(productCacheKey, JSON.toJSONString(product),
                            genProductCacheTimeout(), TimeUnit.SECONDS);
                } else {
                    //当数据为空,则存入一个特俗字符,代表空数据,避免缓存穿透
                    //针对特俗key使用较短的过期时间,可以避免短时间黑客反复攻击,看能避免长时间造成的内存浪费
                    redisUtil.set(productCacheKey, EMPTY_CACHE, genEmptyCacheTimeout(), TimeUnit.SECONDS);
                }

        } finally {
            hotCacheLock.unlock();
        }
        return product;
    }
        //从缓存中读取数据
    private Product getProductFromCache(String productCacheKey) {
        Product product = null;

        String productStr = redisUtil.get(productCacheKey);
        if (!StringUtils.isEmpty(productStr)) {
            if (EMPTY_CACHE.equals(productStr)) {
                //未查询到数据,需要设置一个空对象返回,并设置较短的过期时间
                redisUtil.expire(productCacheKey, genEmptyCacheTimeout(), TimeUnit.SECONDS);
                return new Product();
            }
            //如果真是查询到数据,设置读延期
            product = JSON.parseObject(productStr, Product.class);
            redisUtil.expire(productCacheKey, genProductCacheTimeout(), TimeUnit.SECONDS); //读延期
        }
        return product;
    }
    //设置较长的过期时间
    private Integer genProductCacheTimeout() {
        return PRODUCT_CACHE_TIMEOUT + new Random().nextInt(5) * 60 * 60;
    }
    //设置较短的过期时间
    private Integer genEmptyCacheTimeout() {
        return 60 + new Random().nextInt(30);
    }



}

良好的Redis使用习惯

一、键值设计

1)key名设计

  • (1)【建议】: 可读性和可管理性

以业务名(或数据库名)为前缀(防止key冲突),用冒号分隔,比如业务名:表名:id

trade:order:1

  • (2)【建议】:简洁性

保证语义的前提下,控制key的长度,当key较多时,内存占用也不容忽视,例如:

user:{uid}:friends:messages:{mid} 简化为 u:{uid}:fr:m:{mid}

  • (3)【强制】:不要包含特殊字符

反例:包含空格、换行、单双引号以及其他转义字符

2) value设计

(1)【强制】:拒绝bigkey(防止网卡流量、慢查询)

在Redis中,一个字符串最大512MB,一个二级数据结构(例如hash、list、set、zset)可以存储大约40亿个(2^32-1)个元素,但实际中如果下面两种情况,我就会认为它是bigkey。

  1. 字符串类型:它的big体现在单个value值很大,一般认为超过10KB就是bigkey。
  2. 非字符串类型:哈希、列表、集合、有序集合,它们的big体现在元素个数太多。

一般来说,string类型控制在10KB以内,hash、list、set、zset元素个数不要超过5000。

反例:一个包含200万个元素的list。

非字符串的bigkey,不要使用del删除,使用hscan、sscan、zscan方式渐进式删除,同时要注意防止bigkey过期时间自动删除问题(例如一个200万的zset设置1小时过期,会触发del操作,造成阻塞)

(2)【推荐】:选择适合的数据类型。

例如:实体类型(要合理控制和使用数据结构,但也要注意节省内存和性能之间的平衡)

反例:

set user:1:name tom set user:1:age 19 set user:1:favor football

正例:

hmset user:1 name tom age 19 favor football

(3)【推荐】:控制key的生命周期,redis不是垃圾桶。

建议使用expire设置过期时间(条件允许可以打散过期时间,防止集中过期)。

二、命令使用

1. O(N)命令关注N的数量

例如hgetall、lrange、smembers、zrange、sinter等并非不能使用,但是需要明确N的值。有遍历的需求可以使用hscan、sscan、zscan代替。

2.:禁用命令

禁止线上使用keys、flushall、flushdb等,通过redis的rename机制禁掉命令,或者使用scan的方式渐进式处理。

3.合理使用select

redis的多数据库较弱,使用数字进行区分,很多客户端支持较差,同时多业务用多数据库实际还是单线程处理,会有干扰。

4.使用批量操作提高效率

原生命令:例如mget、mset。 非原生命令:可以使用pipeline提高效率。

但要注意控制一次批量操作的元素个数(例如500以内,实际也和元素字节数有关)。

5.Redis事务功能较弱,不建议过多使用,可以用lua替代

三、客户端处理

1.避免多个应用使用一个Redis实例

正例:不相干的业务拆分,公共数据做服务化。

2.使用带有连接池的数据库,可以有效控制连接,同时提高效率,标准使用方式:

1 JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
2 jedisPoolConfig.setMaxTotal(5);
3 jedisPoolConfig.setMaxIdle(2);
4 jedisPoolConfig.setTestOnBorrow(true);
5
6 JedisPool jedisPool = new JedisPool(jedisPoolConfig, "192.168.0.60", 6379, 3000, null);
7
8 Jedis jedis = null;
9 try {
10 jedis = jedisPool.getResource();
11 //具体的命令
12 jedis.executeCommand()
13 } catch (Exception e) {
14 logger.error("op key {} error: " + e.getMessage(), key, e);
15 } finally {
16 //注意这里不是关闭连接,在JedisPool模式下,Jedis会被归还给资源池。
17 if (jedis != null)
18 jedis.close();
19 }

3.高并发下建议客户端添加熔断功能(例如sentinel、hystrix)

4.设置合理的密码,如有必要可以使用SSL加密访问

5.Redis对于过期键有三种清除策略:

  1. 被动删除:当读/写一个已经过期的key时,会触发惰性删除策略,直接删除掉这个过期key
  2. 主动删除:由于惰性删除策略无法保证冷数据被及时删掉,所以Redis会定期(默认每100ms)主动淘汰一批已过期的key,这里的一批只是部分过期key,所以可能会出现部分key已经过期但还没有被清理掉的情况,导致内存并没有被释放
  3. 当前已用内存超过maxmemory限定时,触发主动清理策略

举报

相关推荐

0 条评论