redis总结与在java中的简单使用

2020-06-30 技术记录评论

本文简单记录redis的相关总结与简单使用

一、使用篇

1.简单认识redis

关于NoSQL

NoSQL(NoSQL = Not Only SQL )，意即“不仅仅是SQL”，泛指非关系型的数据库。

随着互联网web2.0网站的兴起，传统的关系数据库在应付web2.0网站，特别是超大规模和高并发的SNS类型的web2.0纯动态网站已经显得力不从心，暴露了很多难以克服的问题，而非关系型的数据库则由于其本身的特点得到了非常迅速的发展。NoSQL数据库的产生就是为了解决大规模数据集合多重数据种类带来的挑战，尤其是大数据应用难题。

而redis是一个以key-value形式存储数据的非关系型数据库。

redis应用场景

会话缓存。即session cache。
全页缓存。
队列。
排行榜、计时器
发布、订阅

2.redis的安装与启动

安装步骤不再详细写出，参考菜鸟教程-redis安装

windows下：

启动：进入redis目录，打开cmd，输入命令“redis-server redis.windows.conf”以开启redis服务。注意不能关闭cmd窗口。（当然可以把服务注册到windows系统服务以在后台运行，这里先不给出）

连接到redis数据库：再开一个cmd窗口，输入命令”redis-cli -h localhost -p 6379” (默认端口为6379，没有密码)

尝试插入一个key，输入“set aaa 111”，回车：

一般会下载redis界面化客户端方便查看数据，官网下载。

3.springboot集成redis与简单使用

3-1.数据准备

create table user(
  id int primary key auto_increment,
  name varchar(20),
  age int
)
insert into user(name,age)values("xiaoming",20)
insert into user(name,age)values("daming",50)

3-2.创建springboot项目，添加依赖

<!-- mysql 数据库驱动 -->
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>8.0.11</version>
</dependency>
<!-- jpa 数据库ORM框架 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>
<!-- spring boot redis 缓存引入 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<!-- lettuce pool 缓存连接池 -->
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-pool2</artifactId>
</dependency>
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-pool2</artifactId>
</dependency>

3-3. application.properties 配置文件加入redis配置

Spring.dataSource.driver-class-name=com.mysql.cj.jdbc.Driver
Spring.dataSource.url=jdbc:mysql://localhost:3306/redis-demo?serverTimezone=UTC&characterEncoding=utf-8&useSSL=false
Spring.dataSource.username=root
Spring.dataSource.password=123456

# Redis数据库索引（默认为0）
spring.redis.database=0  
# Redis服务器地址
spring.redis.host=localhost
# Redis服务器连接端口
spring.redis.port=6379  
# Redis服务器连接密码（默认为空）
spring.redis.password=
# 连接池最大连接数（使用负值表示没有限制） 默认 8
spring.redis.lettuce.pool.max-active=8
# 连接池最大阻塞等待时间（使用负值表示没有限制） 默认 -1
spring.redis.lettuce.pool.max-wait=-1
# 连接池中的最大空闲连接 默认 8
spring.redis.lettuce.pool.max-idle=8
# 连接池中的最小空闲连接 默认 0
spring.redis.lettuce.pool.min-idle=0

3-4. 新建config包，创建RedisConfig配置类

@Configuration
public class RedisConfig {
    @Bean(name = {"redisTemplate", "stringRedisTemplate"})
    public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory redisConnectionFactory){
        StringRedisTemplate stringRedisTemplate = new StringRedisTemplate();
        stringRedisTemplate.setConnectionFactory(redisConnectionFactory);
        return stringRedisTemplate;
    }
}

RedisTemplate默认只能支持RedisTemplate<String,String>形式的，也就是key-value只能是字符串，不能是其他对象。所以我们自己定义一个RedisTemplate对象，返回一个自己想要的RedisTemplate对象，自己定义序列化方式。

3-5.新建entity包，创建User实体类

@Entity
public class User implements Serializable {
    private static final long serialVersionUID = -3055518357571903443L;
    @Id
    private Integer id;
    private String name;
    private Integer age;

    ...getter and setter省略
}

3-6.新建dao包，创建UserDao类

1
2
3

@Repository("user")
public interface UserDao extends JpaRepository<User,Integer> {
}

3-7.新建service包，创建UserService

@Service
public class UserServiceImpl implements UserService {

    @Autowired
    UserDao userDao;

    @Override
    public List<User> list() {
        List<User> all = userDao.findAll();
        return all;
    }
}

3-8.新建util包，创建RedisUtil工具类

@Component
public class RedisUtil {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    public void setRedisTemplate(RedisTemplate<String, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }
    //=============================common============================  
    /**
     * 指定缓存失效时间 
     * @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);
    }

    /**
     * 判断key是否存在 
     * @param key 键 
     * @return true 存在 false不存在 
     */
    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=================================  

    //============================set=============================  
    
    //===============================list=================================  

   
}

Map、Set、List三种数据类型的方法省略。

3-9.在测试类里测试功能

@SpringBootTest
class DemoApplicationTests {

    @Autowired
    UserService userService;

    @Autowired
    RedisUtil redisUtil;

    @Test
    void testRedis(){
        List<User> userList = userService.list();
        System.out.println(userList);

        if (redisUtil.hasKey(userList.get(0).getName())){
            System.out.println("已存在");
        }else {
            System.out.println("不存在");
            System.out.println(redisUtil.set(userList.get(0).getName(), userList.get(0).getAge().toString()) ? "成功" : "失败");
        }
    }
}

3-10.关于注解方式

以上就是java使用redis作为缓存数据库的用法。当然，也可以使用注解的方式来实现。

https://segmentfault.com/a/1190000017057950 可参考此连接进行相关配置。

4.redis用作分布式锁

4-1.关于锁

线程锁：主要用来给方法、代码块加锁。当某个方法或者代码块使用锁时，那么在同一时刻至多仅有有一个线程在执行该段代码。当有多个线程访问同一对象的加锁方法/代码块时，同一时间只有一个线程在执行，其余线程必须要等待当前线程执行完之后才能执行该代码段。例如在java中的synchronized和lock。
进程锁：为了控制同一操作系统中多个进程访问一个共享资源，只是因为程序的独立性，各个进程是无法控制其他进程对资源的访问的，但是可以使用本地系统的信号量控制。
分布式锁：当多个进程不在同一个系统之中时，使用分布式锁控制多个进程对资源的访问。

4-2.分布式锁的实现方案

分布式锁主要是三种方式实现：

基于关系型数据库。用一个字段记录是否当前数据已被取走。又或者监控时间戳，若返回保存时发现时间戳跟取走的时候不一致，证明期间有其他进程访问，则保存失败。
基于缓存。
基于zookeeper。

4-3.redis用作分布式锁的相关命令

setnx。setnx当且仅当 key 不存在。若给定的 key 已经存在，则 setnx不做任何动作。setnx 是『set if not exists』(如果不存在，则 set)的简写,setnx 具有原子性。
incr。将 key 中储存的数字值增一。如果 key 不存在，那么 key 的值会先被初始化为 0 ，然后再执行 INCR 操作。如果值包含错误的类型，或字符串类型的值不能表示为数字，那么返回一个错误。
getset。getset先 get 旧值，后set 新值，并返回 key 的旧值(old value），具有原子性。当 key 存在但不是字符串类型时，返回一个错误;当key 不存在的时候，返回nil ，在Java里就是 null。
expire 设置 key 的有效期
del 删除 key

官方推荐用SETNX实现分布式锁
利用SETNX非常简单地实现分布式锁。例如：某客户端要获得一个名字foo的锁，客户端使用下面的命令进行获取：
SETNX lock.foo <current Unix time + lock timeout + 1>

如返回1，则该客户端获得锁，把lock.foo的键值设置为时间值表示该键已被锁定，该客户端最后可以通过DEL lock.foo来释放该锁。

如返回0，表明该锁已被其他客户端取得，这时我们可以先返回或进行重试等对方完成或等待锁超时。

4-4.分布式锁的简单例子

4-4-1.redisUtil类增加setnx对应封装方法：

/**
 * setnx
 * @param key 键
 * @param value 值
 * @return true成功 false失败
 */
public boolean setnx(String key,Object value) {
    try {
        return redisTemplate.opsForValue().setIfAbsent(key, value);
    } catch (Exception e) {
        e.printStackTrace();
        return false;
    }

}

/**
 * setnx
 * @param key 键
 * @param value 值
 * @param time 时间(秒) time要大于0 如果time小于等于0 将设置无限期
 * @return true成功 false 失败
 */
public boolean setnx(String key,Object value,long time){
    try {
        if(time>0){
            return redisTemplate.opsForValue().setIfAbsent(key, value, time, TimeUnit.SECONDS);
        }else{
            return setnx(key, value);
        }
    } catch (Exception e) {
        e.printStackTrace();
        return false;
    }
}

4-4-2.创建一个分布式锁的应用测试方法

@Test
   void testRedis2(){
       String lockKey = "lockKey";
       boolean flag = redisUtil.setnx(lockKey, 1 + "", 10);
       if (flag) {
           try {
               int stock = Integer.parseInt((String) redisUtil.get("stock"));
               if (stock > 0) {
                   // 库存-1
                   --stock;
                   redisUtil.set("stock", stock + "");
                   System.out.println("交易成功，库存为：" + stock);
               } else {
                   System.out.println("交易失败，库存为：" + stock);
               }
           } finally {
               redisUtil.del("lockKey");
           }
       }
   }

简要说明：

要点一、必须有finally把“锁”（即key）删掉，避免在运行此方法的过程中遇到异常，导致锁一直在，造成后续无法交易的问题。

要点二、必须给锁设置过期时间，避免比如服务器崩溃时（finally也没执行）的情况下锁一致在。

4-4-3.设置stock初始值并进行压测

可以把测试方法写成正式的接口，用压力测试工具jmeter来进行测试。这里不详细给出测试过程。

测试结果我们发现，没有出现“超卖”的现象。但有问题

仅有1个线程成功。可能请求量不够。
其他问题，比如我们设置了10s过期时间，但是由于某种原因，10s内没执行完毕，导致线程2（另外一个请求）获得了锁，进入了业务代码段。这里的解决方法可以是，给线程1新增一个子线程，作为监控的作用，比如每隔5秒，查看业务代码段是否已经真正完成了，若未完成，延长超时时间。

但是经验告诉我们，往往我们想到的问题已经有了解决方案，以下介绍一个强大的框架，它不仅解决了这些问题，还有很多丰富的功能。

4-5.使用开源 redisson 实现分布式锁

官网介绍：

Redisson - Redis Java client with features of In-Memory Data Grid

现在尝试使用它：

4-5-1.引入依赖

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

4-5-2.配置文件

1	redisson.address=redis://127.0.0.1:6379

4-5-3.配置类

redisson基本配置类，也是大家熟悉的套路，因为redisson支持多种模式下的配置，比如单机、集群、哨兵模式等，都可以根据实际业务需要进行配置，这里为演示方便使用单机配置。

@Configuration
public class RedissonManager {
	
	@Value("${redisson.address}")
	private String addressUrl;
	
	@Bean
	public RedissonClient getRedisson() throws Exception{
		RedissonClient redisson = null;
		Config config = new Config();
		config.useSingleServer()
			  .setAddress(addressUrl);
		redisson = Redisson.create(config);
		
		System.out.println(redisson.getConfig().toJSON().toString());
		return redisson;
	}
}

4-5-4.新增分布式锁测试方法

@Test
void threadTest(){
    int threadCount = 100;
    final CountDownLatch latch = new CountDownLatch(threadCount);
    for(int i=0; i< threadCount; i++){
        new Thread(new Runnable() {
            @Override
            public void run() {
                testRedis3();
                latch.countDown();
            }
        }).start();
    }
    try {
        latch.await();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println(threadCount+"个线程已经执行完毕");
}

@Test
void testRedis3(){
    String lockKey = "lockKey";
    RLock lock = redisson.getLock(lockKey);
    try {
        lock.lock();
        int stock = Integer.parseInt((String) redisUtil.get("stock"));
        if (stock > 0) {
            // 库存-1
            --stock;
            redisUtil.set("stock", stock + "");
            System.out.println("线程id=" + Thread.currentThread().getId() + "交易成功，库存为：" + stock);
        } else {
            System.out.println("线程id=" + Thread.currentThread().getId() + "交易失败，库存为：" + stock);
        }
    }finally {
        lock.unlock();
    }
}

我们用CountDownLatch类来创建多个线程并发执行，模拟网络请求，测试结果：

共100个请求，只有前10个交易成功，符合预期效果。

至此，完成使用redisson实现分布式锁。

本次示例demo源码地址 https://github.com/lizxing/redis-demo

二、总结篇

1.redis的数据类型

String。基本操作：SET Key Value
hash。存放结构化对象。基本操作：a、单个单个存：HSET Key Field Value。 b、多个存：HMSET Key Field1 Value1 Field2 Value2
list。顺序队列之类的。基本操作：LPUSH Key Value；LPOP Key
set。去重。由于很多服务是集群，如果直接使用java的set还得建一个全局服务去把数据合并起来比较。基本操作：SADD Key value1；SADD Key value2
sorted set。有序set。基本操作：ZADD key 1 value1；ZADD key 2 value2

2.redis的持久化机制

redis通过持久化机制把内存中的数据同步到硬盘文件，当redis重启后把硬盘文件重新加载到内存，以达到恢复数据的目的。参考文章Redis持久化 - RDB和AOF

有两种方式：

RDB。RDB持久化方式能够在指定的时间间隔能对你的数据进行快照存储。在默认情况下， Redis 将数据库快照保存在名字为 dump.rdb的二进制文件中。
在 Redis 运行时， RDB 程序将当前内存中的数据库快照保存到磁盘文件中，在 Redis 重启动时， RDB 程序可以通过载入 RDB 文件来还原数据库的状态。
- 优势：RDB是一个非常紧凑的文件，它保存了某个时间点得数据集，非常适用于数据集的备份；与AOF相比，在恢复大的数据集的时候，RDB方式会更快一些。
- 劣势：耗时、耗性能。RDB 需要经常fork子进程来保存数据集到硬盘上；在子进程写数据的过程中不能对主进程的当前操作保存，若此时意外宕机，则丢失这部分数据。
AOF。每当 Redis 执行一个改变数据集的命令时（比如 SET），这个命令就会被追加到 AOF 文件的末尾，类似于日志文件。这样的话，当 Redis 重新启时，程序就可以通过重新执行 AOF 文件中的命令来达到重建数据集的目的。AOF文件也有重写功能，它把命令集进行优化，生成一个新的 AOF 文件，这个文件包含重建当前数据集所需的最少命令。redis2.4会可以通过配置自动触发重写。AOF有三种策略：always，每次有新命令追加到 AOF 文件时就执行一次 fsync；everysec，每秒 fsync 一次；no，从不 fsync。显然always消耗性能最大。
- 优势：使用默认的每秒fsync策略，Redis的性能依然很好(fsync是由后台线程进行处理的，主线程会尽力处理客户端请求)，一旦出现故障，你最多丢失1秒的数据；AOF 文件有序地保存了对数据库执行的所有写入操作，这些写入操作以 Redis 协议的格式保存，因此 AOF 文件的内容非常容易被人读懂。
- 劣势：对于相同的数据集来说，AOF 文件的体积通常要大于 RDB 文件的体积；根据所使用的 fsync 策略，AOF 的速度可能会慢于 RDB 。

3.redis的过期策略及内存淘汰机制

redis使用定期删除+惰性删除策略。

定期删除：指的是redis默认是每隔100ms就随机抽取一些设置了过期时间的key，检查其是否过期，如果过期就删除

惰性删除：在你获取某个key的时候，redis会检查一下，这个key如果设置了过期时间那么是否过期了，如果过期了此时就会删除，不会给你返回任何东西

但是，也会存在一些key从来没被检查过，导致越堆越多，这时候就要用到内存淘汰机制。

noeviction：当内存使用达到阈值的时候，所有引起申请内存的命令会报错。
allkeys-lru：在主键空间中，优先移除最近未使用的key。
volatile-lru：在设置了过期时间的键空间中，优先移除最近未使用的key。
allkeys-random：在主键空间中，随机移除某个key。
volatile-random：在设置了过期时间的键空间中，随机移除某个key。
volatile-ttl：在设置了过期时间的键空间中，具有更早过期时间的key优先移除。

可以在 redis.conf中配置：

1	# maxmemory-policy noeviction

4.redis为什么快

完全基于内存，绝大部分请求是纯粹的内存操作，非常快速。数据存在内存中，类似于HashMap，HashMap的优势就是查找和操作的时间复杂度都是O(1)；
数据结构简单，对数据操作也简单，Redis中的数据结构是专门进行设计的；
采用单线程，避免了不必要的上下文切换和竞争条件，也不存在多进程或者多线程导致的切换而消耗 CPU，不用去考虑各种锁的问题，不存在加锁释放锁操作，没有因为可能出现死锁而导致的性能消耗；
使用多路I/O复用模型，非阻塞IO；这里“多路”指的是多个网络连接，“复用”指的是复用同一个线程。
使用底层模型不同，它们之间底层实现方式以及与客户端之间通信的应用协议不一样，Redis直接自己构建了VM 机制，因为一般的系统调用系统函数的话，会浪费一定的时间去移动和请求；

5.redis如何保证原子性

redis是单线程。

6.redis如何实现异步和非阻塞

对于Redis的网络请求，Redis会有一个EventLoop，里面有两个数组events,fired。events存放被注册的事件，fired用于存放EventLoop从多路复用器（epoll）中读取到的，将要执行的事件。

异步和非阻塞就反映在这里，注册到多路复用器（epoll）后去做其他事，之后通过主动轮询多路复用器，来逐个取出将要执行的事件，放入fired，逐个执行，这个过程是单线程的，因此不会出现并发问题。

7.redis事务

事务相关命令：

MULTI。标记一个事务块的开始。
EXEC。执行所有事务块内的命令。
DISCARD。取消事务，放弃执行事务块内的所有命令。
WATCH key [key …]。监视一个(或多个) key ，如果在事务执行之前这个(或这些) key 被其他命令所改动，那么事务将被打断。
UNWATCH。取消 WATCH 命令对所有 key 的监视。

From redis docs on transactions:

It’s important to note that even when a command fails, all the other commands in the queue are processed – Redis will not stop the processing of commands.

单个 Redis 命令的执行是原子性的，但 Redis 没有在事务上增加任何维持原子性的机制，所以 Redis 事务的执行并不是原子性的。

事务可以理解为一个打包的批量执行脚本，但批量指令并非原子化的操作，中间某条指令的失败不会导致前面已做指令的回滚，也不会造成后续的指令不做。

例如：

redis 127.0.0.1:7000> multi
OK
redis 127.0.0.1:7000> set a aaa
QUEUED
redis 127.0.0.1:7000> set b bbb
QUEUED
redis 127.0.0.1:7000> set c ccc
QUEUED
redis 127.0.0.1:7000> exec
1) OK
2) OK
3) OK

如果在 set b bbb 处失败，set a 已成功不会回滚，set c 还会继续执行。

小结：（参考Redis的事务讲解）

事务阶段
- 开启：以MULTI 开启一个事务
- 入队：将多个命令入队到事务中，接到这些命令不会立即执行，而是放到等待执行的事务队列里面
- 执行：由EXEC命令触发事务
事务特性
- 单独的隔离操作：事务中的所有命令都会序列化、按顺序地执行。事务在执行的过程中，不会被其他客户端发送来的命令请求所打断。
- 没有隔离级别的概念：队列中的命令没有提交之前都不会实际的被执行，因为事务提交前任何指令都不会被实际执行，也就不存在”事务内的查询要看到事务里的更新，在事务外查询不能看到”这个让人万分头痛的问题
- 不保证原子性：redis同一个事务中如果有一条命令执行失败，其后的命令仍然会被执行，没有回滚

8.关于缓存存在的问题

缓存雪崩：同一时间原缓存失效，大量请求直接访问数据库，导致宕机。解决方法两种：
- a、缓存设置过期时间分散。
- b、使用互斥锁(redis的话用setnx)，当缓存失效后，任一线程重新构建key，其他线程等待构建完成再重新执行。
缓存穿透：要查询的数据在数据库没有，那么在缓存中也肯定没有。这时每次请求都会访问到缓存和数据库，相当于有两次无用的访问。解决方法两种：
- a、第一次查不到的数据，直接以空结果为value存进缓存中，这样后面的请求就不会访问到数据库，但过期时间不宜过长。
- b、利用布隆过滤器（特点：存在的可能不存在，不存在的一定不存在），将所有可能存在的数据哈希到一个bitmap中，这样一个一定不存在的数据会被直接拦截掉。
缓存预热：系统上线后将缓存数据加载到缓存系统，避免直接请求数据库。
缓存更新：更新最新的缓存数据。
缓存降级：保证核心服务可用。

本文链接： https://lizxing.github.io/2020/06/30/20063001/

版权声明： 本博客所有文章除特别声明外，均采用 CC BY 4.0 CN协议许可协议。转载请注明出处！

lizxing投资砖家

个人简介。