分布式文件存储与数据缓存（二）| Redis

Redis概述_什么是NoSQL

泛指非关系型的数据库。

NoSQL的四大分类

KV型NoSql（代表----Redis）

KV型NoSql顾名思义就是以键值对形式存储的非关系型数据库
特点：

• 数据基于内存，读写效率高
• KV型数据，时间复杂度为O(1)，查询速度快

列式NoSql（代表----HBase）

列式NoSql，大数据时代最具代表性的技术之一了，以HBase为代表。

注意：
查询时只有指定的列会被读取，不会读取所有列
列数据被组织到一起，一次磁盘IO可以将一列数据一次性读取到内存中

文档型NoSql（代表----MongoDB）

文档型NoSql指的是将半结构化数据存储为文档的一种NoSql，文档型NoSql通常以JSON或者XML格式存储数据。

搜索型NoSql（代表----ElasticSearch）

传统关系型数据库主要通过索引来达到快速查询的目的，但是在全文搜索的场景下，索引是无能为力的，like查询一来无法满足所有模糊匹配需求，二来使用限制太大且使用不当容易造成慢查询，搜索型NoSql的诞生正是为了解决关系型数据库全文搜索能力较弱的问题，ElasticSearch是搜索型NoSql的代表产品。

关系型数据库和非关系型数据及其区别

关系型数据库

关系型数据库最典型的数据结构是表，由二维表及其之间的联系所组成的一个数据组织

优点：

• 易于维护：都是使用表结构，格式一致；
• 使用方便：SQL语言通用，可用于复杂查询；
• 复杂操作：支持SQL，可用于一个表以及多个表之间非常复杂的查询。

缺点：

• 读写性能比较差，尤其是海量数据的高效率读写；
• 固定的表结构，灵活度稍欠；

非关系型数据库

优点：

• 格式灵活：存储数据的格式可以是key,value形式、文档形式、图片形式等等，文档形式、图片形式等等，使用灵活，应用场景广泛，而关系型数据库则只支持基础类型。
• 速度快：nosql可以使用硬盘或者随机存储器作为载体，而关系型数据库只能使用硬盘；
• 高扩展性；
• 成本低：nosql数据库部署简单，基本都是开源软件。

缺点：

• 不提供sql支持，学习和使用成本较高；
• 无事务处理；
• 数据结构相对复杂，复杂查询方面稍欠。

Redis概述_Redis是什么

Redis是一个使用ANSI C编写的开源、包含多种数据结构、支持网络、基于内存、可选持久性的键值对存储数据库。

特性：

• 基于内存运行，性能高效
• 支持分布式，理论上可以无限扩展
• key-value存储系统
• 开源的使用ANSI C语言编写、遵守BSD协议、支持网络、可基于内存亦可持久化的日志型、KeyValue数据库，并提供多种语言的API

Redis安装_Linux下安装Redis

下载地址

Redis官方网址：https://redis.io/

wget https://github.com/redis/redis/archive/7.2.4.tar.gz
tar -zcvf 7.2.4.tar.gz
1
2

安装GCC

yum install -y gcc
1

编译Redis

在redis-7.2.4目录下执行：

make && make install
1

在Redis根目录下会出现一个src文件
redis-benchmark：Redis自带的基准性能测试工具
redis-check-aof：对有问题的 AOF 文件进行修复，AOF和RDB文件后面会说明
redis-check-rdb：对有问题的 RDB文件进行修复
redis-sentinel：Redis集群使用
redis-cli：客户端
redis-server：服务器启动

前台启动：

./redis-server
1

后台启动

修改redis.conf文件

vim redis.conf

# 修改其中的
daemonize yes     #由no改为yes
1
2
3
4

启动服务

 ./redis-server ../redis.conf
1

客户端启动

./redis-cli
1

Redis安装_Docker下安装Redis

下载最新Redis镜像

docker pull redis
1

启动Redis容器

docker run -d --name myredis -p 6379:6379 redis
1

进入Redis容器

docker exec -it myredis /bin/bash
1

Redis安装_基本知识

默认16数据库

Redis默认支持16个数据库，可以通过调整Redis的配置文件redis/redis.conf中的databases来修改这一个值，设置完毕后重启Redis便完成配置。

Redis 使用的到底是多线程还是单线程？

因为Redis是基于内存的操作，CPU不是Redis的瓶颈，Redis的瓶颈最有可能是机器内存的大小或者网络带宽。既然单线程容易实现，而且CPU不会成为瓶颈，那就顺理成章地采用单线程的方案了。

IO多路复用技术

redis 采用网络IO多路复用技术来保证在多连接的时候， 系统的高吞吐量。

这里“多路”指的是多个网络连接，“复用”指的是复用同一个线程。采用多路 I/O 复用技术可以让单个线程高效的处理多个连接请求（尽量减少网络IO的时间消耗），且Redis在内存中操作数据的速度非常快（内存内的操作不会成为这里的性能瓶颈），主要以上两点造就了Redis具有很高的吞吐量。

切换数据库

# 使用 1 号数据库
redis 127.0.0.1:6379> select 1 
1
2

清空当前库

127.0.0.1:6379> flushdb
1

清空全部库

redis 127.0.0.1:6379>flushall
1

Redis数据类型_key键

查看当前库中所有的key 。

keys *
1

生产已经禁止。因为长时间阻塞redis而导致其他客户端的命令请求一直处于阻塞状态。 更安全的做法是采用scan。
redis-cli --scan "*"

判断某个key是否存在

返回1表示存在，0不存在

exists key
1

查看当前key 所储存的值的类型。

type key
1

删除已存在的key

不存在的 key 会被忽略

del key
1

给key设置time秒的过期时间。

设置成功返回 1 。 当 key 不存在返回 0。

expire key time
1

以秒为单位返回 key 的剩余过期时间。

ttl key
1

移除给定 key 的过期时间

使得 key 永不过期

persist key
1

Redis数据类型_String

String是Redis最基本的类型，一个key对应一个value。String是二进制安全的，意味着String可以包含任何数据，比如序列化对象或者一张图片。String最多可以放512M的数据。

常用命令

set

用于设置给定 key 的值。如果 key 已经存储其他值， set 就重写旧值，且无视类型。

set key value
1

get

用于获取指定 key 的值。如果 key 不存在，返回 nil 。

get key
1

append

将给定的value追加到key原值末尾。
如果 key 已经存在并且是一个字符串， append 命令将 value 追加到 key 原来的值的末尾。
如果 key 不存在， append 就简单地将给定 key 设为 value ，就像执行 set key value 一样。

append key value
1

strlen

获取指定 key 所储存的字符串值的长度。当 key 储存的不是字符串值时，返回一个错误。

strlen key
1

setex

给指定的 key 设置值及time 秒的过期时间。如果 key 已经存在，setex命令将会替换旧的值，并设置过期时间。

setex key time value
1

setnx

只有在key不存在时设置key的值

setnx key value
1

getrange

获取指定区间范围内的值，类似between…and 的关系

getrange key start end
1

setrange

替换指定区间范围内的值，类似between…and 的关系

setrange key offset value
1

incr/decr

将 key 中储存的数字值增一/减一。
如果 key 不存在，那么 key 的值会先被初始化为 0 ，然后再执行 incr 操作。
如字符串类型的值不能表示为数字、或者是其他类型，那么返回一个错误。

incr key
decr key
1
2

incrby/decrby key step

将key存储的数字值按照step进行增减。

127.0.0.1:6379> incrby k1 10
127.0.0.1:6379> decrby k1 10
1
2

mset

同时设置一个或多个 key-value 。

mset key1 value1 key2 value2
1

mget

返回所有(一个或多个)给定 key 的值。

mget key1 key2
1

getset

将给定key值设为value，并返回key的旧值（old value），简单一句话（先get然后立即set）。

getset key value
1

使用场景

value 除了是字符串以外还可以是数字。

• 计数器
• 统计多单位的数量
• 粉丝数
• 对象缓存存储
• 分布式锁

Redis数据类型_List

List是简单的字符串列表，按照插入顺序排序。你可以添加一个元素到列表的头部（左边）或者尾部（右边）。底层是一个双向链表，对两段操作性能极高，通过索引操作中间的节点性能较差。
一个List最多可以包含 $2^{32}-1$个元素 （ 每个列表超过40亿个元素）。

常用命令

Redis数据类型_Set

Redis数据类型_Hash

Redis数据类型_Zset

Redis数据类型_Bitmaps

Redis数据类型_Geospatia

Redis数据类型_Hyperloglog

连接Redis服务

关闭防火墙

修改redis.conf配置文件

# 关闭保护模式
protected-mode no
# 开启远程访问 可以使所有的ip访问redis
# bind 127.0.0.1 
1
2
3
4

Java整合Redis_Jedis操作

Jedis是Redis官方推荐的Java连接开发工具。

创建maven工程 引入maven依赖

<dependency>
   <groupId>redis.clients</groupId>
   <artifactId>jedis</artifactId>
   <version>3.6.0</version>
</dependency>
<dependency>
   <groupId>junit</groupId>
   <artifactId>junit</artifactId>
   <version>4.12</version>
   <scope>test</scope>
</dependency>
1
2
3
4
5
6
7
8
9
10
11

Jedis连接到redis

/**
 * jedis测试用例
 */
public class JedisTest {

    Jedis jedis;

    /**
     * 初始化jedis实例
     */
    @Before
    public void init() {
        // 创建redis连接实例
        jedis = new Jedis("192.168.52.138", 6379);
    }

    /**
     * String 操作
     */
    @Test
    public void stringTest() {
        // 设置一个key
        jedis.set("name", "zhangsan");
        String s = jedis.get("name");
        System.out.println(s);
        // 设置key过期时间为10s
        jedis.setex("sex", 10, "man");
    }

    /**
     * List 操作
     */
    @Test
    public void listTest() {
        // 添加元素操作
        jedis.lpush("list1", "1", "2", "3");
        jedis.rpush("list1", "4", "5", "6");
        // 获取所有元素
        List<String> list1 = jedis.lrange("list1", 0, -1);
        list1.forEach(System.out::println);
    }

    /**
     * Set 操作
     */
    @Test
    public void setTest() {
        // 设置一个key
        jedis.sadd("name3", "v1", "v2", "v3");
        // 获取元素
        Set<String> set1 = jedis.smembers("name3");
        set1.forEach(System.out::println);
    }

    /**
     * Hash 操作
     */
    @Test
    public void hashTest() {
        //设置一个hash
        jedis.hset("user", "age", "25");
        //获取该key的所有value
        List<String> user = jedis.hvals("user");
        user.forEach(System.out::println);
    }

    /**
     *  zset 操作
     */
    @Test
    public void zsetTest() {
        //向zset中添加一条数据
        jedis.zadd("zset1", 100, "java");
        //获取所有的值
        Set<String> zset1 = jedis.zrange("zset1", 0, -1);
        zset1.forEach(System.out::println);
    }

    /**
     *  Bitmaps  操作
     */
    @Test
    public void  bitTest() {
        // 给张三添加上班打开记录
        jedis.setbit("zhangsan",0, "1");
        jedis.setbit("zhangsan",1, "1");
        jedis.setbit("zhangsan",2, "0");
        jedis.setbit("zhangsan",3, "1");
        // 获取张三第三天上班记录
        Boolean getbit = jedis.getbit("zhangsan", 1);
        // 如果是1 true 0 false
        System.out.println(getbit);
    }

    /**
     *  Geospatia  操作
     */
    @Test
    public void  geoTest() {
        //添加一条地理信息数据
        jedis.geoadd("china",130,10,"beijing");
        // 获取地理信息
        List<GeoCoordinate> geo = jedis.geopos("china", "beijing");
        geo.forEach(System.out::println);
    }

    /**
     *  Hyperloglog  操作
     */
    @Test
    public void hllTest () {
        //将所有元素参数添加到 Hyperloglog 数据结构中。
        jedis.pfadd("book","c++","java","php");
        //获取Hyperloglog数据结构中所有元素的个数
        Long count = jedis.pfcount("book");
        System.out.println(count);
    }

    /**
     * 关闭jedis实例
     */
    @After
    public void close() {
        //关闭redis连接
        jedis.close();
    }
}

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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

Java整合Redis_Spring-Data-Redis

Spring-Data-Redis是spring大家族的一部分，通过简单的配置访问Redis服务，对Reids底层开发包(Jedis, JRedis, and RJC)进行了高度封装，RedisTemplate提供了Redis各种操作、异常处理及序列化，支持发布订阅。

RedisTemplate介绍

Spring封装了RedisTemplate对象来进行对Redis的各种操作，它支持所有的Redis原生的api。

RedisTemplate中定义了对5种数据结构操作

redisTemplate.opsForValue();//操作字符串
redisTemplate.opsForHash();//操作hash
redisTemplate.opsForList();//操作list
redisTemplate.opsForSet();//操作set
redisTemplate.opsForZSet();//操作有序set
1
2
3
4
5

StringRedisTemplate与RedisTemplate

• 两者的关系是StringRedisTemplate继承RedisTemplate。
• 两者的数据是不共通的；也就是说StringRedisTemplate只能管理StringRedisTemplate里面的数据，RedisTemplate只能管理RedisTemplate中的数据。

SDR默认采用的序列化策略有两种，一种是String的序列化策略，一种是JDK的序列化策略。
StringRedisTemplate默认采用的是String的序列化策略，保存的key和value都是采用此策略序列化保存的。
RedisTemplate默认采用的是JDK的序列化策略，保存的key和value都是采用此策略序列化保存的。

pom.xml添加依赖

 <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis-reactive</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

在application.properties中配置

#Redis服务器连接地址
spring.redis.host=192.168.52.138
#Redis服务器连接端口
spring.redis.port=6379
#连接池最大连接数（使用负值表示没有限制）
spring.redis.pool.max-active=8
#连接池最大阻塞等待时间（使用负值表示没有限制）
spring.redis.pool.max-wait=-1
#连接池中的最大空闲连接
spring.redis.pool.max-idle=8
#连接池中的最小空闲连接
spring.redis.pool.min-idle=0
#连接超时时间（毫秒）
spring.redis.timeout=30000
1
2
3
4
5
6
7
8
9
10
11
12
13
14

自定义序列化

/**
 * 自定义序列化方式
 */
@Configuration
public class RedisConfig {
   @Bean
   public RedisTemplate<String,Object> redisTemplate(RedisConnectionFactory redisConnectionFactory){
     RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
     redisTemplate.setKeySerializer(new StringRedisSerializer());
     redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());
     redisTemplate.setHashKeySerializer(new StringRedisSerializer());
     redisTemplate.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());
     redisTemplate.setConnectionFactory(redisConnectionFactory);
     return redisTemplate;
   }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

使用redisTemplate进行各类型的CURD操作

@SpringBootTest
class SpingdataredisDemoApplicationTests {

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;


    /**
     * redisTemplate 操作redis服务
     */
    @Test
    void contextLoads() {
        // 保存数据
        redisTemplate.opsForValue().set("k1", "v1");
        // 获取数据
        System.out.println(redisTemplate.opsForValue().get("k1"));
    }

    /**
     *  list 操作
     */
    @Test
    void listTest() {
        // 保存数据
        redisTemplate.opsForList().rightPush("k2", "v2");
        redisTemplate.opsForList().rightPush("k2", "v3");
        redisTemplate.opsForList().rightPush("k2", "v4");
        redisTemplate.opsForList().rightPush("k2", "v5");
        // 获取数据
        System.out.println(redisTemplate.opsForList().range("k2", 0, -1));
    }

    /**
     *  hash 操作
     */
    @Test
    void hashTest() {
        // 保存数据
        redisTemplate.opsForHash().put("user","name","zhangang");
        redisTemplate.opsForHash().put("user","age","18");
        // 获取数据
        System.out.println(redisTemplate.opsForHash().entries("user"));
    }

    /**
     *  set 操作
     */
    @Test
    void setTest() {
        // 保存数据
        redisTemplate.opsForSet().add("s1", "v1");
        redisTemplate.opsForSet().add("s1", "v1");
        redisTemplate.opsForSet().add("s1", "v2");
        redisTemplate.opsForSet().add("s1", "v2");
        // 获取数据
        System.out.println(redisTemplate.opsForSet().members("s1"));

        // 获取集合长度
        System.out.println(redisTemplate.opsForSet().size("s1"));
    }

    /**
     * zset 操作
     */
    @Test
    void zsetTest() {
        // 保存数据
        redisTemplate.opsForZSet().add("z1", "v1", 100);
        redisTemplate.opsForZSet().add("z1", "v2", 87);
        redisTemplate.opsForZSet().add("z1", "v3", 67);
        redisTemplate.opsForZSet().add("z1", "v4", 45);
        // 获取数据
        System.out.println(redisTemplate.opsForZSet().range("z1", 0, -1));
        // 通过分数获取数据
        System.out.println(redisTemplate.opsForZSet().rangeByScore("z1", 50, 100));
    }
}
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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

Redis构建web应用实践_网页缓存

Redis其他功能_发布与订阅

![在这里插入图片描述](https://img-blog.csdnimg.cn/direct/85a0ec7921324c8fb690561fe249df5b.png

Redis 发布订阅 (pub/sub) 是一种消息通信模式：发送者 (pub) 发送消息，订阅者 (sub) 接收消息。

看到发布订阅的特性，用来做一个简单的实时聊天系统再适合不过了。再比如，在一个博客网站中，有100个粉丝订阅了你，当你发布新文章，就可以推送消息给粉丝们拉。

发布订阅命令行实现

订阅

127.0.0.1:6379> subcribe 主题名字
1

发布

127.0.0.1:6379> publish 主题名称 hello
1

Redis其他功能_慢查询

Redis命令执行的整个过程

两点说明：

1. 慢查询发生在第3阶段
2. 客户端超时不一定慢查询，但慢查询是客户端超时的一个可能因素
3. 慢查询日志是存放在Redis内存列表中。

慢查询日志是Redis服务端在命令执行前后计算每条命令的执行时长，当超过某个阈值是记录下来的日志。日志中记录了慢查询发生的时间，还有执行时长、具体什么命令等信息，它可以用来帮助开发和运维人员定位系统中存在的慢查询。

如何获取慢查询日志

可以使用slowlog get命令获取慢查询日志，在slowlog get后面还可以加一个数字，用于指定获取慢查询日志的条数，比如，获取3条慢查询日志：

127.0.0.1:6379> SLOWLOG get 3
1) 1) (integer) 0
   2) (integer) 1640056567
   3) (integer) 11780
   4) 1) "FLUSHALL"
   5) "127.0.0.1:43406"
   6) ""
1
2
3
4
5
6
7

参数：
1 唯一标识ID
2 命令执行的时间戳
3 命令执行时长
4 执行的命名和参数

如何获取慢查询日志的条数

> slowlog len
(integer) 121 # 当前Redis中有121条慢查询日志。
1
2

配置慢查询的参数

slowlog-log-slower-than。

slowlog-log-slower-than的作用是指定命令执行时长的阈值，执行命令的时长超过这个阈值时就会被记录下来。

slowlog-max-len。

slowlog-max-len的作用是指定慢查询日志最多存储的条数。实际上，Redis使用了一个列表存放慢查询日志，slowlog-max-len就是这个列表的最大长度。

查看慢日志配置

127.0.0.1:6379> config get slow*
1) "slowlog-max-len"
2) "128"
3) "slowlog-log-slower-than"
4) "10000"
1
2
3
4
5

慢日志说明：
10000阈值，单位微秒，此处为10毫秒
128慢日志记录保存数量的阈值，此处保存128条。

修改慢查询参数

修改Redis配置文件

vim redis.conf
1

slowlog-log-slower-than 1000
slowlog-max-len 1200
1
2

使用 config set 命令动态修改。

> config set slowlog-log-slower-than 1000
OK
> config set slowlog-max-len 1200
OK
> config rewrite
OK
1
2
3
4
5
6

实践建议

slowlog-max-len配置建议

• 线上建议调大慢查询列表，记录慢查询时Redis会对长命令做截断操作，并不会占用大量内存。
• 增大慢查询列表可以减缓慢查询被剔除的可能，例如线上可设置为1000以上。

slowlog-log-slower-than配置建议

• 默认值超过10毫秒判定为慢查询，需要根据Redis并发量调整该值。
• 由于Redis采用单线程响应命令，对于高流量的场景，如果命令执行时间在1毫秒以上，那么Redis
  最多可支撑OPS不到1000。因此对于高OPS场景的Redis建议设置为1毫秒。

OPS通常指的是“每秒操作次数”

Redis其他功能_流水线pipeline

经历了 1次pipeline(n条命令) = 1次网络时间 + n次命令时间，这大大减少了网络时间的开销，这就是流水线。

在执行批量操作而没有使用pipeline功能，会将大量的时间耗费在每一次网络传输的过程上；而使用pipeline后，只需要经过一次网络传输，然后批量在redis端进行命令操作。这会大大提高
了效率。

pipeline-Jedis实现

引入jedis依赖包

<dependency>
 <groupId>redis.clients</groupId>
 <artifactId>jedis</artifactId>
 <version>2.9.0</version>
</dependency> 
1
2
3
4
5

 private Jedis jedis;
 
/**
 * 初始化jedis实例
 */
@Before
public void init() {
    // 创建redis连接实例
    jedis = new Jedis("192.168.52.138", 6379);
}
 
/**
 * 没有 Pipeline  操作
 * 耗时 2066
 */
@Test
public void noPipelineTest() {
    // 开始时间
    long start = System.currentTimeMillis();

    for (int i = 0; i < 10000; i++) {
        jedis.hset("hashkey:" + i, "field" + i, "value" + i);
    }
    // 结束时间
    long end = System.currentTimeMillis();
    System.out.println(end - start);
}

/**
 * 有 Pipeline 操作
 * 耗时 62
 */
@Test
public void pipelineTest() {
    // 开始时间
    long start = System.currentTimeMillis();
    for (int i = 0; i < 100; i++) {
        Pipeline pipeline = jedis.pipelined();
        for (int j = i * 100; j < (i + 1) * 100; j++) {
            pipeline.hset("hashkey:" + j, "field" + j, "value" + j);
        }
        pipeline.syncAndReturnAll();
    }
    // 结束时间
    long end = System.currentTimeMillis();
    System.out.println(end - start);
}
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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

Redis技术中Pipeline主要作用是减少了网络时间的开销

Redis数据安全_持久化机制概述

由于Redis的数据都存放在内存中，如果没有配置持久化，Redis重启后数据就全丢失了，于是需要开启Redis的持久化功能，将数据保存到磁盘上，当Redis重启后，可以从磁盘中恢复数据。

对于Redis而言，持久化机制是指把内存中的数据存为硬盘文件，这样当Redis重启或服务器故障时能根据持久化后的硬盘文件恢复数据。

持久化机制的意义

redis持久化的意义，在于故障恢复。比如部署了一个redis，作为cache缓存，同时也可以保存一些比较重要的数据。

Redis提供了两个不同形式的持久化方式

RDB(Redis DataBase)

RDB是什么在指定的时间间隔内将内存的数据集快照写入磁盘，也就是行话讲的快照，它恢复时是将快照文件直接读到内存里。这种格式是经过压缩的二进制文件。

配置dump.rdb文件

RDB保存的文件，在redis.conf中配置文件名称，默认为dump.rdb。

440 # The filename where to dump the DB
441 dbfilename dump.rdb
1
2

rdb文件的保存路径，也可以修改。默认为Redis启动时命令行所在的目录下

454 dir ./
1

触发机制-主要三种方式

RDB配置

在配置文件中快照默认配置：

# save 3600 1：表示3600秒内（一小时）如果至少有1个key的值变化，则保存。
# save 300 100：表示300秒内（五分钟）如果至少有100个 key 的值变化，则保存。
# save 60 10000：表示60秒内如果至少有 10000个key的值变化，则保存。
1
2
3

flushall

执行flushall命令，也会触发rdb规则。

save与bgsave

手动触发Redis进行RDB持久化的命令有两种：
1. save
该命令会阻塞当前Redis服务器，执行save命令期间，Redis不能处理其他命令，直到RDB过程完成为止，不建议使用。
2. bgsave
执行该命令时，Redis会在后台异步进行快照操作，快照同时还可以响应客户端请求。

高级配置

stop-writes-on-bgsave-error

默认值是yes。当Redis无法写入磁盘的话，直接关闭Redis的写操作。

rdbcompression

默认值是yes。对于存储到磁盘中的快照，可以设置是否进行压缩存储。如果是的话，redis会采用LZF算法进行压缩。如果你不想消耗CPU来进行压缩的话，可以设置为关闭此功能，但是存储在磁盘上的快照会比较大。

rdbchecksum

默认值是yes。在存储快照后，我们还可以让redis使用CRC64算法来进行数据校验，但是这样做会增加大约10%的性能消耗，如果希望获取到最大的性能提升，可以关闭此功能。

恢复数据

只需要将rdb文件放在Redis的启动目录，Redis启动时会自动加载dump.rdb并恢复数据。

优势

• 适合大规模的数据恢复
• 对数据完整性和一致性要求不高更适合使用
• 节省磁盘空间
• 恢复速度快

劣势
在备份周期在一定间隔时间做一次备份，所以如果Redis意外down掉的话，就会丢失最后一次快照后的所有修改。

AOF(Append Only File)

以日志的形式来记录每个写操作，将Redis执行过的所有写指令记录下来。
AOF默认不开启
可以在redis.conf中配置文件名称，默认为appendonly.aof。

AOF文件的保存路径，同RDB的路径一致，如果AOF和RDB同时启动，Redis默认读取AOF的数据。

开启AOF

appendonly yes
1

AOF同步频率设置

# appendfsync always   # 始终同步，每次Redis的写入都会立刻记入日志，性能较差但数据完整性比较好。
appendfsync everysec # 每秒同步，每秒记入日志一次，如果宕机，本秒的数据可能丢失。
# appendfsync no       # redis不主动进行同步，把同步时机交给操作系统。
1
2
3

优势

• 备份机制更稳健，丢失数据概率更低。
• 可读的日志文本，通过操作AOF稳健，可以处理误操作。

劣势

• 比起RDB占用更多的磁盘空间。
• 恢复备份速度要慢。
• 每次读写都同步的话，有一定的性能压力。

如何选用持久化方式

综合使用AOF和RDB两种持久化机制
用AOF来保证数据不丢失，作为数据恢复的第一选择，用RDB来做不同程度的冷备，在AOF文件都丢失或损坏不可用的时候，还可以使用RDB来进行快速的数据恢复。

事务的概念与ACID特性

Redis事务

Redis事务是一组命令的集合，一个事务中的所有命令都将被序列化，按照一次性、顺序性、排他性的执行一系列的命令。

Redis事务三大特性

**单独的隔离操作：**事务中的所有命令都会序列化、按顺序地执行。事务在执行的过程中，不会被其他客户端发送来的命令请求所打断；
**没有隔离级别的概念：**队列中的命令没有提交之前都不会实际的被执行，因为事务提交前任何指令都不会被实际执行，也就不存在”事务内的查询要看到事务里的更新，在事务外查询不能看到”。
不保证原子性： redis同一个事务中如果有一条命令执行失败，其后的命令仍然会被执行，没有回滚；

Redis事务执行的三个阶段

• 开启： 以 MULTI 开始一个事务；
• 入队： 将多个命令入队到事务中，接到这些命令并不会立即执行，而是放到等待执行的事务队列里面；
• 执行： 由 EXEC 命令触发事务；

Redis事务_Redis事务基本操作

Multi、Exec、discard

事务从输入Multi命令开始，输入的命令都会依次压入命令缓冲队列中，并不会执行，直到输入Exec后，Redis会将之前的命令缓冲队列中的命令依次执行。
组队过程中，可以通过discard来放弃组队。

命令集合中含有错误的指令（注意是语法错误），均连坐，全部失败。

127.0.0.1:6379> MULTI
OK
127.0.0.1:6379(TX)> set name z3
QUEUED
127.0.0.1:6379(TX)> get name
QUEUED
127.0.0.1:6379(TX)> incr t1
QUEUED
127.0.0.1:6379(TX)> get t1
QUEUED
127.0.0.1:6379(TX)> set email
(error) ERR wrong number of arguments for 'set' command
127.0.0.1:6379(TX)> exec
(error) EXECABORT Transaction discardedbecause of previous errors.
1
2
3
4
5
6
7
8
9
10
11
12
13
14

运行时错误，即非语法错误，正确命令都会执行，错误命令返回错误。

127.0.0.1:6379> MULTI
OK
127.0.0.1:6379(TX)> set age 11
QUEUED
127.0.0.1:6379(TX)> incr t1
QUEUED
127.0.0.1:6379(TX)> set email abc@163.com
QUEUED
127.0.0.1:6379(TX)> incr email
QUEUED
127.0.0.1:6379(TX)> get age
QUEUED
127.0.0.1:6379(TX)> exec
1) OK
2) (integer) 5
3) OK
4) (error) ERR value is not an integer or out of range
5) "11"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

Redis集群_主从复制

redis单机服务面临问题

• 机器故障。我们部署到一台 Redis 服务器，当发生机器故障时，需要迁移到另外一台服务器并且要保证数据是同步的。
• 容量瓶颈。当我们有需求需要扩容 Redis 内存时，从 16G 的内存升到 64G，单机肯定是满足不了。当然，你可以重新买个 128G 的新机器。

解决办法
要实现分布式数据库的更大的存储容量和承受高并发访问量，我们会将原来集中式数据库的数据分别存储到其他多个网络节点上。

Redis 为了解决这个单一节点的问题，也会把数据复制多个副本部署到其他节点上进行复制，实现 Redis的高可用，实现对数据的冗余备份从而保证数据和服务的高可用。

主从复制，是指将一台Redis服务器的数据，复制到其他的Redis服务器。前者称为主节点(master)，后者称为从节点(slave),数据的复制是单向的，只能由主节点到从节点。

主从复制的作用

1. 数据冗余：主从复制实现了数据的热备份，是持久化之外的一种数据冗余方式。故障恢复：当主节点出现问题时，可以由从节点提供服务，实现快速的故障恢复；实际上是一种服务的冗余。
2. 负载均衡：在主从复制的基础上，配合读写分离，可以由主节点提供写服务，由从节点提供读服务（即写Redis数据时应用连接主节点，读Redis数据时应用连接从节点），分担服务器负载；尤其是在写少读多的场景下，通过多个从节点分担读负载，可以大大提高Redis服务器的并发量。
3. 高可用基石：除了上述作用以外，主从复制还是哨兵和集群能够实施的基础，因此说主从复制是Redis高可用的基础。

Redis集群_主从复制环境搭建

编写配置文件

新建redis6379.conf

include /usr/local/redis-7.2.4/redis.conf
pidfile /var/run/redis_6379.pid
port 6379
dbfilename dump6379.rdb
1
2
3
4

新建redis6380.conf

include /usr/local/redis-7.2.4/redis.conf
pidfile /var/run/redis_6380.pid
port 6380
dbfilename dump6380.rdb
1
2
3
4

新建redis6381.conf

include /usr/local/redis-7.2.4/redis.conf
pidfile /var/run/redis_6381.pid
port 6381
dbfilename dump6381.rdb
1
2
3
4

启动三台redis服务器

./redis-server ../redis6379.conf
./redis-server ../redis6380.conf
./redis-server ../redis6381.conf
1
2
3

查看系统进程

[root@localhost redis-7.2.4]# ps aux | grep redis
root      17303  0.0  0.4 241784  8968 ?        Ssl  02:18   0:01 ./redis-server 127.0.0.1:6379
root      17363  0.0  0.4 239224  8836 ?        Ssl  02:20   0:01 ./redis-server 127.0.0.1:6380
root      17369  0.0  0.4 239224  8840 ?        Ssl  02:20   0:01 ./redis-server 127.0.0.1:6381
root      17881  0.0  0.0 112708   976 pts/0    S+   02:52   0:00 grep --color=auto redis
1
2
3
4
5
6

查看三台主机运行情况

#打印主从复制的相关信息
./redis-cli -p 6379
./redis-cli -p 6380
./redis-cli -p 6381
127.0.0.1:6379> info replication
127.0.0.1:6380> info replication
127.0.0.1:6381> info replication
1
2
3
4
5
6
7

配从库不配主库

在6380和6381上执行。

127.0.0.1:6380> SLAVEOF 127.0.0.1 6379
1

Redis集群_主从复制原理剖析

主从复制可以分为3个阶段

连接建立阶段（即准备阶段）
数据同步阶段
命令传播阶段

复制过程大致分为6个过程

1、保存主节点（master）信息。
2、从节点（slave）内部通过每秒运行的定时任务维护复制相关逻辑，当定时任务发现存在新的主节点后，会尝试与该节点建立网络连接

从节点会建立一个 socket 套接字，从节点建立了一个端口为51234的套接字，专门用于接受主节点发送的复制命令。

3、发送ping命令
连接建立成功后从节点发送 ping 请求进行首次通信。

作用：
检测主从之间网络套接字是否可用。
检测主节点当前是否可以接受命令 。

4、权限验证。
如果主节点设置了requirepass参数，则需要密码验证，从节点必须配置masterauth参数保证与主节点相同的密码才能通过验证；
如果验证失败复制将终止，从节点重新发起复制流程。
5、同步数据集。
主从复制连接正常通信后，对于首次建立复制的场景，主节点会把持有的数据全部发送给从节点，这部分操作是耗时最长的步骤。

主从同步策略
主从刚刚连接的时候，进行全量同步；全同步结束后，进行增量同步。当然，如果有需要，slave 在任何时候都可以发起全量同步。
redis 策略是，无论如何，首先会尝试进行增量同步，如不成功，要求从机进行全量同步。

6、命令持续复制。
当主节点把当前的数据同步给从节点后，便完成了复制的建立流程。接下来主节点会持续地把写命令发送给从节点，保证主从数据一致性。

Redis集群_ 哨兵监控

哨兵模式是一种特殊的模式，首先Redis提供了哨兵的命令，哨兵是一个独立的进程，作为进程，它会独立运行。其原理是哨兵通过发送命令，等待Redis服务器响应，从而监控运行的多个Redis实例。

哨兵作用

• 集群监控：负责监控redis master和slave进程是否正常工作
• 消息通知：如果某个redis实例有故障，那么哨兵负责发送消息作为报警通知给管理员
• 故障转移：如果master node挂掉了，会自动转移到slave node上
• 配置中心：如果故障转移发生了，通知client客户端新的master地址

哨兵监控环境搭建

新建sentinel-26379.conf文件

#端口
port 26379
#守护进程运行
daemonize yes
#日志文件
logfile "26379.log"
sentinel monitor mymaster 192.168.92.128 6379 2
1
2
3
4
5
6
7

新建sentinel-26380.conf文件

#端口
port 26380
#守护进程运行
daemonize yes
#日志文件
logfile "26380.log"
sentinel monitor mymaster 192.168.92.128 6379 2
1
2
3
4
5
6
7

新建sentinel-26381.conf文件

#端口
port 26381
#守护进程运行
daemonize yes
#日志文件
logfile "26381.log"
sentinel monitor mymaster 192.168.92.128 6379 2
1
2
3
4
5
6
7

sentinel monitor mymaster 192.168.92.128 6379 2
配置的含义是：该哨兵节点监控192.168.92.128:6379这个主节点，该主节点的名称是mymaster
最后的2的含义与主节点的故障判定有关：至少需要2个哨兵节点同意，才能判定主节点故障并进行故障转移。

哨兵节点的启动方式

redis-sentinel sentinel-26379.conf
redis-sentinel sentinel-26380.conf
redis-sentinel sentinel-26381.conf
1
2
3

查看哨兵节点状态

[root@localhost src]# ./redis-cli -p 26379
127.0.0.1:26379>
127.0.0.1:26379>
127.0.0.1:26379> info sentinel
# Sentinel
sentinel_masters:1
sentinel_tilt:0
sentinel_running_scripts:0
sentinel_scripts_queue_length:0
sentinel_simulate_failure_flags:0
master0:name=mymaster,status=ok,address=192.168.66.100:6379,slaves=2,sentinels=3
1
2
3
4
5
6
7
8
9
10
11

哨兵工作原理详解

监控阶段

注意：
sentinel(哨兵1)----->向master(主)和slave(从)发起info，拿到全信息。
sentinel(哨兵2)----->向master(主)发起info，就知道已经存在的sentinel(哨兵1)的信息，并且连接slave(从)。
sentinel(哨兵2)----->向sentinel(哨兵1)发起subscribe(订阅)。

通知阶段

sentinel不断的向master和slave发起通知，收集信息。

故障转移阶段

通知阶段sentinel发送的通知没得到master的回应，就会把master标记为SRI_S_DOWN,并且把master的状态发给各个sentinel，其他sentinel听到master挂了，说我不信，我去看看，并把结果共享给各个sentinel，当有一半的sentinel都认为master挂了的时候，就会把master标记为SRI_0_DOWN。

投票方式

自己最先接到哪个sentinel的竞选通知就会把票投给它。
剔除一些情况：
1 不在线的
2 响应慢的
3 与原来master断开时间久的
4 优先级原则

故障转移

故障转移阶段，哨兵和主从节点的配置文件都会被改写

• 哨兵系统中的主从节点，与普通的主从节点并没有什么区别，故障发现和转移是由哨兵来控制和完成的。
• 哨兵节点本质上是redis节点。
• 每个哨兵节点，只需要配置监控主节点，便可以自动发现其他的哨兵节点和从节点。
• 在哨兵节点启动和故障转移阶段，各个节点的配置文件会被重写(config rewrite)。

Redis集群_Cluster模式

Redis有三种集群模式

主从模式
Sentinel模式
Cluster模式

哨兵模式的缺点

• 当master挂掉的时候，sentinel 会选举出来一个 master，选举的时候是没有办法去访问Redis的，会存在访问瞬断的情况；
• 哨兵模式，对外只有master节点可以写，slave节点只能用于读。尽管Redis单节点最多支持10W的QPS，但是在电商大促的时候，写数据的压力全部在master上。
• Redis的单节点内存不能设置过大，若数据过大在主从同步将会很慢；在节点启动的时候，时间特别长；

Cluster模式概述

Redis集群是一个由多个主从节点群组成的分布式服务集群，它具有复制、高可用和分片特性。

Redis集群的优点

• Redis集群有多个master，可以减小访问瞬断问题的影响
• Redis集群有多个master，可以提供更高的并发量
• Redis集群可以分片存储，这样就可以存储更多的数据

Cluster模式搭建

Redis的集群搭建最少需要3个master节点，我们这里搭建3个master，每个下面挂一个slave节点，总共6个Redis节点；

环境准备

第1台机器： 192.168.66.101 8001端口 8002端口
第2台机器： 192.168.66.102 8001端口 8002端口
第3台机器： 192.168.66.103 8001端口 8002端口
1
2
3

创建文件夹

mkdir -p /usr/local/redis/redis-cluster/8001 /usr/local/redis/redis-cluster/8002
1

拷贝配置文件

将redis安装目录下的 redis.conf 文件分别拷贝到8001目录下

cp /usr/local/redis/redis.conf /usr/local/redis/redis-cluster/8001
1

修改redis.conf文件以下内容

port 8001
daemonize yes
pidfile "/var/run/redis_8001.pid"
#指定数据文件存放位置，必须要指定不同的目录位置，不然会丢失数据
dir /usr/local/redis/redis-cluster/8001/
#启动集群模式
cluster-enabled yes
#集群节点信息文件，这里800x最好和port对应上
cluster-config-file nodes-8001.conf
# 节点离线的超时时间
cluster-node-timeout 5000
#去掉bind绑定访问ip信息
#bind 127.0.0.1
#关闭保护模式
protected-mode no
#启动AOF文件
appendonly yes
#如果要设置密码需要增加如下配置：
#设置redis访问密码
#requirepass baizhan
#设置集群节点间访问密码，跟上面一致
#masterauth baizhan
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

文件拷贝到8002文件夹

#将8001修改为8002：
cp /usr/local/redis/rediscluster/8001/redis.conf /usr/local/redis/redis-cluster/8002
# 批量修改字符串 底行命令
:%s/8001/8002/g
1
2
3
4

使用redis-cli创建整个redis集群

./redis-cli --cluster create --cluster-replicas 1 
192.168.52.138:8001 192.168.52.138:8002 
192.168.52.139:8001 192.168.52.139:8002 
192.168.52.140:8001 192.168.52.140:8002
1
2
3
4

参数：
-a：密码
–cluster-replicas 1：表示1个master下挂1个slave； --cluster-replicas 2：表示1个master下挂2个slave。

参数：
create：创建一个集群环境host1:port1 … hostN:portN
call：可以执行redis命令
add-node：将一个节点添加到集群里，第一个参数为新节点的ip:port，第二个参数为集群中
任意一个已经存在的节点的ip:port
del-node：移除一个节点
reshard：重新分片
check：检查集群状态

连接任意一个客户端

/usr/local/redisd/src/redis-cli -a redis-pw -c -h 192.168.66.101 -p 8001
1

参数：
‐a表示服务端密码
‐c表示集群模式
-h指定ip地址
-p表示端口号

cluster info # 查看集群信息
cluster node # 查看集群节点
1
2

Cluster模式原理分析

Redis Cluster将所有数据划分为16384个slots(槽位)，每个节点负责其中一部分槽位。槽位的信息存储于每个节点中。只有master节
点会被分配槽位，slave节点不会分配槽位。

槽位定位算法： k1 = 127001
Cluster 默认会对 key 值使用 crc16 算法进行 hash 得到一个整
数值，然后用这个整数值对 16384 进行取模来得到具体槽位。
HASH_SLOT = CRC16(key) % 16384

可以通过{}来定义组的概念，从而是key中{}内相同内容的键值对放到同一个slot中。

mset k1{test} v1 k2{test} v2 k3{test} v3
1

Redis集群_Java操作Redis集群

Jedis整合Redis

/**
 * 集群操作
 */
@Test
public void clusterTest(){
    Set<HostAndPort> redisNodes = new HashSet<>();
    redisNodes.add(new HostAndPort("192.168.52.138", 8001));
    redisNodes.add(new HostAndPort("192.168.52.138", 8002));
    redisNodes.add(new HostAndPort("192.168.52.139", 8001));
    redisNodes.add(new HostAndPort("192.168.52.139", 8002));
    redisNodes.add(new HostAndPort("192.168.52.140", 8001));
    redisNodes.add(new HostAndPort("192.168.52.140", 8002));
    // 构建JedisCluster 建立连接
    JedisCluster jedisCluster = new JedisCluster(redisNodes);
    // 添加元素
    jedisCluster.set("name", "zhangsan");
    String name = jedisCluster.get("name");
    System.out.println(name);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

SpringBoot 整合 Redis

只需要修改配置文件就可以了

##单服务器
spring.redis.cluster.nodes=192.168.159.129:7001,192.168.159.129:7002,192.168.159.129:7003,192.168.159.129:7004,192.168.159.129:7005,192.168.159.129:7006
## 连接池最大连接数（使用负值表示没有限制） 
spring.redis.pool.max-active=300
## Redis数据库索引(默认为0) 
spring.redis.database=0
## 连接池最大阻塞等待时间（使用负值表示没有限制） 
spring.redis.pool.max-wait=-1
## 连接池中的最大空闲连接 
spring.redis.pool.max-idle=100
## 连接池中的最小空闲连接 
spring.redis.pool.min-idle=20
## 连接超时时间（毫秒） 
spring.redis.timeout=60000
1
2
3
4
5
6
7
8
9
10
11
12
13
14

Redis企业级解决方案_缓存预热

缓存冷启动

缓存中没有数据，由于缓存冷启动一点数据都没有，如果直接就对外提供服务了，那么并发量上来Mysql就裸奔挂掉了。

新启动的系统没有任何缓存数据，在缓存重建数据的过程中，系统性能和数据库负载都不太好，所以最好是在系统上线之前就把要缓存的热点数据加载到缓存中，这种缓存预加载手段就是缓存预热。

解决思路

• 提前给redis中灌入部分数据，再提供服务
• 如果数据量非常大，就不可能将所有数据都写入redis，因为数据量太大了，第一是因为耗费的时间太长了，第二根本redis容纳不下所有的数据
• 需要根据当天的具体访问情况，实时统计出访问频率较高的热数据
• 然后将访问频率较高的热数据写入redis中，肯定是热数据也比较多，我们也得多个服务并行读取数据去写，并行的分布式的缓存预热
  

Redis企业级解决方案_缓存穿透

缓存穿透是指缓存和数据库中都没有的数据，而用户不断发起请求，如发起为id为“-1”的数据或id为特别大不存在的数据。这时的用户很可能是攻击者，攻击会导致数据库压力过大。

缓存穿透是指用户查询数据，在数据库没有，自然在缓存中也不会有。这样就导致用户查询的时候，在缓存中找不到，每次都要去数据库再查询一遍，然后返回空（相当于进行了两次无用的查询）。这样请求就绕过缓存直接查数据库，这也是经常提的缓存命中率问题。

解决方案

1. 对空值缓存：如果一个查询返回的数据为空（不管数据是否存在），我们仍然把这个空结果缓存，设置空结果的过期时间会很短，最长不超过5分钟。

2. 布隆过滤器：如果想判断一个元素是不是在一个集合里，一般想到的是将集合中所有元素保存起来，然后通过比较确定。

布隆过滤器

布隆过滤器是一种数据结构，比较巧妙的概率型数据结构（probabilistic data structure），特点是高效地插入和查询，可以用来告诉你 “某样东西一定不存在或者可能存在”。

注意：
布隆说不存在一定不存在，布隆说存在你要小心了，它有可能不存在。

代码实现

引入hutool包

<dependency>
   <groupId>cn.hutool</groupId>
   <artifactId>hutool-all</artifactId>
   <version>5.7.17</version>
</dependency>
1
2
3
4
5

java代码实现

/**
 * 测试布隆过滤器
 */
@Test
public void buLongFileter(){
    // 初始化布隆过滤器 构造方法的参数大小10 决定了布隆过滤器BitMap的大小
    BitMapBloomFilter bitMapBloomFilter = new BitMapBloomFilter(10);
    // 添加元素
    bitMapBloomFilter.add("123");
    bitMapBloomFilter.add("abc");
    bitMapBloomFilter.add("ert");
    bitMapBloomFilter.add("weq");
    bitMapBloomFilter.add("www");
    System.out.println(bitMapBloomFilter.contains("abc"));
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

Redis企业级解决方案_缓存击穿

某一个热点 key，在缓存过期的一瞬间，同时有大量的请求打进来，由于此时缓存过期了，所以请求最终都会走到数据库，造成瞬时数据库请求量大、压力骤增，甚至可能打垮数据库。

解决方案

1. 互斥锁：在并发的多个请求中，只有第一个请求线程能拿到锁并执行数据库查询操作，其他的线程拿不到锁就阻塞等着，等到第一个线程将数据写入缓存后，其他线程直接查询缓存。
2. 热点数据不过期：直接将缓存设置为不过期，然后由定时任务去异步加载数据，更新缓存。

	/**
     * 互斥锁解决缓存击穿
     * @param key 商品key
     */
    @Test
    public void lock(String key) throws InterruptedException {

        // 获取key的值
        String value = jedis.get(key);
        if (value == null){
            // 设置3分钟超时。只有key不存在的时候才能创建
            Long setnx = jedis.setnx(key + "_mutex", "1");
            // 设置过期时间
            jedis.expire(key + "_mutex", 180);
            // 设置成功
            if (setnx == 1){
                // TODO 数据库操作
                value = "db";
                // 保存缓存
                jedis.setex(key,180,value);
                jedis.del(key + "_mutex");
            }else {
                Thread.sleep(5000);
                lock(key);
            }
        }
    }
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
26
27

Redis企业级解决方案_缓存雪崩

缓存雪崩是指在我们设置缓存时采用了相同的过期时间，导致缓存在某一时刻同时失效，请求全部转发到DB，DB瞬时压力过重雪崩。

解决方案

过期时间打散： 既然是大量缓存集中失效，那最容易想到就是让他们不集中生效。可以给缓存的过期时间时加上一个随机值时间，使得每个 key 的过期时间分布开来，不会集中在同一时刻失效。

热点数据不过期： 该方式和缓存击穿一样，也是要着重考虑刷新的时间间隔和数据异常如何处理的情况。

加互斥锁: 该方式和缓存击穿一样，按 key 维度加锁，对于同一个 key，只允许一个线程去计算，其他线程原地阻塞等待第一个线程的计算结果，然后直接走缓存即可。

加锁排队代码

/**
 * 通过加锁的方式解决缓存雪崩
 * @param key 商品key
 */
@Test
public Object saveLock(String key){

    // 加锁的key
    String lockKey =key;

    // 获取key的值
    String value = jedis.get(key);
    // 判断是否过期
    if (value == null){
        synchronized (lockKey){
            // 获取key的value值
            String s = jedis.get(key);
            if (s!=null){
                return s;
            }else {
                // TODO 数据库操作
                 value = "db";
                jedis.set(key,"db");
            }
        }
    }
    return value;
}
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
26
27
28

Redis企业级解决方案_Redis开发规范

key设计技巧

1、把表名转换为key前缀，如 tag:
2、把第二段放置用于区分key的字段，对应msyql中主键的列名，如 user_id
3、第三段放置主键值，如 2,3,4
4、第四段写存储的列名

示例

#   表名 主键 主键值 存储列名字
set user:user_id:1:name baizhan
set user:user_id:1:age 20
#查询这个用户
keys user:user_id:9*
1
2
3
4
5

value设计
拒绝bigkey
防止网卡流量、慢查询，string类型控制在10KB以内，hash、list、set、zset元素个数不要超过5000。

命令使用

1、禁用命令
禁止线上使用keys、flushall、flushdb等，通过redis的rename机制禁掉命令，或者使用scan的方式渐进式处理。
2、合理使用select
redis的多数据库较弱，使用数字进行区分，很多客户端支持较差，同时多业务用多数据库实际还是单线程处理，会有干扰。
3、使用批量操作提高效率

• 原生命令：例如mget、mset。
• 非原生命令：可以使用pipeline提高效率。
  4、不建议过多使用Redis事务功能
  Redis的事务功能较弱(不支持回滚)，而且集群版本(自研和官方)要求一次事务操作的key必须在一个slot上。

客户端使用

1. Jedis ：https://github.com/xetorthio/jedis 重点推荐
2. Spring Data redis ：https://github.com/spring-projects/spring-data-redis 使用Spring框架时推荐
3. Redisson ：https://github.com/mrniko/redisson 分布式锁、阻塞队列的时重点推荐

1、避免多个应用使用一个Redis实例
不相干的业务拆分，公共数据做服务化。
2、使用连接池

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

Redis企业级解决方案_数据一致性

缓存已经在项目中被广泛使用，在读取缓存方面，大家没啥疑问，都是按照下图的流程来进行业务操作。

从理论上来说，给缓存设置过期时间，是保证最终一致性的解决方案。

三种更新策略

1 先更新数据库，再更新缓存
2 先删除缓存，再更新数据库
3 先更新数据库，再删除缓存

先更新数据库，再更新缓存

线程安全角度
同时有请求A和请求B进行更新操作，那么会出现
（1）线程A更新了数据库
（2）线程B更新了数据库
（3）线程B更新了缓存
（4）线程A更新了缓存
这就出现请求A更新缓存应该比请求B更新缓存早才对，但是因为网络等原因，B却比A更早更新了缓存。这就导致了脏数据，因此不考虑。

先删缓存，再更新数据库

该方案会导致不一致的原因是。同时有一个请求A进行更新操作，另一个请求B进行查询操作。那么会出现如下情形:
（1）请求A进行写操作，删除缓存
（2）请求B查询发现缓存不存在
（3）请求B去数据库查询得到旧值
（4）请求B将旧值写入缓存
（5）请求A将新值写入数据库

注意：
该数据永远都是脏数据。

先更新数据库，再延时删缓存

这种情况存在并发问题吗？
（1）缓存刚好失效
（2）请求A查询数据库，得一个旧值
（3）请求B将新值写入数据库
（4）请求B删除缓存
（5）请求A将查到的旧值写入缓存