一.Redis 入门

1.初识 Redis

Redis 是一种键值型的NoSql数据库。

键值型：是指 Redis 中存储的数据都是以 key、value 对的形式存储，而 value 的形式多种多样，可以是字符串、数值、甚至 json

1.1 认识 NoSQL

1.1.1 什么是 NoSQL

  • NoSQL 最常见的解释是”non-relational“， 很多人也说它是”Not Only SQL“
  • NoSQL 仅仅是一个概念，泛指非关系型的数据库
  • 区别于关系数据库，它们不保证关系数据的 ACID 特性
  • NoSQL 是一项全新的数据库革命性运动，提倡运用非关系型的数据存储，相对于铺天盖地的关系型数据库运用，这一概念无疑是一种全新的思维的注入
  • 常见的 NoSQL 数据库有：Redis、MemCache、MongoDB等

1.1.2 NoSQL 与 SQL 的差异

| | SQL | NoSQL |
| ——– | ————— | ————————————— | ————————— |
| 数据结构 | 结构化 | 非结构化 |
| 数据关联 | 关联的 | 无关联的 |
| 查询方式 | SQL 查询 | 非 SQL |
| 事务特性 | ACID | BASE |
| 存储方式 | 磁盘 | 内存 |
| 扩展性 | 垂直 | 水平 |
| 使用场景 | 1）数据结构固定 | 1）数据结构不固定 |
| | | 2）相关业务对数据安全性、一致性要求较高 | 2）对一致性、安全性要求不高 |
| | | 3）对性能要求 |

2.Redis 常见命令

我们可以通过 Redis 的中文文档：http://www.redis.cn/commands.html，来学习各种命令。

也可以通过菜鸟教程官网来学习：https://www.runoob.com/redis/redis-keys.html

2.1 Redis 数据类型

Redis 是一种 key-value 数据库，一般 key 都是 String，value 的类型五花八门。

2.2 Redis 基本命令

通用指令是部分数据类型的，都可以使用的指令

指令 描述
KEYS 查看符合模板的所有 key
DEL 删除一个指定的 key
EXISTS 判断 key 是否存在
EXPIRE 给一个 key 设置有效期，有效期到期时该 key 会被自动删除
TTL 查看一个 KEY 的剩余有效期

通过 help [command] 可以查看一个命令的具体用法

2.3 String 类型

字符串类型，Redis 中最简单的存储类型。

其 value 是字符串，但是根据字符串的格式不同，可以分为三类：

  • string：普通字符串
  • int：整数类型，可以做自增、自减操作。
  • float：浮点类型，可以做自增、自减操作。

不管是哪种格式，底层都是字节数组形式存储，只不过是编码方式不同。字符串类型的最大空间不能超过 512m.

2.3.1 String 类型常用命令

命令 描述
SET 添加或者修改已经存在的一个 String 类型的键值对
GET 根据 key 获取 String 类型的 value
MSET 批量添加多个 String 类型的键值对
MGET 根据多个 key 获取多个 String 类型的 value
INCR 让一个整型的 key 自增 1
INCRBY 让一个整型的 key 自增并指定步长，例如：incrby num 2 让 num 值自增 2
INCRBYFLOAT 让一个浮点类型的数字自增并指定步长
SETNX 添加一个 String 类型的键值对，前提是这个 key 不存在，否则不执行
SETEX 添加一个 String 类型的键值对，并且指定有效期

2.4 Hash 类型

Hash 类型，也叫散列，其 value 是一个无序字典，类似于 Java 中的**HashMap**结构。

  • Hash 结构可以将对象中的每个字段独立存储，可以针对单个字段做 CRUD
  • Hash 的常见命令有：
    命令 描述
    HSET key field value 添加或者修改 hash 类型 key 的 field 的值
    HGET key field 获取一个 hash 类型 key 的 field 的值
    HMSET hmset 和 hset 效果相同 ，4.0 之后 hmset 可以弃用了
    HMGET 批量获取多个 hash 类型 key 的 field 的值
    HGETALL 获取一个 hash 类型的 key 中的所有的 field 和 value
    HKEYS 获取一个 hash 类型的 key 中的所有的 field
    HVALS 获取一个 hash 类型的 key 中的所有的 value
    HINCRBY 让一个 hash 类型 key 的字段值自增并指定步长
    HSETNX 添加一个 hash 类型的 key 的 field 值，前提是这个 field 不存在，否则不执行

2.5 List 类型

Redis 中的 List 类型与 Java 中的 LinkedList 类似，可以看做是一个双向链表结构。既可以支持正向检索和也可以支持反向检索。

特征也与**LinkedList**类似：

  • 有序
  • 元素可以重复
  • 插入和删除快
  • 查询速度一般

常用来存储一个有序数据，例如：朋友圈点赞列表，评论列表等.

List 的常见命令有

命令 描述
LPUSH key  element … 向列表左侧插入一个或多个元素
LPOP key 移除并返回列表左侧的第一个元素，没有则返回 nil
RPUSH key  element … 向列表右侧插入一个或多个元素
RPOP key 移除并返回列表右侧的第一个元素
LRANGE key star end 返回一段角标范围内的所有元素
BLPOP 和 BRPOP 与 LPOP 和 RPOP 类似，只不过在没有元素时等待指定时间，而不是直接返回 nil

2.6 Set 类型

Redis 的 Set 结构与 Java 中的 HashSet 类似，可以看做是一个 value 为 null 的 HashMap。因为也是一个 hash 表，因此具备与 HashSet 类似的特征

  • 无序
  • 元素不可重复
  • 查找快
  • 支持交集、并集、差集等功能

Set 的常见命令有

命令 描述
SADD key member … 向 set 中添加一个或多个元素
SREM key member … 移除 set 中的指定元素
SCARD key 返回 set 中元素的个数
SISMEMBER key member 判断一个元素是否存在于 set 中
SMEMBERS 获取 set 中的所有元素
SINTER key1 key2 … 求 key1 与 key2 的交集
SDIFF key1 key2 … 求 key1 与 key2 的差集
SUNION key1 key2 .. 求 key1 和 key2 的并集

2.7 SortedSet(zset)类型

Redis 的 SortedSet 是一个可排序的 set 集合，与 Java 中的 TreeSet 有些类似，但底层数据结构却差别很大。SortedSet 中的每一个元素都带有一个 score 属性，可以基于 score 属性对元素排序，底层的实现是一个跳表（SkipList）加 hash 表。

SortedSet 具备下列特性：

  • 可排序
  • 元素不重复
  • 查询速度快

因为 SortedSet 的可排序特性，经常被用来实现排行榜这样的功能。

SortedSet 的常见命令有

命令 描述
ZADD key score member 添加一个或多个元素到 sorted set ，如果已经存在则更新其 score 值
ZREM key member 删除 sorted set 中的一个指定元素
ZSCORE key member 获取 sorted set 中的指定元素的 score 值
ZRANK key member 获取 sorted set 中的指定元素的排名
ZCARD key 获取 sorted set 中的元素个数
ZCOUNT key min max 统计 score 值在给定范围内的所有元素的个数
ZINCRBY key increment member 让 sorted set 中的指定元素自增，步长为指定的 increment 值
ZRANGE key min max 按照 score 排序后，获取指定排名范围内的元素
ZRANGEBYSCORE key min max 按照 score 排序后，获取指定 score 范围内的元素
ZDIFF、ZINTER、ZUNION 求差集、交集、并集

注意：所有的排名默认都是升序，如果要降序则在命令的 Z 后面添加**REV**即可

跳跃表

从第 2 层开始，1 节点比 51 节点小，向后比较。

21 节点比 51 节点小，继续向后比较，后面就是 NULL 了，所以从 21 节点向下到第 1 层

在第 1 层，41 节点比 51 节点小，继续向后，61 节点比 51 节点大，所以从 41 向下

在第 0 层，51 节点为要查找的节点，节点被找到，共查找 4 次。

从此可以看出跳跃表比有序链表效率要高

3.Redis 客户端

3.Java 客户端

3.1 Jedis 快速入门

该部分参考：https://www.oz6.cn/articles/58
Jedis 的官网地址： https://github.com/redis/jedis，我们先来个快速入门：

  • 新建一个 Maven 工程并引入以下依赖
1
2
3
4
5
6
7
8
9
10
11
12
13
14
<!--引入Jedis依赖-->
<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>4.2.0</version>
</dependency>

<!--引入单元测试依赖-->
<dependency>
    <groupId>org.junit.jupiter</groupId>
    <artifactId>junit-jupiter</artifactId>
    <version>5.8.2</version>
    <scope>test</scope>
</dependency>
  • 编写测试类并与 Redis 建立连接
1
2
3
4
5
6
7
8
9
10
11
private Jedis jedis;

@BeforeEach //被该注解修饰的方法每次执行其他方法前自动执行
void setUp(){
    // 1. 获取连接
    jedis = new Jedis("192.168.230.88",6379);
    // 2. 设置密码
    jedis.auth("132537");
    // 3. 选择库（默认是下标为0的库）
    jedis.select(0);
}
  • 编写一个操作数据的方法（这里以操作 String 类型为例）
1
2
3
4
5
6
7
8
9
10
@Test
public void testString(){
    // 1.往redis中存放一条String类型的数据并获取返回结果
    String result = jedis.set("url", "www.oz6.cn");
    System.out.println("result = " + result);

    // 2.从redis中获取一条数据
    String url = jedis.get("url");
    System.out.println("url = " + url);
}
  • 最后不要忘记编写一个释放资源的方法
1
2
3
4
5
6
7
@AfterEach //被该注解修饰的方法会在每次执行其他方法后执行
void tearDown(){
    // 1.释放资源
    if (jedis != null){
        jedis.close();
    }
}
  • 执行**testString()**方法后测试结果如图所示

3.2 Jedis 连接池

Jedis 本身是线程不安全的，并且频繁的创建和销毁连接会有性能损耗，因此我们推荐大家使用 Jedis 连接池代替 Jedis 的直连方式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public class JedisConnectionFactory {
    private static final JedisPool jedisPool;

    static {
        //配置连接池
        JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
        jedisPoolConfig.setMaxTotal(8);
        jedisPoolConfig.setMaxIdle(8);
        jedisPoolConfig.setMinIdle(0);
        jedisPoolConfig.setMaxWaitMillis(200);
        //创建连接池对象
        jedisPool = new JedisPool(jedisPoolConfig,"192.168.230.88",6379,1000,"132537");
    }

    public static Jedis getJedis(){
       return jedisPool.getResource();
    }
}

3.3 SpringDataRedis 介绍

SpringData 是 Spring 中数据操作的模块，包含对各种数据库的集成，其中对 Redis 的集成模块就叫做**SpringDataRedis**

官网地址：https://spring.io/projects/spring-data-redis

  • 提供了对不同 Redis 客户端的整合（Lettuce和Jedis）
  • 提供了RedisTemplate统一 API 来操作 Redis
  • 支持 Redis 的发布订阅模型
  • 支持 Redis 哨兵和 Redis 集群
  • 支持基于 Lettuce 的响应式编程
  • 支持基于 JDK、JSON、字符串、Spring 对象的数据序列化及反序列化
  • 支持基于 Redis 的 JDKCollection 实现

SpringDataRedis 中提供了 RedisTemplate 工具类，其中封装了各种对 Redis 的操作。并且将不同数据类型的操作 API 封装到了不同的类型中：

3.4 SpringDataRedis 快速入门

**SpringBoot**已经提供了对**SpringDataRedis**的支持，使用非常简单

  • 首先新建一个 Spring Boot 工程
  • 然后引入连接池依赖
1
2
3
4
5
<!--连接池依赖-->
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-pool2</artifactId>
</dependency>
  • 编写配置文件**application.yml**（连接池的配置在实际开发中是根据需求来的）
1
2
3
4
5
6
7
8
9
10
11
spring:
  redis:
    host: 192.168.230.88 #指定redis所在的host
    port: 6379 #指定redis的端口
    password: 132537 #设置redis密码
    lettuce:
      pool:
        max-active: 8 #最大连接数
        max-idle: 8 #最大空闲数
        min-idle: 0 #最小空闲数
        max-wait: 100ms #连接等待时间
  • 编写测试类执行测试方法
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
@SpringBootTest
class RedisDemoApplicationTests {

	@Resource
	private RedisTemplate redisTemplate;

	@Test
	void testString() {
		// 1.通过RedisTemplate获取操作String类型的ValueOperations对象
		ValueOperations ops = redisTemplate.opsForValue();
		// 2.插入一条数据
		ops.set("blogName","Vz-Blog");

		// 3.获取数据
		String blogName = (String) ops.get("blogName");
		System.out.println("blogName = " + blogName);
	}
}

3.5 RedisSerializer 配置

RedisTemplate 可以接收任意 Object 作为值写入 Redis，只不过写入前会把 Object 序列化为字节形式，**默认是采用JDK序列化**，得到的结果是这样的

缺点：

  • 可读性差
  • 内存占用较大

那么如何解决以上的问题呢？我们可以通过自定义 RedisTemplate 序列化的方式来解决。

  • 编写一个配置类**RedisConfig**
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
@Configuration
public class RedisConfig {

    @Bean
    public RedisTemplate<String,Object> redisTemplate(RedisConnectionFactory factory){
        // 1.创建RedisTemplate对象
        RedisTemplate<String ,Object> redisTemplate = new RedisTemplate<>();
        // 2.设置连接工厂
        redisTemplate.setConnectionFactory(factory);

        // 3.创建序列化对象
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        GenericJackson2JsonRedisSerializer genericJackson2JsonRedisSerializer = new GenericJackson2JsonRedisSerializer();

        // 4.设置key和hashKey采用String的序列化方式
        redisTemplate.setKeySerializer(stringRedisSerializer);
        redisTemplate.setHashKeySerializer(stringRedisSerializer);

        // 5.设置value和hashValue采用json的序列化方式
        redisTemplate.setValueSerializer(genericJackson2JsonRedisSerializer);
        redisTemplate.setHashValueSerializer(genericJackson2JsonRedisSerializer);

        return redisTemplate;
    }
}
  • 此时我们已经将 RedisTemplate 的 key 设置为**String序列化**，value 设置为**Json序列化**的方式，再来执行方法测试
  • 由于我们设置的 value 序列化方式是 Json 的，因此我们可以直接向 redis 中插入一个对象
1
2
3
4
5
6
@Test
void testSaveUser() {
    redisTemplate.opsForValue().set("user:100", new User("Vz", 21));
    User user = (User) redisTemplate.opsForValue().get("user:100");
    System.out.println("User = " + user);
}


尽管 Json 序列化可以满足我们的需求，但是依旧存在一些问题。
如上图所示，为了在反序列化时知道对象的类型，JSON 序列化器会将类的 class 类型写入 json 结果中，存入 Redis，会带来额外的内存开销。
那么我们如何解决这个问题呢？我们可以通过下文的StringRedisTemplate来解决这个问题。

3.6 StringRedisTemplate

为了节省内存空间，我们并不会使用 JSON 序列化器来处理 value，而是统一使用 String 序列化器，要求只能存储 String 类型的 key 和 value。当需要存储 Java 对象时，手动完成对象的序列化和反序列化。

Spring 默认提供了一个 StringRedisTemplate 类，它的 key 和 value 的序列化方式默认就是 String 方式。省去了我们自定义 RedisTemplate 的过程

  • 我们可以直接编写一个测试类使用 StringRedisTemplate 来执行以下方法
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
@SpringBootTest
class RedisStringTemplateTest {

	@Resource
	private StringRedisTemplate stringRedisTemplate;

	@Test
	void testSaveUser() throws JsonProcessingException {
		// 1.创建一个Json序列化对象
		ObjectMapper objectMapper = new ObjectMapper();
		// 2.将要存入的对象通过Json序列化对象转换为字符串
		String userJson1 = objectMapper.writeValueAsString(new User("Vz", 21));
		// 3.通过StringRedisTemplate将数据存入redis
		stringRedisTemplate.opsForValue().set("user:100",userJson1);
		// 4.通过key取出value
		String userJson2 = stringRedisTemplate.opsForValue().get("user:100");
		// 5.由于取出的值是String类型的Json字符串，因此我们需要通过Json序列化对象来转换为java对象
		User user = objectMapper.readValue(userJson2, User.class);
		// 6.打印结果
		System.out.println("user = " + user);
	}

}
  • 执行完毕回到 Redis 的图形化客户端查看结果

3.7 总结

RedisTemplate 的两种序列化实践方案，两种方案各有各的优缺点，可以根据实际情况选择使用。

方案一：

  1. 自定义 RedisTemplate
  2. 修改 RedisTemplate 的序列化器为 GenericJackson2JsonRedisSerializer

方案二：

  1. 使用 StringRedisTemplate
  2. 写入 Redis 时，手动把对象序列化为 JSON
  3. 读取 Redis 时，手动把读取到的 JSON 反序列化为对象

二.Redis 实战

1.缓存

1.1 为什么用缓存

速度快，好用

缓存数据存储于代码中,而代码运行在内存中,内存的读写性能远高于磁盘,缓存可以大大降低用户访问并发量带来的服务器读写压力

1.2 缓存模型与思路

标准的操作方式就是查询数据库之前先查询缓存，如果缓存数据存在，则直接从缓存中返回，如果缓存数据不存在，再查询数据库，然后将数据存入 redis。

代码如下：

1.3 缓存更新策略

缓存更新是 redis 为了节约内存而设计出来的一个东西，主要是因为内存数据宝贵，当我们向 redis 插入太多数据，此时就可能会导致缓存中的数据过多，所以 redis 会对部分数据进行更新，或者把他叫为淘汰更合适。

  • 内存淘汰：当 redis 内存达到我们设置的阈值（max-memery）时，自动触发淘汰机制。
  • 超时剔除：给数据添加 TTL 时间，到时间自动剔除。
  • 主动更新：我们可以手动调用方法把缓存删掉，通常用于解决缓存和数据库不一致问题

1.3.1 缓存与数据库数据不一致解决方案及选择

缓存中的数据是来自于数据库的，但是数据库的数据是会改变的，如果数据库数据改变，缓存的数据没有改变，就会导致一致性问题，就会导致用户使用的是过时的数据，影响用户体验，有如下三种解决方案：

方法 描述
Cache Aside Pattern ✔ 人工编码方式：缓存调用者在更新完数据库后再去更新缓存，也称之为双写方案
Read/Write Through Pattern 由系统本身完成，数据库与缓存的问题交由系统本身去处理
Write Behind Caching Pattern 调用者只操作缓存，其他线程去异步处理数据库，实现最终一致

这里综合考虑我们选择方案一！✔

操作缓存和数据库时有三个问题需要考虑：

如果采用第一个方案，那么假设我们每次操作数据库后，都操作缓存，但是中间如果没有人查询，那么这个更新动作实际上只有最后一次生效，中间的更新动作意义并不大，我们可以把缓存删除，等待再次查询时，将缓存中的数据加载出来

  • 删除缓存还是更新缓存？
    • 更新缓存：每次更新数据库都更新缓存，无效写操作较多
    • 删除缓存：更新数据库时让缓存失效，查询时再更新缓存 ✔
  • 如何保证缓存与数据库的操作的同时成功或失败？
    • 单体系统，将缓存与数据库操作放在一个事务
    • 分布式系统，利用 TCC 等分布式事务方案

应该具体操作缓存还是操作数据库，我们应当是先操作数据库，再删除缓存，原因在于，如果你选择第一种方案，在两个线程并发来访问时，假设线程 1 先来，他先把缓存删了，此时线程 2 过来，他查询缓存数据并不存在，此时他写入缓存，当他写入缓存后，线程 1 再执行更新动作时，实际上写入的就是旧的数据，新的数据被旧数据覆盖了。

  • 先操作缓存还是先操作数据库？
    • 先删除缓存，再操作数据库
    • 先操作数据库，再删除缓存 ✔

1.4 缓存穿透

缓存穿透 ：缓存穿透是指客户端请求的数据在缓存中和数据库中都不存在，这样缓存永远不会生效，这些请求都会打到数据库。

解决方案：

  • 缓存空对象
    • 优点：实现简单，维护方便
    • 缺点：
      • 额外的内存消耗
      • 可能造成短期的不一致
  • 布隆过滤
    • 优点：内存占用较少，没有多余 key
    • 缺点：
      • 实现复杂
      • 存在误判可能（布隆过滤器走的是哈希思想，只要哈希思想，就可能存在哈希冲突）

编码思路：

小总结：

缓存穿透产生的原因是什么？

  • 用户请求的数据在缓存中和数据库中都不存在，不断发起这样的请求，给数据库带来巨大压力

缓存穿透的解决方案有哪些？

  • 缓存 null 值
  • 布隆过滤
  • 增强 id 的复杂度，避免被猜测 id 规律
  • 做好数据的基础格式校验
  • 加强用户权限校验
  • 做好热点参数的限流
  • 进行实时监控：当发现 Redis 的命中率开始急速降低，需要排查访问对象和访问的数据，和运维人员配合，可以设置黑名单限制服务
  • 设置可访问的名单（白名单）：使用 bitmaps 类型定义一个可以访问的名单，名单 id 作为 bitmaps 的偏移量，每次访问和 bitmap 里面的 id 进行比较，如果访问 id 不在 bitmaps 里面，进行拦截，不允许访问。

1.5 缓存击穿

缓存击穿问题也叫热点 Key 问题，就是一个被高并发访问并且缓存重建业务较复杂的key 突然失效了，无数的请求访问会在瞬间给数据库带来巨大的冲击。

常见的解决方案有两种：

  • 预先设置热门数据：把一些热门数据提前存入到 redis 里面，加大这些热门数据 key 的时长
  • 互斥锁
  • 逻辑过期

逻辑分析：当多个线程过来查询缓存，发现都没有命中，于是乎都去查数据库，然后数据库就压力山大，直接爆炸。

1.5.1 互斥锁解决缓存击穿

可以利用锁来防止多个线程同时去查询数据库，写缓存，只有拿到互斥锁的线程才能去查询数据库写缓存操作，其他线程拿不到互斥锁就休眠重试，等缓存中写入了之后，自然就能拿到数据。

加锁会影响查询的性能，因为此时会让查询的性能从并行变成了串行，我们可以采用 tryLock 方法 + double check 来解决这样的问题。

这里的互斥锁通过 setnx 的特性可以实现，setnx 只有在不存在数据时才能添加，有数据时添加失败。

1.5.2 逻辑过期解决缓存击穿

之所以会发生缓存击穿，是因为我们设置的 key 有过期时间，那么我们可以不给 key 设置过期时间啊，那这就有人会问，这不是会一直占用内存吗？，所以我们可以采用逻辑过期时间，把过期时间写入 value 中，用代码逻辑来判断该 key 是否过期。

这种方案巧妙在于，异步的构建缓存，缺点在于在构建完缓存之前，返回的都是脏数据。

1.5.3 两种方案的对比

互斥锁方案：由于保证了互斥性，所以数据一致，且实现简单，因为仅仅只需要加一把锁而已，也没其他的事情需要操心，所以没有额外的内存消耗，缺点在于有锁就有死锁问题的发生，且只能串行执行性能肯定受到影响

逻辑过期方案： 线程读取过程中不需要等待，性能好，有一个额外的线程持有锁去进行重构数据，但是在重构数据完成前，其他的线程只能返回之前的数据，且实现起来麻烦

1.5.4 业务代码实现

互斥锁方案：

逻辑过期方案：

1.6 缓存雪崩

缓存雪崩是指在同一时段大量的缓存 key 同时失效或者 Redis 服务宕机，导致大量请求到达数据库，带来巨大压力。

解决方案：

  • 给不同的 Key 的 TTL 添加随机值
  • 利用 Redis 集群提高服务的可用性
  • 给缓存业务添加降级限流策略
  • 给业务添加多级缓存：nginx 缓存 + redis 缓存 +其他缓存（ehcache 等）

2.分布式锁

2.1 基本原理和实现方式对比

分布式锁：满足分布式系统或集群模式下多进程可见并且互斥的锁。

分布式锁必须满足的条件：

  • 可见性：多个线程都能看到相同的结果，注意：这个地方说的可见性并不是并发编程中指的内存可见性，只是说多个进程之间都能感知到变化的意思
  • 互斥：互斥是分布式锁的最基本的条件，使得程序串行执行
  • 高可用：程序不易崩溃，时时刻刻都保证较高的可用性
  • 高性能：由于加锁本身就让性能降低，所有对于分布式锁本身需要他就较高的加锁性能和释放锁性能
  • 安全性：安全也是程序中必不可少的一环

常见的三种分布式锁以及对比：

2.2 Redis 实现分布式锁

实现分布式锁时需要实现的两个基本方法：

  • 获取锁：
    • 互斥：确保只能有一个线程获取锁
    • 非阻塞：尝试一次，成功返回 true，失败返回 false
  • 释放锁：
    • 手动释放（del key）
    • 超时释放：获取锁时添加一个超时时间

2.2.1 核心思路

利用 setnx 的特性：只有当 key 不存在时才可以设置成功，来实现分布式锁的互斥性，释放锁即把这个 key 删除即可。

2.2.2 实现分布式锁版本一

Ilock 接口

1
2
3
4
5
6
7
8
9
10
11
12
13
public interface ILock {
    /**
     * 尝试获取锁
     * @param timeoutSec 锁持有的超时时间，过期后自动释放
     * @return true代表获取锁成功; false代表获取锁失败
     */
    boolean tryLock(long timeoutSec);

    /**
     * 释放锁
     */
    void unlock();
}

实现类SimpleRedisLock

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
private static final String KEY_PREFIX="lock:"
//获取锁逻辑
@Override
public boolean tryLock(long timeoutSec) {
    // 获取线程标示
    String threadId = Thread.currentThread().getId()
    // 获取锁
    Boolean success = stringRedisTemplate.opsForValue()
            .setIfAbsent(KEY_PREFIX + name, threadId + "", timeoutSec, TimeUnit.SECONDS);
    return Boolean.TRUE.equals(success);
}

//释放锁逻辑
 public void unlock() {    
     //通过del删除锁    
     stringRedisTemplate.delete(KEY_PREFIX + name);
 }

2.2.3 实现分布式锁版本二

版本一面临的问题：因为线程可能阻塞导致锁超时自动释放，在释放锁时有可能释放的是别人的锁

解决方案：在释放锁的时候先判断该锁是不是自己的锁，是自己的锁才释放，不是自己的锁不释放。

核心逻辑：在存入锁时，放入自己线程的标识（UUID），在删除锁时，判断当前这把锁的标识是不是自己存入的，如果是，则进行删除，如果不是，则不进行删除。

具体代码修改：

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
//在value上拼接UUID作为唯一标识
private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";
//加锁逻辑
@Override
public boolean tryLock(long timeoutSec) {
   // 获取线程标示
   String threadId = ID_PREFIX + Thread.currentThread().getId();
   // 获取锁
   Boolean success = stringRedisTemplate.opsForValue()
                .setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);
   return Boolean.TRUE.equals(success);
}

//释放锁逻辑
public void unlock() {
    // 获取线程标示
    String threadId = ID_PREFIX + Thread.currentThread().getId();
    // 获取锁中的标示
    String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
    // 判断标示是否一致
    if(threadId.equals(id)) {
        // 释放锁
        stringRedisTemplate.delete(KEY_PREFIX + name);
    }
}

有关代码实操说明：

在我们修改完此处代码后，我们重启工程，然后启动两个线程，第一个线程持有锁后，手动释放锁，第二个线程 此时进入到锁内部，再放行第一个线程，此时第一个线程由于锁的 value 值并非是自己，所以不能释放锁，也就无法删除别人的锁，此时第二个线程能够正确释放锁，通过这个案例初步说明我们解决了锁误删的问题。

2.2.4 实现分布式锁版本三

考虑更加极端的情况：在删除锁的时候，已经判断成功唯一标识一致准备释放锁，在这之间进行了阻塞（JVM Full GC），那么仍然会导致误删的情况。

解决方案：让判断标志和释放锁两个动作具有原子性，保证一起进行中间不能停。

利用 lua 脚本来解决多条命令的原子性问题：这里不做讨论，我不会

unlock.lua

1
2
3
4
5
6
-- 比较线程标示与锁中的标示是否一致
if(redis.call('get', KEYS[1]) ==  ARGV[1]) then
    -- 释放锁 del key
    return redis.call('del', KEYS[1])
end
return 0
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;
    static {
        UNLOCK_SCRIPT = new DefaultRedisScript<>();
        UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));
        UNLOCK_SCRIPT.setResultType(Long.class);
    }

public void unlock() {
    // 调用lua脚本
    stringRedisTemplate.execute(
            UNLOCK_SCRIPT,
            Collections.singletonList(KEY_PREFIX + name),
            ID_PREFIX + Thread.currentThread().getId());
}
//经过以上代码改造后，我们就能够实现 拿锁比锁删锁的原子性动作了~

2.2.5 小总结

基于 Redis 的分布式锁实现思路：

  • 利用 set nx ex 获取锁，并设置过期时间，保存线程标示
  • 释放锁时先判断线程标示是否与自己一致，一致则删除锁
    • 特性：
      • 利用 set nx 满足互斥性
      • 利用 set ex 保证故障时锁依然能释放，避免死锁，提高安全性
      • 利用 Redis 集群保证高可用和高并发特性

我们一路走来，利用添加过期时间，防止死锁问题的发生，但是有了过期时间之后，可能出现误删别人锁的问题，这个问题我们开始是利用删之前 通过拿锁，比锁，删锁这个逻辑来解决的，也就是删之前判断一下当前这把锁是否是属于自己的，但是现在还有原子性问题，也就是我们没法保证拿锁比锁删锁是一个原子性的动作，最后通过 lua 表达式来解决这个问题

但是目前还剩下一个问题，锁不住，什么是锁不住呢，你想一想，如果当过期时间到了之后，我们可以给他续期一下，比如续个 30s，就好像是网吧上网， 网费到了之后，然后说，来，网管，再给我来 10 块的，是不是后边的问题都不会发生了，那么续期问题怎么解决呢，可以依赖于我们接下来要学习 redission 啦

2.3 Redission

Redission 的 github
Redisson 官网

2.3.1 setnx 实现的分布式锁的问题

  • 重入问题：重入问题是指 获得锁的线程可以再次进入到相同的锁的代码块中，可重入锁的意义在于防止死锁，比如 HashTable 这样的代码中，他的方法都是使用 synchronized 修饰的，假如他在一个方法内，调用另一个方法，那么此时如果是不可重入的，不就死锁了吗？所以可重入锁他的主要意义是防止死锁，我们的 synchronized 和 Lock 锁都是可重入的。
  • 不可重试：是指目前的分布式只能尝试一次，我们认为合理的情况是：当线程在获得锁失败后，他应该能再次尝试获得锁。
  • 超时释放：我们在加锁时增加了过期时间，这样的我们可以防止死锁，但是如果卡顿的时间超长，虽然我们采用了 lua 表达式防止删锁的时候，误删别人的锁，但是毕竟没有锁住，有安全隐患
  • 主从一致性： 如果 Redis 提供了主从集群，当我们向集群写数据时，主机需要异步的将数据同步给从机，而万一在同步过去之前，主机宕机了，就会出现死锁问题。

image.png

2.3.2 Redission 是什么

Redisson 是一个在 Redis 的基础上实现的 Java 驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的 Java 常用对象，还提供了许多分布式服务，其中就包含了各种分布式锁的实现。
image.png

2.3.3 Redission 可重入锁原理

就是将原本 string 类型的 value 转换成了 hash 类型，hash 对应的 field 是线程名，value 则是一个计数器，当前线程每获取一次锁，则 value 加一，删除锁 value 减一，若 value 等于零则 del 锁。利用 lua 脚本实现。
image.png

2.3.4 Redission 锁重试和 WatchDog 机制

实战篇-20.分布式锁-Redisson 的锁重试和 WatchDog 机制哔哩哔哩 bilibili

2.3.5 Redission 锁的 MutiLock 原理

实战篇-21.分布式锁-Redisson 的 multiLock 原理哔哩哔哩 bilibili