Redis非关系型数据库

1. redis

1.1 什么是NOSQL？

nosql(not only sql)不仅仅是sql。NoSQL，泛指非关系型的数据库。非关系型的数据库则由于其本身的特点得到了非常迅速的发展

非关系数据库和关系型数据库之间的区别:

RDBMS---关系型数据

- 高度组织化结构化数据。

- 结构化查询语言（SQL） select

- 数据和关系都存储在单独的表中。

- 数据操纵语言DML，数据定义语言DDL

- 严格的一致性. 事务 ACID

- 基于事务

NoSQL--非关系型数据库---缓存数据

- 代表着不仅仅是SQL

- 没有声明性查询语言  

- 键 - 值对存储 key value

- 非结构化和不可预知的数据  

- 高性能，高可用性和可伸缩性。 适合搭建集群。

NOSQL的产品

Mongodb：

redis：

Hbase:针对大数据

1.2 redis

Redis是一种开放源代码（BSD许可）的内存中数据结构存储，用作数据库，缓存和消息代理。Redis提供数据结构，例如字符串，哈希，列表，集合，带范围查询的排序集合，位图，超日志，地理空间索引和流。Redis具有内置的复制，Lua脚本，LRU驱逐，事务和不同级别的磁盘持久性，并通过Redis Sentinel和Redis Cluster自动分区提供了高可用性。

 1.3 redis的优点

1.Redis读取的速度是110000次/s，写的速度是81000次/s

2.原子 。Redis的所有操作都是原子性的，同时Redis还支持对几个操作全并后的原子性执行。

3.支持多种数据结构：string（字符串）；list（列表）；hash（哈希），set（集合）；zset(有序集合)

4.持久化--磁盘，主从复制（集群）

5.官方不支持window系统,但是又第三方版本。 linux系统。---

1.4 如何安装redis

安装redis的依赖。
yum install -y gc-c++

解压redis安装包

进入redis解压目录
make 编译c语言
make install 安装redis

启动redis
redis-server  redis配置文件名

连接redis
redis-cli   默认连接为127.0.0.1 端口号6379

redis-cli -h ip  -p port  远程连接其他人的redis

 1.5 设置redis.conf

1.设置redis后台启动

 2.修改端口号

3.设置远程连接。

1.6 redis常用命令

Redis 命令参考 — Redis 命令参考/

1.6.1 对key操作的命令

1. 查看你所有的key      keys *

2. 为指定的key设置过期时间          expire key seconds

3. 查看key的剩余存活时间         ttl key   返回-1表示永远存在 -2不存在该key

4. 删除指定的key          delete key...  返回结果为删除的个数
5. 判断指定的key是否存在          exists key

1.6.2 对redis数据库的操作

默认redis有16个库，可以通过修改redis配置文件更改数量

select n: 切换redis库。n[0~databases-1]
flushdb: 清空当前所在的库。
flushall: 清空所有库的内容。

 1.7 redis支持的数据类型

我们使用频率最高是: Strings字符串类型，Lists列表类型，Sets集合类型，Hashes哈希类型，Sorted Sets 有序集合。这里所谓的类型，就是value的类型。

1.7.1 Strings类型

它的value值为String类型，在实际开发中，它可以存储任意的数据类型。因为任何对象可以转换为json字符串。它的默认存放的大小512M.

set key value: 存储指定key和value的值。
get key: 获取指定key的value值。
mset key value key value...:存储多个key和value的值
mget key key ...:获取多个key对应的value。
setnx key value: 如果指定的key存在，则不存入。如果不存在，则存入。
setex key second value: 存储指定的key和value并设置过期时间。
incr key: 对指定key的value递增。----点赞 收藏数 主键的递增
decr key: 对指定key的value递减

 1.7.2 Hash哈希类型

它的value值为hash类型，hash类型由field和value组成。适合存储对象。

hset key field value: 存储指定key的field和value值。
hget key field: 获取指定key的field对应的value值。
hgetall key: 获取在哈希表中指定 key 的所有字段和值
hkeys key: 获取指定key所有field名称
hvals key: 获取指定key的所有value值。
hdel key field: 删除指定key对应的field值

 1.7.3 list列表类型

它的value类型为list列表类型，它的value可以是多个值，而且这些者可以重复，有序。一般使用在消息队列。

lpush key value value....:从左边存储指定key的对应列表值。
lpop key: 移出并获取列表的第一个元素
lrange key start end: 获取列表指定范围内的元素
lindex key index: 根据下标获取指定的元素
lset key index value: 修改指定坐标的元素内容

 1.7.4 set集合类型

它的value类型是一个set集合类型，这个集合类型元素无需，且不能重复。

sadd key value....:向集合添加一个或多个成员
smembers key :返回集合中的所有成员
spop key: 随机获取某个元素并移除
sinter k1 k2.。。：    返回给定所有集合的交集

 1.7.5 sort set有序集合类型

它的value类型为一个带分数的集合类型。按照分数排序。应用在: 排行榜

zadd key score value score value.....: 往redis中添加有序集合
zrange key start end: 获取指定返回的元素
ZREVRANGE key 0 -1 WITHSCORES： 分数从高到低

1.8 redis的使用场景

1. 热点数据的缓存： 减少对数据库的访问频率和减轻数据库的压力。
2. 限时业务的运用： 秒杀  存储登录者用户信息  存储短信验证码
3. 计数器相关问题： 点赞数 收藏数 播放量
4. 排行榜相关问题： sort set
5. 分布式锁:

 1.9 redis的持久化

持久化：把内存中的数据库保存到磁盘上,防止数据的丢失。

redis支持的持久化方式两种:
  (1)RDB:快照  其实就是把数据以快照的形式保存在磁盘上,什么是快照呢，你可以理解成把当前时刻的数据拍成一张照片保存下来。
 （2)AOF :日志追加 记录服务器接受的每个写入操作，当服务器启动时再次加载该日志，会把日志中的命令重新执行一遍。

1.9.1 RDB快照持久化方式

RDB的触发方式：
                1.手动触发
                    [1] save堵塞型保存
                    [2] bgsave非堵塞型保存
                2.自动触发

默认保存的文件名：dump.rdb和redis.conf文件在同个文件夹

 save：

该命令会阻塞当前Redis服务器，执行save命令期间，Redis不能处理其他命令，直到RDB过程完成为止。具体流程如下：

执行完成时候如果存在老的RDB文件，就把新的替代掉旧的。我们的客户端可能都是几万或者是几十万，这种方式显然不可取。

bgsave:

执行该命令时，Redis会在后台异步进行快照操作，快照同时还可以响应客户端请求。具体流程如下：

自动触发rdb,修改redis的配置文件

1.9.2 AOF日志追加持久化方式

默认该文件的名称为：

 默认aof不开启

 aof模式会把每个写操作，记录到一个日志文件，当redis启动时会把该日志中每个指令重新执行一遍。 数据恢复速度慢。数据完整性高。

如果两则都使用，恢复数据时按照aof恢复。因为redis认为它的完整性比较好。大多数使用rdb.

1.10 redis集群模式

redis单机版，出现单机故障后，导致redis无法使用，如果程序使用redis，间接导致程序出错。

redis的集群模式:

1.主从复制模式

一主多从模式。一个主节点，多个从节点，那么主节点可以负责:读操作，写操作。 从节点只能负责读操作，不能负责写操作。 这样就可以把读的压力从主节点分摊到从节点，以减少主节点的压力。

当主节点执行完写命令，会把数据同步到从节点。

2.哨兵模式

由于主从模式，主节点单机后，从节点不会自动上位。 增加一个哨兵服务，该哨兵时刻监控master，如果master挂了，哨兵会在从节点中选举一位为主节点【哨兵投票机制】。

3.集群化模式

不管是主从模式还是哨兵模式，都无法解决单节点写操作的问题。如果这时写操作的并发比较高。这是可以实验集群化模式【去中心化模式】

1.11 java连接redis

1.11.1 使用redis

(1)添加jedis依赖

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

(2)代码测试

    @Test
    public void test01(){
        //连接redis--->必须保证你的redis服务运行远程连接。
        //该对象中把每个redis命令封装成对应的方法了。
        Jedis jedis=new Jedis("192.168.23.185",6380);
        //对于字符串操作的命令
        String s = jedis.set("k1", "v1");
        System.out.println(s);
        String setex = jedis.setex("k2", 10l, "v2");
        System.out.println(setex);
        Long aLong = jedis.setnx("k3", "v3");
        System.out.println(aLong);

        //对于hash操作
        jedis.hset("k4","name","Aimei");
        jedis.hset("k4","age","18");

        Map<String,String> map=new HashMap();
        map.put("name","Yelou");
        map.put("age","30");
        jedis.hset("k5",map);


        jedis.close();

    }

 1.11.2 使用连接池连接redis

public void test02(){
        //创建连接池的配置类
        JedisPoolConfig jedisPoolConfig=new JedisPoolConfig();
        jedisPoolConfig.setMaxIdle(20);
        jedisPoolConfig.setMinIdle(5);
        jedisPoolConfig.setMaxWait(Duration.ofMillis(3000));
        JedisPool jedisPool=new JedisPool(jedisPoolConfig,"192.168.23.185",6380);

        long start=System.currentTimeMillis();

        for (int i = 0; i <1000 ; i++) {
            //从jedis连接池获取资源
            Jedis jedis = jedisPool.getResource();
            String ping = jedis.ping();
            jedis.close();//是否资源到池子
        }

        long end=System.currentTimeMillis();

        System.out.println("总耗时:"+(end-start));

    }

 1.11.3 java连接redis集群模式

    @Test
    public void test03(){
        Set<HostAndPort> nodes=new HashSet<>();
        nodes.add(new HostAndPort("192.168.23.185",6001));
        nodes.add(new HostAndPort("192.168.23.185",6002));
        nodes.add(new HostAndPort("192.168.23.185",6003));
        nodes.add(new HostAndPort("192.168.23.185",6004));
        nodes.add(new HostAndPort("192.168.23.185",6005));
        nodes.add(new HostAndPort("192.168.23.185",6006));
        JedisCluster jedisCluster=new JedisCluster(nodes);

        jedisCluster.set("k1","Aimei");

        jedisCluster.close();

    }

 1.12 springboot整合redis

springboot对redis的操作封装了两个StringRedisTemplate和RedisTemplate类，StringRedisTemplate是RedisTemplate的子类，StringRedisTemplate它只能存储字符串类型，无法存储对象类型。要想用StringRedisTemplate存储对象必须把对象转为json字符串。

1.12.1 StringRedisTemplate

 (1) 引入相关的依赖

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

 (2)注入StringRedisTemplate该类对象

 @Autowired
 private StringRedisTemplate redisTemplate;

 (3)使用StringRedisTemplate

package com.zpb;

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;

import java.util.concurrent.TimeUnit;

@SpringBootTest
class RedisSpringbootApplicationTests {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Test
    public void test01(){
        //对字符串数据类型的操作ValueOperations
        ValueOperations<String, String> forValue = stringRedisTemplate.opsForValue();
        //存储字符串类型--key  value long unit  setex();
        forValue.set("k1","v1",2, TimeUnit.HOURS);
        //删除指定的key
        //redisTemplate.delete();
        //查看所有的key
        //redisTemplate.keys()
        //是否存在指定的key
        //redisTemplate.hasKey()
        
        //等价于setnx  存入成功返回true，失败返回false
        Boolean absent = forValue.setIfAbsent("k5", "v5", 30, TimeUnit.SECONDS);
        System.out.println(absent);
        Integer append = forValue.append("k5", "xx");
        String key = forValue.get("k5");
    }

    @Test
    public void test02(){
        //对hash类型的操作。
        HashOperations<String, Object, Object> forHash = redisTemplate.opsForHash();
        forHash.put("k1","name","张三");
        forHash.put("k1","age","15");
        Map<String,String> map=new HashMap<>();
        map.put("name","李四");
        map.put("age","25");
        forHash.putAll("k2",map);

        Object o = forHash.get("k1", "name");
        System.out.println(o);

        Set<Object> k1 = forHash.keys("k1");
        System.out.println(k1);
        List<Object> k11 = forHash.values("k1");
        System.out.println(k11);


        //获取k1对于的所有的field和value
        Map<Object, Object> k12 = forHash.entries("k1");
        System.out.println(k12);

    }
}

1.12.2 RedisTemplate

 

package com.zpb;

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;

@SpringBootTest
class RedisSpringbootApplicationTests02 {

     //当你存储的value类型为对象类型使用redisTemplate
     //存储的value类型为字符串。StringRedisTemplate  验证码
     @Autowired
     private RedisTemplate redisTemplate;

     @Test
     public void test01(){
         //必须认为指定序列化方式
         redisTemplate.setKeySerializer(new StringRedisSerializer());
         redisTemplate.setValueSerializer(new Jackson2JsonRedisSerializer<Object>(Object.class));

         //对String类型操作类
         ValueOperations forValue = redisTemplate.opsForValue();
         //redis中key和value都变成乱码了。
         //key和value都没有指定序列化方式，默认采用jdk的序列化方式。
         forValue.set("k1","张三");


         //value默认采用jdk，类必须实现序列化接口
         forValue.set("k2",new User(1,"刘德华",22));
     }

}

 上面的RedisTemplate需要每次都指定key value以及field的序列化方式，可以搞一个配置类，为RedisTemplate指定好序列化。以后再用就无需指定。

package com.zpb.config;

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.cache.CacheManager;
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.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.time.Duration;

@Configuration
public class RedisConfig {
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        RedisSerializer<String> redisSerializer = new StringRedisSerializer();
        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);
        template.setConnectionFactory(factory);
        //key序列化方式
        template.setKeySerializer(redisSerializer);
        //value序列化
        template.setValueSerializer(jackson2JsonRedisSerializer);
        //value hashmap序列化  filed value
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
        template.setHashKeySerializer(redisSerializer);
        return template;
    }
}

1.13 redis的使用场景

1.13.1 作为缓存

（1）数据存储在内存中，数据查询速度快。可以分摊数据库压力。

 (2)什么样的数据适合放入缓存

查询频率比较高，修改频率比较低。

安全系数低的数据

(3)使用redis作为缓存

a.把缓存的配置类加入

package com.zpb.config;

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.cache.CacheManager;
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.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.time.Duration;

@Configuration
public class RedisConfig {
    @Bean
    public CacheManager cacheManager(RedisConnectionFactory factory) {
        RedisSerializer<String> redisSerializer = new StringRedisSerializer();
        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);
        // 配置序列化（解决乱码的问题）,过期时间600秒
        RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig()
                .entryTtl(Duration.ofSeconds(600)) //缓存过期10分钟 ---- 业务需求。
                .serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(redisSerializer))//设置key的序列化方式
                .serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(jackson2JsonRedisSerializer)) //设置value的序列化
                .disableCachingNullValues();
        RedisCacheManager cacheManager = RedisCacheManager.builder(factory)
                .cacheDefaults(config)
                .build();
        return cacheManager;
    }
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        RedisSerializer<String> redisSerializer = new StringRedisSerializer();
        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);
        template.setConnectionFactory(factory);
        //key序列化方式
        template.setKeySerializer(redisSerializer);
        //value序列化
        template.setValueSerializer(jackson2JsonRedisSerializer);
        //value hashmap序列化  filed value
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
        template.setHashKeySerializer(redisSerializer);
        return template;
    }
}

b.使用开启缓存注解

 c.使用注解

 

package com.zpb.service;

import com.zpb.dao.StudentMapper;
import com.zpb.entity.Student;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.stereotype.Service;

@Service
public class StudentService {
    @Autowired
    private StudentMapper studentMapper;

    //业务代码
    //使用查询注解:cacheNames表示缓存的名称 key:唯一标志---student::key
    //先从缓存中查看key为（cacheNames::key）是否存在，如果存在则不会执行方法体，如果不存在则执行方法体并把方法的返回值存入缓存中
    @Cacheable(cacheNames = {"student"},key = "#id")
    public Student findById(Integer id){
        Student student = studentMapper.selectById(id);
        return student;
    }
}

1.13.2 分布式锁

使用synchronized或者lock锁解决高并发下，同一个库存被使用了n次，数据库中库存为负数的线程安全问题。

package com.zpb.service.impl;

import com.zpb.distrinctlock.dao.ProductStockDao;
import com.zpb.distrinctlock.service.ProductStockService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class ProductStockServiceImpl2 implements ProductStockService {
    @Autowired
    private ProductStockDao productStockDao;

    @Override
    public  String decreaseStock(Integer productId) {
              synchronized (this) {
                  //查看该商品的库存数量
                  Integer stock = productStockDao.findStockByProductId(productId);
                  if (stock > 0) {
                      //修改库存每次-1
                      productStockDao.updateStockByProductId(productId);
                      System.out.println("扣减成功！剩余库存数:" + (stock - 1));
                      return "success";
                  } else {
                      System.out.println("扣减失败！库存不足！");
                      return "fail";
                  }
              }

    }
}

 如果我们搭建了项目集群，使用synchronized 或者lock锁 ，那么该锁无效。可以使用redis加锁

package com.zpb.distrinctlock.service.impl;

import com.zpb.distrinctlock.dao.ProductStockDao;
import com.zpb.distrinctlock.service.ProductStockService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;

@Service
public class ProductStockServiceImpl2 implements ProductStockService {
    @Autowired
    private ProductStockDao productStockDao;
    @Autowired
    private StringRedisTemplate redisTemplate;

    @Override
    public  String decreaseStock(Integer productId) {
        ValueOperations<String, String> forValue = redisTemplate.opsForValue();
        Boolean flag = forValue.setIfAbsent("aaa::" + productId, "******",5, TimeUnit.MINUTES);
        if (flag) {
            try {
                //查看该商品的库存数量
                Integer stock = productStockDao.findStockByProductId(productId);
                if (stock > 0) {
                    //修改库存每次-1
                    productStockDao.updateStockByProductId(productId);
                    System.out.println("扣减成功！剩余库存数:" + (stock - 1));
                    return "success";
                } else {
                    System.out.println("扣减失败！库存不足！");
                    return "fail";
                }
            }finally {
                redisTemplate.delete("aaa::"+productId);
            }
        }
        return "服务器正忙，请稍后再试......";
    }
}

但是Redis分布式锁解决不了超时的问题，分布式锁有一个超时时间，程序的执行如果超过了锁的超时时间就会出现问题。

解决方案：使用redission解决超时问题

redission原理

 为持有锁的线程开启一个守护线程，守护线程会每隔10秒检查当前线程是否还持有锁，如果持有则延迟生存时间。

（1）加入依赖

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

（2）使用

package com.zpb.distrinctlock;

import org.redisson.Redisson;
import org.redisson.config.Config;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

@SpringBootApplication
public class DistrinctLockApplication {

    public static void main(String[] args) {
        SpringApplication.run(DistrinctLockApplication.class, args);
    }

    //获取redisson对象并交于spring容器管理
    @Bean
    public Redisson redisson(){
        Config config =new Config();
        config.useSingleServer().
                setAddress("redis://192.168.23.185:7001").
                //redis默认有16个数据库
                setDatabase(0);
        return (Redisson) Redisson.create(config);
    }

}

    @Override
    public String decreaseStock(Integer productId) {
        //获取锁对象
        RLock lock = redisson.getLock("aaa::" + productId);
        try {
            lock.lock(30, TimeUnit.SECONDS);
            //查看该商品的库存数量
            Integer stock = productStockDao.findStockByProductId(productId);
            if (stock > 0) {
                //修改库存每次-1
                productStockDao.updateStockByProductId(productId);
                System.out.println("扣减成功！剩余库存数:" + (stock - 1));
                return "success";
            } else {
                System.out.println("扣减失败！库存不足！");
                return "fail";
            }
        } finally {
            lock.unlock();
        }


    }

1.14 什么是缓存穿透？怎么解决？

1. 数据库中没有该记录，缓存中也没有该记录，这时由人恶意大量访问这样的数据。这样就会导致该请求绕过缓存，直接访问数据，从而造成数据库压力过大。

2.解决办法:
   [1]在controller加数据校验。
   [2]我们可以在redis中存入一个空对象，而且要设置过期时间不能太长。超过5分钟
   [3]我们使用布隆过滤器。底层:有一个bitmap数组，里面存储了该表的所有id.

//伪代码
String get(String key) { //布隆过滤器钟存储的是数据库表钟对应的id
    String value = redis.get(key);  //先从缓存获取。  
    if (value  == null) { //缓存没有命中
        if(!bloomfilter.mightContain(key)){//查看布隆过滤器钟是否存在
            return null; 
        }else{
            value = db.get(key); //查询数据库
            redis.set(key, value); 
        }    
    }
    return value；
}

1.15 什么是缓存雪崩？如何解决？

缓存雪崩是指缓存中数据大批量到过期时间，而查询数据量巨大，引起数据库压力过大甚至down机。和缓存击穿不同的是， 缓存击穿指并发查同一条数据，缓存雪崩是不同数据都过期了，很多数据都查不到从而查数据库。
1.什么下会发生缓存雪崩:
  [1]项目刚上线，缓存中没有任何数据
  [2]缓存出现大量过期。
  [3]redis宕机
 
2.解决办法:
   1.上线前预先把一些热点数据放入缓存。
   2.设置过期时间为散列值
   3.搭建redis集群

1.16 什么是缓存击穿？如何解决？

缓存击穿是指缓存中没有但数据库中有的数据（一般是缓存时间到期），这时由于并发用户特别多，同时读缓存没读到数据，又同时去数据库去取数据，引起数据库压力瞬间增大，造成过大压力。

缓存击穿解决方案:
1.设置永久不过期。【这种只适合内存】
2.使用互斥锁(mutex key)业界比较常用的做法。

1.17 Redis 淘汰策略有哪些？