Libz's Blog

# 安装#

大致步骤 :

1. 安装依赖 gcc
2. 上传安装包 , 并解压
3. 进入 redis 根目录 , 运行编译
4. 检查成功
5. 启动

官方下载 : https://redis.io/download/

安装依赖

Redis 是基于 C 语言运行 , 因此依赖 gcc

1
yum install -y gcc tcl

上传并解压

1
# 上传位置自选
2
/usr/local/src
3
# 解压 (注意自己的版本
4
tar -xzf redis-6.2.6.tar.gz

编译运行

1
# 进入目录
2
cd redis-6.2.6
3
# 编译运行
4
make && make install

检查成功

1
# 查看环境变量 (看见reids相关配置即可)
2
ll /usr/local/bin/

该目录以及默认配置到环境变量 , 配置环境后 , 可在任意路径执行命令 .

# 启动#

启动方式有多种 :

1. 默认启动 (不建议)
2. 指定配置启动
3. 开机自启

默认启动

通过命令 直接启动 Redis:

1
redis-server

提示

该启动方式 , 会阻塞会话窗口 , 需要手动关闭窗口

指定配置启动

通过指定配置文件进行启动 Redids , 通过 Reids 根路径的reids.conf配置文件进行操作

配置启动 点击展开

1. 进入 Reids 根路径

2. 拷贝备份(防止误操作)

   1
   cp redis.conf redis.conf.bck
   

3. vim 进入配置

   1
   # 进入
   2
   vim reids.conf
   

   配置内容 , 使用 Vim 查询 更改/添加内容

   1
   # 允许访问的地址，默认是127.0.0.1，会导致只能在本地访问。修改为0.0.0.0则可以在任意IP访问，生产环境不要设置为0.0.0.0
   2
   bind 0.0.0.0
   3
   # 守护进程，修改为yes后即可后台运行
   4
   daemonize yes
   5
   # 密码，设置后访问Redis必须输入密码
   6
   requirepass 123123
   

4. 启动 , 进入文件根目录 指定文件运行 / 全限定名路径也可以

   1
   redis-server redis.conf
   

5. 关闭 Redis 服务

   1
   # 因为之前配置了密码，因此需要通过 -user 来指定密码
   2
   redis-cli --user 123123 shutdown
   

开机启动

默认情况是没有systemctl命令 , 因此我们需要手动配置

开机启动 点击展开

1. 创建系统文件

   1
   vim /etc/systemd/system/redis.service
   

2. 配置以下内容 (注意自己的安装路径 , 启动配置文件的路径)

   1
   [Unit]
   2
   Description=redis-server
   3
   After=network.target
   4
   
   5
   [Service]
   6
   Type=forking
   7
   ExecStart=/usr/local/bin/redis-server /usr/local/src/redis-6.2.6/redis.conf
   8
   PrivateTmp=true
   9
   
   10
   [Install]
   11
   WantedBy=multi-user.target
   

3. 通过 systemctl进程 控制

   1
   # 启动
   2
   systemctl start redis
   3
   # 停止
   4
   systemctl stop redis
   5
   # 重启
   6
   systemctl restart redis
   7
   # 查看状态
   8
   systemctl status redis
   

4. 设置开机启动

   1
   systemctl enable redis
   

注意

云服务器 , 需要开启安全组连接端口 : 6379

虚拟机 , 需要 防火墙开端口6379/关闭防火墙

# 终端连接#

Redis 连接通过 redis-cli 命令 :

1
# reids-cli [options] [commonds]
2
reids-cli -h 127.0.0.1 -p 6379

进入 Redis 命令控制台后 , 需要登录

1
# AUTH [username] password
2
# 无账号密码登录
3
AUTH 123123

# 图形化工具#

QuickReids : https://quick123.net/ (opens new window)

开箱即用

# 数据类型#

Redis 支持五种数据类型：

• string(字符串)
• hash(哈希)
• list(列表)
• set(集合)
• zset/sortedSet(有序集合)

偏移剪切
查询、删除 指定节点 | | Set(集合) | 集合中的每个元素都是一个字符串，且他们都是唯一的 | 可 增删查 元素，
检测元素是否存在集合
计算集合 交、并、差集 等
随机读取元素 | | SortedSet(集合) | 可排序 Set 集合 每个元素都携带 score属性 | | | Hash(哈希散列表) | Java 中的 Map类 ，<K , V> | 可 增删改查 键值对
可获取所有键值对 | | Zset(有序集合) | 有序集合，每个元素都携带score属性 , 元素基于该属性排序 | 可 增删改查 元素
根据分值范围或成员 获取对应元素 | | HyperLogLog(基数) | 计算重复的值，确定存储数量 | 只提供基数运算，不提供返回功能 |

# 常用命令#

Redis 命令用于在 redis 服务上执行操作

命令参考 :

• https://redis.io/commands (opens new window)
• http://doc.redisfans.com/index.html (opens new window)

基本类型操作

HyoperLogLog 命令

随机化的算法，以少量内存提供集合唯一的元素数量的近似值

可接受多个元素作为输入，并给出输入元素的基数估算值

注意

基数：集合中不同元素的数量。 例如{‘Sanscan’,‘Bobo’,‘Sanscan’,‘Tomy’,‘Sanscan’}的基数为 3

估算值：算法给出的基数并非精确，有些许偏差，但 会控制在范围内

# Jedis#

Jedis 是用于 Reids 命令的 Java 客户端库 , 命令通常以方法名使用 . 使用简单快速 , 含有线程安全问题

GitHub : https://github.com/redis/jedis (opens new window)

大致步骤 :

1. 引入依赖
2. 实例 jedis 对象(连接)
3. 使用 jedis(操作)
4. 释放资源

依赖

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

连接服务器 连接可直接通过 实例Jedis类即可食用

1
Jedis jedis=new Jedis("192.168.197.129",6379);
2
// 如果 Redis 服务设置了密码，需要下面这行，没有就不需要
3
// jedis.auth("123456");
4
jedis.set("java001","java工程师");
5
String java001 = jedis.get("java001");
6
System.out.println(java001);
7
if(jedis != null){
8
    jedis.close();
9
}

提示

指令和方法名称类似 , 可以根据命令传参即可使用

1
public class ConnectionTest {
2
    public static void main(String[] args) {
3
        JedisPoolConfig config = new JedisPoolConfig();
4
        //最大连接数
5
        config.setMaxTotal(30);
6
        //最大空闲数
7
        config.setMaxIdle(10);
8
        //获取连接池
9
        JedisPool jedisPool = new JedisPool(config,"192.168.74.131",6379);
10
        Jedis jedis = null;
11

12
        try {
13
            jedis = jedisPool.getResource();
14
            jedis.set("name","张三");
15
            String name = jedis.get("name");
16
            System.out.println("name : " + name);
17
        } catch (Exception e) {
18
            e.printStackTrace();
19
        }finally {
20
            if (jedis != null) {
21
                jedis.close();
22
            }
23
            if (jedisPool != null) {
24
                jedisPool.close();
25
            }
26
        }
27

28
    }
29
}

1
public class Demo {
2
    public static void main(String[] args) {
3

4
        // 创建连接
5
        Set<HostAndPort> nodes = new HashSet<HostAndPort>();
6
        nodes.add(new HostAndPort("192.168.74.131",7001));
7
        nodes.add(new HostAndPort("192.168.74.131",7002));
8
        nodes.add(new HostAndPort("192.168.74.131",7003));
9
        nodes.add(new HostAndPort("192.168.74.131",7004));
10
        nodes.add(new HostAndPort("192.168.74.131",7005));
11
        nodes.add(new HostAndPort("192.168.74.131",7006));
12

13
        // 集群搭建
14
        JedisCluster cluster = null;
15
        cluster = new JedisCluster(nodes);
16

17
        // 执行JedisCluster对象中的方法，方法和redis指令一一对应
18
        cluster.set("name","柏竹");
19
        String name = cluster.get("name");
20
        System.out.println("name : " + name);
21

22
        //存储List数据到列表中
23
        cluster.lpush("site-list", "java");
24
        cluster.lpush("site-list", "c");
25
        cluster.lpush("site-list", "mysql");
26
        List<String> stringList = cluster.lrange("site-list",0,2);
27
        System.out.println("=============");
28
        for (String s : stringList) {
29
            System.out.println(s);
30
        }
31

32
        //关闭集群 JedisCluster对象
33
        try {
34
            if (cluster != null) {
35
                cluster.close();
36
            }
37
        } catch (IOException e) {
38
            e.printStackTrace();
39
        }finally {
40
            System.out.println("集群测试完成！！！");
41
        }
42

43
    }
44
}

# SpringDataRedis#

SpringData 是 Spring 中数据操作的模块 , 该模块集成了很多数据库操作 , 其中也包括 Reids

官方 : https://spring.io/projects/spring-data-redis/ (opens new window)

优点 :

• 整合了 Lettuce 和 jedis
• 提供 ReidsTemplate 统一 API 操作
• 支持 同步/异步/响应式编程
• 线程安全
• 支持 哨兵模式/集群/管道 模式
• 支持 发布订阅模型
• 支持基于 JDK/JSON/字符串/Spring 对象的 序列化/反序列化

SpringDataRedis 提供 RedisTemplate 工具类 , 里面封装了各种 Reids 操作功能 , 分别介绍 :

大致步骤 :

1. 引入依赖
2. 配置参数
3. 配置类配置
4. CRUD 测试

代码示例 点击展开

引入依赖

1
<dependency>
2
    <groupId>org.springframework.boot</groupId>
3
    <artifactId>spring-boot-starter-data-redis</artifactId>
4
</dependency>

配置 Reids 基本参数

1
# redis 配置 (端口/地址/密码/连接池配置)
2
spring:
3
  data:
4
    redis:
5
      port: 6379
6
      host: localhost
7
      password: 123123
8
      lettuce:
9
        pool:
10
          max-active: 8
11
          max-idle: 8
12
          min-idle: 0
13
          max-wait: 100ms

我使用的是 SprinBoot3.0.4 版本 , 旧版本没有 data 节点

配置类 配置

1
@Configuration
2
public class RidesConfig {
3

4
    /**
5
     * 自定义配置 RedisTemplate
6
     * @param connectionFactory 连接工厂
7
     * @return
8
     */
9
    @Bean
10
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
11
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
12
        // 设置 key序列化器 RedisSerializer
13
        redisTemplate.setKeySerializer(RedisSerializer.string());
14
        redisTemplate.setHashKeySerializer(RedisSerializer.string());
15
        // 设置 连接工厂
16
        redisTemplate.setConnectionFactory(connectionFactory);
17
        return redisTemplate;
18
    }
19
}

测试 CRUD

1
@Resource
2
private RedisTemplate redisTemplate;
3

4
@Test
5
public void add() {
6
    ValueOperations ops = redisTemplate.opsForValue();
7

8
    ops.set("String","Sans");
9
    String[] stat = {"zs1","zs2"};
10
    ops.set("List",Arrays.toString(stat));
11
    ops.set("Int",666);
12
    ops.set("double",6.6);
13
    User user = new User();
14
    user.setId(9);
15
    user.setUsername("Sans111");
16
    ops.set("user",user);
17
}
18

19
@Test
20
public void show() {
21
    ValueOperations ops = redisTemplate.opsForValue();
22
    System.out.println("String =>" + ops.get("String"));
23
    System.out.println("List =>" + ops.get("List"));
24
    System.out.println("Int =>" + ops.get("Int"));
25
    System.out.println("double =>" + ops.get("double"));
26
    System.out.println("user =>" + ops.get("user"));
27
}
28

29
@Test
30
public void del() {
31
    redisTemplate.delete("String");
32
    redisTemplate.delete("List");
33
    redisTemplate.delete("Int");
34
    redisTemplate.delete("double");
35
    redisTemplate.delete("user");
36
}

提示

SpringBoot3.02 版本以下 , 采用 @Resource注解 自动注入 , 需要添加 name 参数进行指定名称 Bean 的方法名称 , 例如 :

1
@Resource(name="redisTemplate")
2
private RedisTemplate redisTemplate;

1
@Configuration
2
public class RidesConfig {
3
    /**
4
     * 自定义配置 RedisTemplate
5
     * @param connectionFactory 连接工厂
6
     * @return
7
     */
8
    @Bean
9
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
10
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
11
        // 设置 key序列化器 RedisSerializer
12
        redisTemplate.setKeySerializer(RedisSerializer.string());
13
        redisTemplate.setHashKeySerializer(RedisSerializer.string());
14

15
        // 设置 连接工厂
16
        redisTemplate.setConnectionFactory(connectionFactory);
17
        return redisTemplate;
18
    }
19
}

# Redisson#

Redisson 是基于 Reids 实现的分布式 , 可伸缩 Java 数据结构集合等类型对象

介绍 : https://github.com/redisson/redisson (opens new window)

版本 : https://github.com/redisson/redisson/tree/master/redisson-spring-boot-starter (opens new window)

高级 Sedisson 应用 : 传送门跳转

快速入门

快速应用 点击展开

1. 引入依赖

   1
   <dependency>
   2
       <groupId>org.springframework.session</groupId>
   3
       <artifactId>spring-session-data-redis</artifactId>
   4
   </dependency>
   

   版本迭代快 , 非 SprinBoot 谨慎选择 Redisson 版本

2. 写入配置

   1
   @Bean
   2
   public RedissonClient redissonClient() {
   3
       // 创建配置
   4
       Config config = new Config();
   5
       String redisAddress = String.format("redis:127.0.0.1:6379");
   6
       config.useSingleServer()
   7
         .setAddress(redisAddress)
   8
         .setDatabase(0);
   9
   
   10
       // 创建实例
   11
       RedissonClient redissonClient = Redisson.create(config);
   12
       return redissonClient;
   13
   }
   

3. 测试应用

   1
   @Resource
   2
   private RedissonClient redissonClient;@Test
   3
   public void redissonTest() {
   4
       // JVM 本地操作
   5
       List<Object> list = new ArrayList<>();
   6
       list.add("sasn");
   7
       System.out.println("list.get(0) = " + list.get(0));
   8
   
   9
       // reids 操作
   10
       // RLIST 继承了 List特性
   11
       RList<Object> rList = redissonClient.getList("test-list");
   12
       rList.add("123123");
   13
       System.out.println("rList.get(0) = " + rList.get(0));
   14
   
   15
       // Redisson 其他集合...
   16
       //redissonClient.getMap("test-map");
   17
   }
   

1
@Resource
2
private RedissonClient redissonClient;
3

4
@Scheduled(cron = "0 0 0 * * *")
5
public void doCacheRecommendUser() {
6
    RLock lock = redissonClient.getLock("sans:precachejob:docache:lock");
7
    /*
8
      参数
9
      1. 等待获取(0无需等待)
10
      2. 过期时长
11
      3. 时间单位
12
     */
13
    try {
14
        // 只有一个线程获取到锁
15
        if (lock.tryLock(0, 30000, TimeUnit.MILLISECONDS)) {
16
            System.out.println("lockName op :" + Thread.currentThread().getName());
17
            for (Long userId : mainUserList) {
18
                String redisKey = String.format("sans:user:recommend:%s", userId);
19
                ValueOperations<String, Object> ops = redisTemplate.opsForValue();
20
                QueryWrapper<User> qw = new QueryWrapper<>();
21
                Page<User> userPage = userService.page(new Page<>(1, 20), qw);
22
                try {
23
                    ops.set(redisKey, userPage, 1, TimeUnit.DAYS);
24
                } catch (Exception e) {
25
                    log.error("redis set key");
26
                }
27
            }
28
        }
29
    } catch (InterruptedException e) {
30
        log.error("doCacheRecommendUser error " + e.getMessage());
31
    } finally {
32
        if (lock.isHeldByCurrentThread()) {
33
            lock.unlock();
34
            System.out.println("lockName ed :" + Thread.currentThread().getName());
35
        }
36
    }
37
}

1
@Autowired
2
private RedissonClient redisson; //自动装配RedissonClient
3
RLock lock = redisson.getLock("onelock"); //获取锁
4
lock.lock(); //加锁
5
lock.unlock(); //释放锁

1
/**
2
 * 看门狗机制测试
3
 */
4
@Test
5
public void redissonLookDoorDog() {
6
    RLock lock = redissonClient.getLock("sans:precachejob:docache:lock");
7
    System.out.println("start");
8
    try {
9
        if (lock.tryLock(0, -1, TimeUnit.MILLISECONDS)) {
10
            Thread.sleep(300000);
11
        }
12
    } catch (InterruptedException e) {
13
        System.out.println(e.getMessage());
14
    } finally {
15
        // 只能释放本身线程的锁(以防释放其他线程的锁)
16
        if (lock.isHeldByCurrentThread()) lock.unlock();
17
    }
18
}

# 数据结构#

# 缓存淘汰#

Redis 的缓存数据是基于内存存储的 , 内存终究会有不够用的时候 , 为了保证内存不会爆满 , 导致宕机 , 因此 Reids 的缓存淘汰机制就起到关键作用

淘汰分类 :

对 缓存与数据库 一致性要求高 , 可采取以下方案 :

读数据 :

• 缓存查到直接返回
• 缓存查不到 , 写入缓存 , 并设超时时间

写数据 :

• 先写数据库 , 后删除缓存
• 确保数据库和缓存操作是过程无异常(事务)

# 事务#

Redis 事务可以一次执行多个命令， 并且带有以下两个重要的保证：

• 事务是一个单独的隔离操作：事务中的所有命令都会序列化、按顺序地执行。事务在执行的过程中，不会被其他客 户端发送来的命令请求所打断
• 事务中的命令要么全部被执行，要么全部都不执行

事务阶段：

1. 开始事务
2. 命令入队
3. 执行事务

事务命令

# 发布订阅#

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

但发布者发布消息发送到信道 , 当有消费者订阅信道(频道)channel , 将会接收到发布的相关消息

特点 :

• 可实现广播形式发布消息
• 不能对消息不支持持久化
• 消息堆积有上限 , 超出会丢失

命令 :

pattern 参数

字符串 通配符应用 , 支持有以下通配符 :

• ? : 统配 一个 字符
• * : 通配 一个/多个 字符
• [] : 通配 多个常量 . 例如 : h[ae]llo -> hello / hallo 识别两种!

测试：

1. 打开 3 个客户端 , 分别 1 个发布者 , 2 个订阅者
2. 订阅者 1 : 订阅 SUBSCRIBE order.q1
3. 订阅者 2 : 订阅 PSUBSCRIBE order.q?
4. 发布者发布 :
   • SUBSCRIBE order.q1 msg1
   • SUBSCRIBE order.q2 msg2
5. 分别查看他们接收情况

# 消息队列#

Redis5.0 引入了新数据类型 Stream , 用于实现消息队列

了解即可 , 消息队列还得要学 RabbitMQ ~

特点 :

• 消息可回溯 (不会出现读完删除)
• 一个消息可多消费者读取
• 可阻塞读取
• 消息可能会漏读

命令 :

BLOCK milliseconds : 等待时长 ms(0 永久阻塞)
key : 指定队列名称
ID : 起始 id 开始读取 (0 第一个开始 ; $最新消息开始) | 读 消息队列 |

读取方式 :

• 阻塞等待读取最新一条消息 : XREAD BLOCK 0 STREAMS key $
• 读取所有消息(数值越大读越多) : XREAD COUNT 99 STREAMS key 0

1
# 创建 , 创建消费者组 . ID : 起始id开始读取 (0第一个开始 ; $最新消息开始)
2
XGROUP CREATE key groupName ID [MKSTREAM]
3
# 删除 , 指定 消费者组
4
XGROUP DESTORY key groupName
5
# 添加 , 指定 消费者组 添加 消费者
6
XGROUP CREATECONSUMER key groupname consumername
7
# 删除 , 指定 消费者组 删除指定 消费者
8
XGROUP DELCONSUMER key groupname consumername
9
# 读取 , 指定 消费者组中的消费者 的数据(ID可以为 '>' 下一个未消费的消息开始)
10
XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] STREAMS key ID

# 持久化#

Redis 值放在内存中 , 为防止突然断电等特殊情况的发生 , 需要对数据进行持久化备份 . 即将内存数据保存到硬盘

Reids 有两种持久化 :

• RDB 持久化
• AOF 持久化

1
# dbfilename：RDB 持久化的文件名称
2

3
dbfilename dump.rdb
4

5
# dir：文件保存的路径(reids 根目录)
6

7
dir ./
8

9
# 快照触发 save 机制
10

11
# 900s 内 更变操作 1 : 触发 save
12

13
# 300s 内 更变操作 10 : 触发 save
14

15
# 60s 内 更变操作 10000 : 触发 save
16

17
save 900 1
18
save 300 10
19
save 60 10000
20

21
# 当 save 中途出现异常时 , 是否阻塞客户端 更变操作 (可能因为磁盘 满了/故障 导致异常)
22

23
stop-writes-on-bgsave-error yes
24

25
# rdbcompression : 是否压缩备份文件(会损耗 CPU,不建议启动)
26

27
rdbcompression no

1
# appendonly : 启动 AOF 功能 (默认 no)
2

3
appendonly yes
4

5
# appendfilename : 指定 aof 文件名称
6

7
appendfilename appendonly.aof
8

9
# appendfsync : 指定 aof 操作中文件同步策略
10

11
# 三个参数：always/everysec/no (默认 everysec)
12

13
appendfsync everysec
14

15
# todo 在 aof-rewrite 期间，appendfsync 是否暂缓文件同步，"no"表示“不暂缓”，“yes”表示“暂缓”，默认为“no”
16

17
no-appendfsync-on-rewrite no
18

19
# AOF 触发重写 : 超过指定容量 单位 mg,gb(默认 64mb;建议 512mb)
20

21
auto-aof-rewrite-min-size 64mb
22

23
# AOF 触发重写 : 比上次重写超过指定百分比
24

25
# 每次触发 AOF 记录 , 都会检测大小 , 从而进行判断下次重写时机
26

27
auto-aof-rewrite-percentage 100

# 主从机制#

Redis 是指 Reids 主服务器 , 将数据拷贝到其他 Reids 从服务器上 , 从而搭建成主从集群 , 实现读写分离

为了更好理解

• 主服务器 master
• 从服务器 slave/replication

主从复制作用 :

• 故障恢复 : master 发生故障，由 slave 提供服务，直至 master 修复
• 数据冗余 : 实现数据同步备份，持久化之外的数据备份方式

主从复制原理 :

1. slave 向 master 发送同步请求 , 携带 replid 和 offset , master 也响应 replid 和 offset , slave 更新版本信息 (此时会响应 OK)
2. master 每次触发执行 bgsave 后 , 都会建 RDB 文件 发送给 slave , slave 接收后会清空内存并加载 RDB 文件
3. 发送记录 master 期间的命令 , 通过 repl_baklog 发送到 slave 进行同步指令信息

步骤 1 关键字说明

replid : 数据集标记 , 一般用于判断数据集是否一致/第一次同步

offset : 偏移量 , 随着记录 repl_baklog 越来越大 . 需要通过偏移量控制更新范围

实现 :

1. 启动 Reids 两个以上
2. 连接 , slave 执行连接命令 以下两个其中一个连接命令
   • SLAVEOF host port (5.0 前版本)
   • SRPLICAOF host port (5.0 后版本)
3. 验证 , master 查看状态命令 INFO replication

提示

搭建主从思路可以根据不同 ip/端口 进行搭建多个连接应用

注意

• 一旦连接成功 , 自动同步
• slave 只能读不能写 ; master 能读写
• slave 宕机时间不能够久 , 过久可能会导致数据同步不到位

# 哨兵模式#

哨兵模式(Sentinel) 是 监视多个 Redis 服务器的状态 . Sentinel 可以有多个 组成 Sentinel 系统

作用 :

• 系统监控 : 按指定频率 PING 命令检查 master 和 slave 状态 (默认 1s 一次)
• 故障修复 : 如果 master 故障 , Sentinel 会自选举一个 slave 升为 master , 当故障实例恢复后以新 master 为主(按照权重选举)
• 状态通知 : Sentinel 充当 Reids 实例 , 当集群故障转移 , 会进行一次通知

心跳监控机制

Sentinel 每秒会 PING 检查实例运作状态 , 下线状态有两种可能 : (Sentinel 集群状态下)

• 主观下线 : 某个 Sentinel 检测实例未响应 , 视为 客观下线
• 客观下线 : 多个 Sentinel 检测实例未响应 , 且超过指定 quorun 值 , 则视为 客观下线

建议 : quorun值 最好是 Sentinel 集群 数量的一般以上

选举机制

Sentinel 一旦发现 master 故障 , Sentinel 选举一个 slave 升为 master , 选举依据 :

1. 排除选举 , 会判断原先的 slave 和 master 断开时长记录 , 如果超过指定值 (down - after - milliseconds * 10) , 则视为 老旧实例节点(不靠谱) , 进行排除
2. 判断 slave 的 slave-priority 值(权重) , 越小优先级越高 , 0 则不选举 (默认 100)
3. 如果 slave-priority 值 一致的实例 , 那么判断 offset 值 , 越大数据越新 , 优先级也就越高
4. 判断 slave 运行 id 的大小 , 越小优先级越高

转移机制

Sentinel 选举一个实例作为新的 master 时 , 运作步骤如下 :

1. 选举 master , 对选中的 slave 发送执行命令 SLAVEOF no noe , 称为 新的 master
2. 广播其他 slave 发送执行命令 SLAVEOF host port/SLAVEOF host port (指定 新 master 的 IP 和端口)
3. 旧实例的 master , 会被标记为 slave , 但实例故障恢复后也会执行 步骤 2 的命令

Redis 版本指令更变

• SLAVEOF host port (5.0 前版本)
• SRPLICAOF host port (5.0 后版本)

实现 :

哨兵代码示例 点击展开

假设节点实例

1. 创建目录

   1
   # 进入/tmp目录
   2
   cd /tmp
   3
   # 创建目录
   4
   mkdir s1 s2
   

2. 在 s1 , s2 目录中 , 配置sentinel.conf文件 (要配置 s1 和 s2)

   1
   # 当前 s1 端口
   2
   
   3
   port 20001
   4
   
   5
   # 当前 s1 IP 地址
   6
   
   7
   sentinel announce-ip 192.168.150.101
   8
   
   9
   # 定义主节点 master 名为 mymaster(任意) ; IP ; 端口 ; quorun(掉线判断)
   10
   
   11
   sentinel monitor mymaster 192.168.150.101 7001 2
   12
   
   13
   # slave 与 master 断开的 超时时间
   14
   
   15
   sentinel down-after-milliseconds mymaster 5000
   16
   
   17
   # 实例故障恢复的 超时时间
   18
   
   19
   sentinel failover-timeout mymaster 60000
   20
   
   21
   # 工作目录
   22
   
   23
   dir "/tmp/s1"
   

3. 启动 Sentinel 哨兵模式

   1
   redis-sentinel /tmp/s1/sentinel.conf
   2
   redis-sentinel /tmp/s2/sentinel.conf
   3
   # 也可 Reids中的Sentinel配置运行(不用服务器情况下)
   4
   redis-server redis.conf --sentinel
   

4. 测试 , 保证 s1 , s2 , r1 , r2 , r3 正常运作

   1. 进程查看 ps -aux|grep redis
   2. 进程消灭 r1(master) KILL 9 pid (模拟 master 宕机)
   3. 状态检查 INFO replication , 查看是否转化为 master
   4. 日志检查 Reids 根据的 reids.log 文件可以查阅(可得知变化的状态)

# 集群#

Redis 集群是由一个以上的多个实例节点组成的分布式服务器 , 解决了 高并发/高可用/稳定高 问题

优点 :

• 将数据 自动切分(slot) 到多个节点的能力
• 部分实例节点宕机 , 仍然保持通信应用 , 无需哨兵 , 自带主从切换

redis-cluste 集群方案

Redis-Cluster 采用无中心结构 , 每个节点保存数据和整个集群状态 , 每个节点都和其他所有节点连接

结构/特点 说明

1. 所有 redis 节点都是彼此互联 , 且每个 Reids 都是 master . 内部使用二进制协议优化传输速率
2. 所有 reids 之间通过 PING 命令 检测彼此健康状态
3. client 与 redis 节点是直连的 , 无需中间层 , client 只需连接任意一个 redis 节点即可应用
4. 当 Redis 集群 任意 master 宕机 , 且当前 master 没有 slave , 则 Redis-Cluster 为 fail 状态 (但 slot 映射 不完全进入 fail 状态)
5. 当 Redis-Cluster 中 master 挂掉一半以上 , 则 Redis-Cluster 为 fail 状态

1
# 先查看节点 id/插槽量 , 记录他们的 集群节点的id
2
redis-cli -p 7001 cluster nodes
3
# 将 7007 作为 7001 的 slave
4
redis-cli --cluster --cluster-slave --cluster-master-id {7001节点Id} 192.168.150.101:7007 192.168.150.101:7001

# 分布式锁#

分布式锁是多个服务器在同一系统中可服务器多个进程对资源的访问

Redis 为单进程 单线程模式 , 采用队列模式将并发访问变成串行访问 , 且多客户端对 Redis 的连接并不存在竞争 关系。redis 也可实现分布式锁

点击了解线程同步 (opens new window)

意图 : 节省资源空间

应用满足条件

1. 系统是一个分布式的系统
2. 资源共享 (各个系统访问同一数据库)
3. 同步访问 (多个进程同时访问同一个资源)

redis 分布式锁命令

SETNX SETNX key value key 存在 , 不做操作 ; key 不存在则设值

GETSET GETSET key value 先获取 key 对应的旧值 , 且新值覆盖替换旧值

注意

• 用完锁一定要释放
• 锁一定要加过期时间

# 多级缓存#

多级缓存架构图 :

多层缓存 , 分别说明缓存层级 : (可以根据情况优化)

1. OpenResty Nginx 字典 本地缓存
2. Redis 缓存
3. JVM 缓存 (集群需要依赖 负载均衡的 Hash 分配策略)

笔记

以上实现目的是为了突破 Tomcat 接收压力瓶颈问题 , 从而选举优化方案!

1
<dependency>
2
    <groupId>com.github.ben-manes.caffeine</groupId>
3
    <artifactId>caffeine</artifactId>
4
</dependency>

1
@Test
2
void testBasicOps() {
3
    // 创建缓存对象
4
    Cache<String, String> cache = Caffeine.newBuilder().build();
5

6
    // 存数据
7
    cache.put("name", "Sans");
8

9
    // 取数据，不存在则返回null
10
    String name = cache.getIfPresent("name");
11
    System.out.println("name = " + name);
12

13
    // 取数据，不存在则去数据库查询
14
    String defaultGF = cache.get("name", key -> {
15
        // 这里可以去数据库根据 key查询value
16
        return "张三";
17
    });
18
    System.out.println("defaultGF = " + defaultGF);
19
}

# 可重入锁#

可重入锁 是 在一个业务方法中调用了其他业务 , 且该业务也存在锁 , 可以理解为锁中锁 .

在 Session 中 可重入锁 value 采用 Hash 结构存储 , field(Hash 中的 key)和 value 它们分别是 当前线程标识 和 锁的层级(嵌套有多少层) . 每获取一次锁都会自增 1(锁层级)

提示

获取锁和释放锁 , 不管在那个业务它们都是成对出现的 !

可重入锁代码示例 点击展开

1
@Resource
2
private RedissonClient redissonClient;
3

4
@Test
5
public void business1() {
6
    RLock lock = redissonClient.getLock("lock:"+66);
7
    if (!lock.tryLock()) throw new RuntimeException("确保是自己的锁");
8
    try {
9
        log.info("1. 获取锁 成功");
10
        business2(lock);
11
    } finally {
12
        log.info("1. 释放锁");
13
        lock.unlock();
14
    }
15
}
16

17
public void business2(RLock lock) {
18
    if (!lock.tryLock()) throw new RuntimeException("确保是自己的锁");
19
    try {
20
        log.info("2. 获取锁 成功");
21
    } finally {
22
        log.info("2. 释放锁");
23
        lock.unlock();
24
    }
25
}

PS : 断点观察锁中的 Hash 变化情况~

# 获取锁#

视频学习 : 详细学习 (opens new window)

1
"if (redis.call('exists', KEYS[1]) == 0) then " +
2
    "redis.call('hset', KEYS[1], ARGV[2], 1); " +
3
    "redis.call('pexpire', KEYS[1], ARGV[1]); " +
4
    "return nil; " +
5
"end; " +
6
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
7
    "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
8
    "redis.call('pexpire', KEYS[1], ARGV[1]); " +
9
    "return nil; " +
10
"end; " +
11
"return redis.call('pttl', KEYS[1]);"

lua 脚本

传参就说明到此 , Redis 命令 细品

# 释放锁#

1
"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
2
  "return nil;" +
3
"end; " +
4
"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
5
"if (counter > 0) then " +
6
  "redis.call('pexpire', KEYS[1], ARGV[2]); " +
7
  "return 0; " +
8
"else " +
9
  "redis.call('del', KEYS[1]); " +
10
  "redis.call('publish', KEYS[2], ARGV[1]); " +
11
  "return 1; " +
12
"end; " +
13
"return nil;"

# 分布式锁流程#

# 集群分布式锁#

在企业中为了提高效率 , 一般是已集群 Reids 主从形式搭建为例子

问题 : 我们有 5 台 Reids 服务器 , 一台主(写) 四台从(读取) , 当更改数据主会同步从服务器 , 那么这个同步的过程必有可能会存在延迟 , 如果获取锁时主服务器在同步中途突然宕机了 , 因此我们这个锁可能就失效 , 因此可能还存在隐患 !

解决思路 : 将 2 台设为主服务器 , 3 台设为从服务器 , 以防其中一台主服务器宕机导致锁失效的情况

Java 实现

1
@Configuration
2
public class ReidssonConfig {
3

4
    @Bean
5
    @Primary
6
    public RedissonClient redissonClient1() {
7
        Config config = new Config();
8
        config.useSingleServer()
9
                .setAddress("redis://127.0.0.1:6379");
10
        // 创建 RedissonClient
11
        return Redisson.create(config);
12
    }
13

14
    @Bean
15
    public RedissonClient redissonClient2() {
16
        Config config = new Config();
17
        config.useSingleServer()
18
                .setAddress("redis://127.0.0.1:6380");
19
        // 创建 RedissonClient
20
        return Redisson.create(config);
21
    }
22

23
    // 剩下3台代码一样
24

25
}

测试

1
@RunWith(SpringJUnit4ClassRunner.class)
2
@SpringBootTest
3
@Slf4j
4
public class RedissonTest {
5

6
    @Resource
7
    private RedissonClient redissonClient1;
8
    @Resource
9
    private RedissonClient redissonClient2;
10
    @Resource
11
    private RedissonClient redissonClient3;
12
    @Resource
13
    private RedissonClient redissonClient4;
14
    @Resource
15
    private RedissonClient redissonClient5;
16

17
    RLock lock;
18

19
    @Test
20
    public void business1() {
21
        //lock = redissonClient.getLock("lock:"+66);
22

23
        RLock lock1 = redissonClient1.getLock("lock:"+66);
24
        RLock lock2 = redissonClient2.getLock("lock:"+66);
25
        RLock lock3 = redissonClient3.getLock("lock:"+66);
26
        RLock lock4 = redissonClient4.getLock("lock:"+66);
27
        RLock lock5 = redissonClient5.getLock("lock:"+66);
28

29
        // 连锁
30
        lock = redissonClient.getMultiLock(lock1, lock2, lock3, lock4, lock5);
31

32
        if (!lock.tryLock()) throw new RuntimeException("确保是自己的锁");
33
        try {
34
            log.info("1. 获取锁 成功");
35
            business2();
36
        } finally {
37
            log.info("1. 释放锁");
38
            lock.unlock();
39
        }
40
    }
41

42
    public void business2() {
43
        if (!lock.tryLock()) throw new RuntimeException("确保是自己的锁");
44
        try {
45
            log.info("2. 获取锁 成功");
46
        } finally {
47
            log.info("2. 释放锁");
48
            lock.unlock();
49
        }
50
    }
51
}

加锁原理

Redisson 通过 MutiLock 锁 为多个主服务器发送加锁 , 并且这些服务器有原子性的操作

# 布隆过滤器#

布隆过滤器能够在海量数据中快速判断一个元素是否包含在一个集合中 . 在实际场景中能够减轻 Redis 无效交互的负担

深入了解 : 点击跳转 (opens new window)

优点 :

• 能够在海量数据快速判断是否包含指定元素
• 布隆过滤器存储的空间小 , 存储并非对象本身 , 而是对象的 hash 结果取模作为标记
• 不存储数据本身 , 适合保密

缺点 :

• 不存数据本身 , 删除数据后 , 存在误判率
• hash 碰撞问题 (过滤器匹配存在 , 而缓存中却不存在)
• 容量快慢时 , hash 碰撞率变高

根据以上可以得知 只要过滤器匹配不存在则一定会不存在 , 那么过滤器匹配存在则有可能不存在(概率性) , 只要在使用前配置好参数 , 能在一定程度避免无效的 Redis 交互

以下采用 Redisson 自带有布隆过滤器实现

简单示例

1
// 会在Redis创建 filter为key的缓存
2
public static final RBloomFilter<Object> filter = RedisUtils.getClient().getBloomFilter("filter");
3

4
static {
5
    // 初始化过滤器 , 将配置以及空间在Redis创建
6
    filter.tryInit(1000, 0.01);
7
}
8

9
// 布隆过滤
10
@GetMapping("/t9/get/{id}")
11
public R<String> test09Get(@PathVariable Long id) {
12
    String key = "test9" + id;
13
    // 过滤不存在的数据
14
    if (!filter.contains(key)) return R.fail("不存在");
15

16
    // 介入 Redis查询
17
    Object cacheObject = RedisUtils.getCacheObject(key);
18
    if (ObjectUtil.isNull(cacheObject)) return R.fail("不存在");
19
    return R.ok(cacheObject.toString());
20
}
21

22
@GetMapping("/t9/put/{id}")
23
public R<String> test09Put(@PathVariable Long id) {
24
    String key = "test9" + id;
25
    RedisUtils.setCacheObject(key, id);
26
    filter.add(key);
27
    return R.ok();
28
}