记忆角落

  • {{ item.name }}
  • Redis6-学习笔记
  • Redis概述安装
  • Redis概述
  • Redis安装
  • 常用五大数据类型
  • Redis键(key)
  • Redis字符串(String)
  • Redis列表(List) - 单键多值
  • 结构说明
  • Redis集合(Set)
  • Redis哈希(Hash)
  • Redis有序集合Zset(sorted set)
  • 案例：如何利用zset实现一个文章访问量的排行榜？
  • Redis的发布和订阅
  • 发布订阅命令行实现
  • Redis新数据类型
  • Bitmaps
  • 常用命令
  • 实例:
  • HyperLogLog
  • 常用命令:
  • Geospatial
  • Java_Redis_Jedis
  • 引入需要的jar包
  • Jedis常用操作
  • 测试连接是否成功:
  • Jedis-API: Key
  • Jedis-API: String
  • Jedis-API: List
  • Jedis-API: Set
  • Jedis-API: Hash
  • Jedis-API: Zset
  • Redis与Spring Boot整合
  • 引入依赖
  • application.properties配置redis配置
  • 添加redis配置类
  • 我的常用
  • 网上:
  • Redis事务和锁机制
  • Multi、Exec、discard的三个事务命令
  • 事务的错误处理
  • 例子1:
  • 例子2:
  • 事务冲突解决办法
  • 例子:
  • 使用悲观锁
  • 使用乐观锁
  • 使用watch key例子演示
  • unwatch
  • Redis事务三特性
  • 单独的隔离操作
  • 没有隔离级别的概念
  • 不保证原子性
  • Redis持久化
  • RDB的优点
  • RDB的缺点
  • AOF 优点
  • AOF 缺点
  • 如何选择使用哪种持久化方式？
  • RDB（Redis DataBase）
  • save VS bgsave
  • 工作方式
  • 总结
  • AOF（Append Of File）
  • 工作方式
  • 开启aof
  • 修复aof文件
  • AOF同步频率设置
  • 总结
  • Redis_主从复制
  • 实操主从复制
  • 1.创建6379-81的文件夹
  • 2.复制redis.conf文件到6379-81的文件夹,并修改对应端口
  • 3.设置当本机为slave服务时，设置master服务的IP地址及端口，在Redis启动时，它会自动从master进行数据同步
  • 4.启动服务
  • 5.查看状态
  • 常用2招
  • 薪火相传
  • 反客为主
  • 复制原理
  • 添加哨兵模式
  • 实操添加哨兵模式
  • 1.新建一个redis_sentinel文件夹,并新建sentinel.conf文件
  • 2.启动哨兵模式
  • 3.查看是否启动成功,并且在sentinel中查看所监控的master 和slave 连接哨兵：
  • Redis6集群(redis-cluster)
  • 实操搭建redis集群(redis-cluster)
  • 1.新增对应6个端口的文件夹
  • 2.创建每个端口对应的redis.conf配置文件
  • 3.启动各种redis服务
  • 4.将六个节点合成一个集群
  • 5.通过 cluster nodes 命令查看集群信息
  • 6.测试
  • Redis应用常见的问题解决
  • 缓存穿透
  • 缓存击穿
  • 缓存雪崩
  • 分布式锁
  • 使用redis实现分布式锁
  • QA:如果中途崩溃，无法释放锁？
  • QA:由于命令是非原子的，所以还是会死锁,如何解决？
  • 首页
  • 关于
  • 归档
  • 邻居
  • 捐赠
  • 速查表

Redis6-学习笔记

  • 郭良俊只狗
  • 2022-03-04
  • 0

Redis6-学习笔记

Redis概述安装

redis中文网:http://www.redis.cn/

redis官网:https://redis.io/

Redis概述

Redis是一个开源的key-value存储系统。

和Memcached类似，它支持存储的value类型相对更多，包括string(字符串)、list(链表)、set(集合)、zset(sorted set --有序集合)和hash（哈希类型）。

这些数据类型都支持push/pop、add/remove及取交集并集和差集及更丰富的操作，而且这些操作都是原子性的。

在此基础上，Redis支持各种不同方式的排序。

与memcached一样，为了保证效率，数据都是缓存在内存中。

区别的是Redis会周期性的把更新的数据写入磁盘或者把修改操作写入追加的记录文件。

并且在此基础上实现了master-slave(主从)同步。

Redis安装

到官网下载:

image-20220228164720472

安装方式和之前的笔记差不多 -->https://gitee.com/guoliangjun17/bendibiji/blob/master/redis/redis%E7%B3%BB%E5%88%97-%E5%8D%95%E6%9C%BA%E5%AE%89%E8%A3%85.md

常用五大数据类型

redis常见数据类型操作命令http://www.redis.cn/commands.html

Redis键(key)

keys * 查看当前库所有key (匹配：keys *1)

exists key判断某个key是否存在

type key 查看你的key是什么类型

del key 删除指定的key数据

unlink key 根据value选择非阻塞删除

expire key 10 10秒钟：为给定的key设置过期时间

ttl key 查看还有多少秒过期，-1表示永不过期，-2表示已过期

select命令切换数据库

dbsize查看当前数据库的key的数量

flushdb清空当前库

flushall通杀全部库

Redis字符串(String)

set <key><value>添加键值对

image-20220228234135983

*NX：当数据库中key不存在时，可以将key-value添加数据库

*XX：当数据库中key存在时，将value添加数据库，与NX参数互斥

*EX：key的超时秒数

*PX：key的超时毫秒数，与EX互斥

get <key>查询对应键值

append <key> <value>将给定的 追加到原值的末尾

strlen <key>获得值的长度

setnx <key> <value>只有在 key 不存在时 设置 key 的值

incr <key>将 key 中储存的数字值增1,只能对数字值操作，如果为空，新增值为1

decr <key>将 key 中储存的数字值减1,只能对数字值操作，如果为空，新增值为-1

incrby / decrby <key><步长>将 key 中储存的数字值增减。如:incrby k1 10

mset <key1> <value1> <key2> <value2> 同时设置一个或多个 key-value对
mget <key1> <key2> <key3>同时获取一个或多个 value
msetnx <key1> <value1> <key2> <value2>同时设置一个或多个 key-value 对，当且仅当所有给定 key 都不存在。-->原子性，有一个失败则都失败

getrange <key><起始位置><结束位置>获得值的范围，类似java中的substring，前包，后包 如:

127.0.0.1:6379> set name guoliangjun
OK
127.0.0.1:6379> GETRANGE name 0 3
"guol"

setrange <key><起始位置><value>用 覆写所储存的字符串值，从<起始位置>开始(索引从0开始)。如:

127.0.0.1:6379> set name guoliangjun
OK
127.0.0.1:6379> SETRANGE name 3 abc
(integer) 11
127.0.0.1:6379> get name
"guoabcngjun"

setex <key> <过期时间> <value>设置键值的同时，设置过期时间，单位秒。
getset <key> <value>以新换旧，设置了新值同时获得旧值。如:

127.0.0.1:6379> GETSET name glj
"guoabcngjun"

String的数据结构为简单动态字符串(Simple Dynamic String,缩写SDS)。是可以修改的字符串，内部结构实现上类似于Java的ArrayList，采用预分配冗余空间的方式来减少内存的频繁分配.
内部为当前字符串实际分配的空间capacity一般要高于实际字符串长度len。当字符串长度小于1M时，扩容都是加倍现有的空间，如果超过1M，扩容时一次只会多扩1M的空间。需要注意的是字符串最大长度为512M。

Redis列表(List) - 单键多值

Redis 列表是简单的字符串列表，按照插入顺序排序。你可以添加一个元素到列表的头部（左边）或者尾部（右边）。
它的底层实际是个双向链表，对两端的操作性能很高，通过索引下标的操作中间的节点性能会较差。

lpush/rpush <key> <value1> <value2> ...从左边/右边插入一个或多个值。

lpop/rpop <key>从左边/右边吐出一个值。值在键在，值光键亡。

rpoplpush <key1><key2>从列表右边吐出一个值，插到列表左边。

lrange <key> <start> <stop> 按照索引下标获得元素(从左到右)

lindex <key> <index>按照索引下标获得元素(从左到右)

llen <key>获得列表长度

linsert <key> before/after <value> <newvalue>在的前面/后面插入插入值

lrem <key> <n> <value>从左边删除n个value(从左到右) ,如:LREM k2 2 v1

lset <key> <index> <value>将列表key下标为index的值替换成value,如:LSET k2 1 newv33

结构说明

List的数据结构为快速链表quickList

在列表元素较少的情况下会使用一块连续的内存存储，这个结构是ziplist，也即是压缩列表

当数据量比较多的时候才会改成quicklist

因为普通的链表需要的附加指针空间太大，会比较浪费空间。比如这个列表里存的只是int类型的数据，结构上还需要两个额外的指针prev和next。

Redis将链表和ziplist结合起来组成了quicklist。也就是将多个ziplist使用双向指针串起来使用。

这样既满足了快速的插入删除性能，又不会出现太大的空间冗余。

Redis集合(Set)

set对外提供的功能与list类似是一个列表的功能，特殊之处在于set是可以自动排重的，当你需要存储一个列表数据，又不希望出现重复数据时，set是一个很好的选择

Redis的Set是string类型的无序集合。

它底层其实是一个value为null的hash表，所以添加，删除，查找的复杂度都是O(1)

sadd <key><value1><value2> ..... 将一个或多个 member 元素加入到集合 key 中，已经存在的 member 元素将被忽略
smembers <key>取出该集合的所有值。

sismember <key> <value>判断集合是否为含有该值，有1，没有0

scard <key>返回该集合的元素个数。

srem <key> <value1> <value2> .... 删除集合中的某个元素。

spop <key>随机从该集合中吐出一个值。值在键在，值光键亡。

srandmember <key> <n>随机从该集合中取出n个值。不会从集合中删除 。

smove <source> <destination> value 把集合中一个值从一个集合移动到另一个集合,如:

127.0.0.1:6379> sadd k1 v1 v2 v3
(integer) 3
127.0.0.1:6379> SADD k2 v4 v5 v6
(integer) 3
127.0.0.1:6379> 
127.0.0.1:6379> SMOVE k1 k2 v3
(integer) 1
127.0.0.1:6379> SMEMBERS k2
1) "v3"
2) "v4"
3) "v6"
4) "v5"

sinter <key1> <key2>返回两个集合的交集元素。---共同好友!

sunion <key1> <key2> 返回两个集合的并集元素。

sdiff <key1> <key2>返回两个集合的差集元素(key1中的，不包含key2中的)

Redis哈希(Hash)

Redis hash 是一个键值对集合。
Redis hash是一个string类型的field和value的映射表，hash特别适合用于存储对象。
类似Java里面的Map<String,Object>

hset <key> <field> <value> 给集合中的 键赋值 如:HSET user:101 id 1

hget <key1> <field>从集合取出 value 如:HGET user:101 id
hmset <key1> <field1> <value1> <field2> <value2>... 批量设置hash的值,如:HMSET user:102 id 2 name glj2
hexists <key1> <field> 查看哈希表 key 中，给定域 field 是否存在。
hkeys <key>列出该hash集合的所有field
hvals <key>列出该hash集合的所有value
hincrby <key> <field> <increment>为哈希表 key 中的域 field 的值加上增量 1 如:HINCRBY user:102 age 2
hsetnx <key> <field> <value>将哈希表 key 中的域 field 的值设置为 value ，当且仅当域 field 不存在 .

Redis有序集合Zset(sorted set)

Redis有序集合zset与普通集合set非常相似，是一个没有重复元素的字符串集合。
不同之处是有序集合的每个成员都关联了一个评分（score）,这个评分（score）被用来按照从最低分到最高分的方式排序集合中的成员。集合的成员是唯一的，但是评分可以是重复了 。

zadd <key> <score1> <value1> <score2> <value2>…将一个或多个 member 元素及其 score 值加入到有序集 key 当中。如:ZADD topn 20 java 30 c++ 40 mysql

zrange <key> <start> <stop> [WITHSCORES]返回有序集 key 中，下标在之间的元素带WITHSCORES，可以让分数一起和值返回到结果集。

zrevrange <key> <start> <stop> [WITHSCORES] 返回有序集 key 中. 同上，改为从大到小排列

zrangebyscore key min max [withscores] [limit offset count]返回有序集 key 中，所有 score 值介于 min 和 max 之间(包括等于 min 或 max )的成员。有序集成员按 score 值递增(从小到大)次序排列。 如:ZRANGEBYSCORE topn 20 30

zrevrangebyscore key max min [withscores] [limit offset count] 同上，改为从大到小排列。

zincrby <key> <increment> <value> 为元素的score加上增量 如:ZINCRBY topn 50 java

zrem <key> <value>删除该集合下，指定值的元素

zcount <key> <min> <max>统计该集合，分数区间内的元素个数

zrank <key> <value>返回该值在集合中的排名，从0开始。

案例：如何利用zset实现一个文章访问量的排行榜？

image-20220301122026567

Redis的发布和订阅

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

发布订阅命令行实现

1.打开一个客户端订阅channel1

SUBSCRIBE channel1

2.打开另一个客户端，给channel1发布消息hello_redis

publish channel1 hello_redis

image-20220301220213089

Redis新数据类型

Bitmaps

Redis提供了Bitmaps这个“数据类型”可以实现对位的操作：

（1） Bitmaps本身不是一种数据类型， 实际上它就是字符串（key-value） ， 但是它可以对字符串的位进行操作。

（2） Bitmaps单独提供了一套命令， 所以在Redis中使用Bitmaps和使用字符串的方法不太相同。 可以把Bitmaps想象成一个以位为单位的数组， 数组的每个单元只能存储0和1， 数组的下标在Bitmaps中叫做偏移量。

常用命令

setbit <key> <offset> <value>设置Bitmaps中某个偏移量的值（0或1）

127.0.0.1:6379> SETBIT users:20220301 1 1
(integer) 0
127.0.0.1:6379> SETBIT users:20220301 6 1
(integer) 0
127.0.0.1:6379> SETBIT users:20220301 7 1
(integer) 0
127.0.0.1:6379> SETBIT users:20220301 12 1

​ getbit <key> <offset>获取Bitmaps中某个偏移量的值,如:`

127.0.0.1:6379> GETBIT users:20220301 6
(integer) 1
127.0.0.1:6379> GETBIT users:20220301 8
(integer) 0 # 不存在，所以也是返回0

bitcount<key>[start end]统计字符串从start字节到end字节比特值为1的数量,如:BITCOUNT users:20220301

bitop and(or/not/xor) <destkey> [key…]是一个复合操作， 它可以做多个Bitmaps的and（交集） 、 or（并集） 、 not（非） 、 xor（异或） 操作并将结果保存在destkey中。

实例:
2022-03-01 日访问网站的userid=1,2,5,9。
setbit unique:users:20220301 1 1
setbit unique:users:20220301 2 1
setbit unique:users:20220301 5 1
setbit unique:users:20220301 9 1

2022-02-28 日访问网站的userid=0,1,4,9。
setbit unique:users:20220228 0 1
setbit unique:users:20220228 1 1
setbit unique:users:20220228 4 1
setbit unique:users:20220228 9 1

计算出两天都访问过网站的用户数量
127.0.0.1:6379> bitop and unique:users:and:20220301_20220228 unique:users:20220301 unique:users:20220228

计算出任意一天都访问过网站的用户数量（例如月活跃就是类似这种,可以使用or求并集
127.0.0.1:6379> bitop or unique:users:or:20220301_20220228 unique:users:20220301 unique:users:20220228
(integer) 2
127.0.0.1:6379> BITCOUNT unique:users:or:20220301_20220228
(integer) 6

HyperLogLog

Redis HyperLogLog 是用来做基数统计的算法，HyperLogLog 的优点是，在输入元素的数量或者体积非常非常大时，计算基数所需的空间总是固定的、并且是很小的。
在 Redis 里面，每个 HyperLogLog 键只需要花费 12 KB 内存，就可以计算接近 2^64 个不同元素的基数。这和计算基数时，元素越多耗费内存就越多的集合形成鲜明对比。
但是，因为 HyperLogLog 只会根据输入元素来计算基数，而不会储存输入元素本身，所以 HyperLogLog 不能像集合那样，返回输入的各个元素

常用命令:

pfadd <key>< element> [element ...] 添加指定元素到 HyperLogLog 中,如:

127.0.0.1:6379> PFADD stu zhangshan
(integer) 1
127.0.0.1:6379> PFADD stu lisi
(integer) 1
127.0.0.1:6379> PFADD stu wangwu
(integer) 1
127.0.0.1:6379> 
127.0.0.1:6379> PFADD stu lisi
(integer) 0 #已经存在

pfcount <key> [key ...] 计算HLL的近似基数,如:

127.0.0.1:6379> PFADD stu2 lisi
(integer) 1
127.0.0.1:6379> PFADD stu2 test
(integer) 1
127.0.0.1:6379> PFCOUNT stu stu2
(integer) 4

作用:比如用HLL存储每天的UV，计算一周的UV可以使用7天的UV合并计算

pfmerge <destkey> <sourcekey> [sourcekey ...] 将一个或多个HLL合并后的结果存储在另一个HLL中

127.0.0.1:6379> PFMERGE stu3 stu stu2
OK
127.0.0.1:6379> PFCOUNT stu3
(integer) 4

作用:比如每月活跃用户可以使用每天的活跃用户来合并计算

Geospatial

Redis 3.2 中增加了对GEO类型的支持。GEO，Geographic，地理信息的缩写。该类型，就是元素的2维坐标，在地图上就是经纬度。redis基于该类型，提供了经纬度设置，查询，范围查询，距离查询，经纬度Hash等常见操作

geoadd<key> < longitude> <latitude> <member> [longitude latitude member...] 添加地理位置（经度，纬度，名称）

geoadd china:city 121.47 31.23 shanghai

geoadd china:city 106.50 29.53 chongqing 114.05 22.52 shenzhen 116.38 39.90 beijing

geopos <key> <member> [member...] 获得指定地区的坐标值

geodist <key> <member1> <member2> [m|km|ft|mi ] 获取两个位置之间的直线距离

georadius <key> <longitude> <latitude> radius m|km|ft|mi 以给定的经纬度为中心，找出某一半径内的元素

Java_Redis_Jedis

引入需要的jar包

<!-- https://mvnrepository.com/artifact/redis.clients/jedis -->
<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>3.3.0</version>
</dependency>

Jedis常用操作

测试连接是否成功:

@Test
public void testConnectRedis(){
    Jedis jedis = new Jedis("127.0.0.1", 26379);
    String pong = jedis.ping();
    System.out.println("连接成功："+pong);
    jedis.close();
}

Jedis-API: Key

@Test
public void testRedisKey(){
    Jedis jedis = new Jedis("127.0.0.1", 26379);
    jedis.set("k1", "v1");
    jedis.set("k2", "v2");
    jedis.set("k3", "v3");
    Set<String> keys = jedis.keys("*");
    System.out.println(keys.size());
    for (String key : keys) {
        System.out.println(key);
    }
    System.out.println(jedis.exists("k1"));
    System.out.println(jedis.ttl("k1"));
    System.out.println(jedis.get("k1"));
}

Jedis-API: String

@Test
public void testRedisString(){
    Jedis jedis = new Jedis("127.0.0.1", 26379);
    jedis.mset("str1","v1","str2","v2","str3","v3");
    System.out.println(jedis.mget("str1","str2","str3"));
}

Jedis-API: List

@Test
public void testRedisList(){
    Jedis jedis = new Jedis("127.0.0.1", 26379);
    List<String> list = jedis.lrange("mylist",0,-1);
    for (String element : list) {
        System.out.println(element);
    }
}

Jedis-API: Set

@Test
public void testRedisSet(){
    Jedis jedis = new Jedis("127.0.0.1", 26379);
    jedis.sadd("orders", "order01");
    jedis.sadd("orders", "order02");
    jedis.sadd("orders", "order03");
    jedis.sadd("orders", "order04");
    Set<String> smembers = jedis.smembers("orders");
    for (String order : smembers) {
        System.out.println(order);
    }
    jedis.srem("orders", "order02");
}

Jedis-API: Hash

@Test
public void testRedisHash(){
    Jedis jedis = new Jedis("127.0.0.1", 26379);
    jedis.hset("hash1","userName","glj");
    System.out.println(jedis.hget("hash1","userName"));
    Map<String,String> map = new HashMap<String,String>();
    map.put("telphone","13800138000");
    map.put("address","清远");
    map.put("email","admin@199604.com");
    jedis.hmset("hash2",map);
    List<String> result = jedis.hmget("hash2", "telphone","email");
    for (String element : result) {
        System.out.println(element);
    }
}

Jedis-API: Zset

@Test
public void testRedisZset(){
    Jedis jedis = new Jedis("127.0.0.1", 26379);
    jedis.zadd("zset01", 100d, "z3");
    jedis.zadd("zset01", 90d, "l4");
    jedis.zadd("zset01", 80d, "w5");
    jedis.zadd("zset01", 70d, "z6");

    Set<String> zrange = jedis.zrange("zset01", 0, -1);
    for (String e : zrange) {
        System.out.println(e);
    }
}

Redis与Spring Boot整合

引入依赖

<!--SpringBoot与Redis整合依赖-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

application.properties配置redis配置

#Redis服务器地址
spring.redis.host=127.0.0.1
#Redis服务器连接端口
spring.redis.port=26379
#密码
spring.redis.password=Jtcmcc@139.com
#Redis数据库索引（默认为0）
spring.redis.database= 0
#连接超时时间（毫秒）
spring.redis.timeout=1800000
#连接池最大连接数（使用负值表示没有限制）
spring.redis.lettuce.pool.max-active=20
#最大阻塞等待时间(负数表示没限制)
spring.redis.lettuce.pool.max-wait=-1
#连接池中的最大空闲连接
spring.redis.lettuce.pool.max-idle=5
#连接池中的最小空闲连接
spring.redis.lettuce.pool.min-idle=0

添加redis配置类

我的常用

@Configuration
public class RedisConfig {

    @Bean
    public RedisTemplate<String,String> redisTemplate(RedisConnectionFactory factory){
        RedisTemplate<String,String> redisTemplate=new RedisTemplate<>();
        RedisSerializer<String> redisSerializer = new StringRedisSerializer();
        redisTemplate.setConnectionFactory(factory);
        //key序列化
        redisTemplate.setKeySerializer(redisSerializer);
        //value序列化
        redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());

        //value hashmap序列化
        redisTemplate.setHashKeySerializer(redisSerializer);
        //key hashmap序列化
        redisTemplate.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());
        return redisTemplate;
    }
}

网上:

@Configuration
@EnableCaching
public class RedisConfig extends CachingConfigurerSupport{

    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        RedisSerializer<String> redisSerializer = new StringRedisSerializer();
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        template.setConnectionFactory(factory);
        //key序列化方式
        template.setKeySerializer(redisSerializer);
        //value序列化
        template.setValueSerializer(jackson2JsonRedisSerializer);
        //value hashmap序列化
        template.setHashKeySerializer(redisSerializer);
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
        return template;
    }

    @Bean
    public CacheManager cacheManager(RedisConnectionFactory factory) {
        RedisSerializer<String> redisSerializer = new StringRedisSerializer();
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        //解决查询缓存转换异常的问题
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        // 配置序列化（解决乱码的问题）,过期时间600秒
        RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig()
                .entryTtl(Duration.ofSeconds(600))
                .serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(redisSerializer))
                .serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(jackson2JsonRedisSerializer))
                .disableCachingNullValues();
        RedisCacheManager cacheManager = RedisCacheManager.builder(factory)
                .cacheDefaults(config)
                .build();
        return cacheManager;
    }
}

Redis事务和锁机制

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

Redis事务的主要作用就是串联多个命令防止别的命令插队

Multi、Exec、discard的三个事务命令

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

例子:

image-20220302121654815

image-20220302121747161

事务的错误处理

组队中某个命令出现了报告错误，执行时整个的所有队列都会被取消。

image-20220302122014629

例子1:

image-20220302122137362

如果执行阶段某个命令报出了错误，则只有报错的命令不会被执行，而其他的命令都会执行，不会回滚.

image-20220302122031014

例子2:

image-20220302122236342

事务冲突解决办法

例子:

一个请求想给金额减8000

一个请求想给金额减5000

一个请求想给金额减1000

image-20220302122622629

使用悲观锁

image-20220302122647649

悲观锁(Pessimistic Lock), 顾名思义，就是很悲观，每次去拿数据的时候都认为别人会修改，所以每次在拿数据的时候都会上锁，这样别人想拿这个数据就会block直到它拿到锁。传统的关系型数据库里边就用到了很多这种锁机制，比如行锁，表锁等，读锁，写锁等，都是在做操作之前先上锁。

使用乐观锁

image-20220302122718767

乐观锁(Optimistic Lock), 顾名思义，就是很乐观，每次去拿数据的时候都认为别人不会修改，所以不会上锁，但是在更新的时候会判断一下在此期间别人有没有去更新这个数据，可以使用版本号等机制。乐观锁适用于多读的应用类型，这样可以提高吞吐量。Redis就是利用这种check-and-set机制实现事务的

使用watch key例子演示

在执行multi之前，先执行watch key1 [key2],可以监视一个(或多个) key ，如果在事务执行之前这个(或这些) key 被其他命令所改动，那么事务将被打断。

image-20220302123128651

unwatch

取消 WATCH 命令对所有 key 的监视。

如果在执行 WATCH 命令之后，EXEC 命令或DISCARD 命令先被执行了的话，那么就不需要再执行UNWATCH 了。

Redis事务三特性

单独的隔离操作

​ 事务中的所有命令都会序列化、按顺序地执行。事务在执行的过程中，不会被其他客户端发送来的命令请求所打断。

没有隔离级别的概念

​ 队列中的命令没有提交之前都不会实际被执行，因为事务提交前任何指令都不会被实际执行

不保证原子性

​ 事务中如果有一条命令执行失败，其后的命令仍然会被执行，没有回滚

Redis持久化

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

  • RDB（Redis DataBase）
  • AOF（Append Of File）

RDB的优点

  • RDB是一个非常紧凑的文件,它保存了某个时间点得数据集,非常适用于数据集的备份,比如你可以在每个小时报保存一下过去24小时内的数据,同时每天保存过去30天的数据,这样即使出了问题你也可以根据需求恢复到不同版本的数据集.
  • RDB是一个紧凑的单一文件,很方便传送到另一个远端数据中心或者亚马逊的S3（可能加密），非常适用于灾难恢复.
  • RDB在保存RDB文件时父进程唯一需要做的就是fork出一个子进程,接下来的工作全部由子进程来做，父进程不需要再做其他IO操作，所以RDB持久化方式可以最大化redis的性能.
  • 与AOF相比,在恢复大的数据集的时候，RDB方式会更快一些.

RDB的缺点

  • 如果你希望在redis意外停止工作（例如电源中断）的情况下丢失的数据最少的话，那么RDB不适合你.虽然你可以配置不同的save时间点(例如每隔5分钟并且对数据集有100个写的操作),是Redis要完整的保存整个数据集是一个比较繁重的工作,你通常会每隔5分钟或者更久做一次完整的保存,万一在Redis意外宕机,你可能会丢失几分钟的数据.
  • RDB 需要经常fork子进程来保存数据集到硬盘上,当数据集比较大的时候,fork的过程是非常耗时的,可能会导致Redis在一些毫秒级内不能响应客户端的请求.如果数据集巨大并且CPU性能不是很好的情况下,这种情况会持续1秒,AOF也需要fork,但是你可以调节重写日志文件的频率来提高数据集的耐久度.

AOF 优点

  • 使用AOF 会让你的Redis更加耐久: 你可以使用不同的fsync策略：无fsync,每秒fsync,每次写的时候fsync.使用默认的每秒fsync策略,Redis的性能依然很好(fsync是由后台线程进行处理的,主线程会尽力处理客户端请求),一旦出现故障，你最多丢失1秒的数据.
  • AOF文件是一个只进行追加的日志文件,所以不需要写入seek,即使由于某些原因(磁盘空间已满，写的过程中宕机等等)未执行完整的写入命令,你也也可使用redis-check-aof工具修复这些问题.
  • Redis 可以在 AOF 文件体积变得过大时，自动地在后台对 AOF 进行重写： 重写后的新 AOF 文件包含了恢复当前数据集所需的最小命令集合。 整个重写操作是绝对安全的，因为 Redis 在创建新 AOF 文件的过程中，会继续将命令追加到现有的 AOF 文件里面，即使重写过程中发生停机，现有的 AOF 文件也不会丢失。 而一旦新 AOF 文件创建完毕，Redis 就会从旧 AOF 文件切换到新 AOF 文件，并开始对新 AOF 文件进行追加操作。
  • AOF 文件有序地保存了对数据库执行的所有写入操作， 这些写入操作以 Redis 协议的格式保存， 因此 AOF 文件的内容非常容易被人读懂， 对文件进行分析（parse）也很轻松。 导出（export） AOF 文件也非常简单： 举个例子， 如果你不小心执行了 FLUSHALL 命令， 但只要 AOF 文件未被重写， 那么只要停止服务器， 移除 AOF 文件末尾的 FLUSHALL 命令， 并重启 Redis ， 就可以将数据集恢复到 FLUSHALL 执行之前的状态。

AOF 缺点

  • 对于相同的数据集来说，AOF 文件的体积通常要大于 RDB 文件的体积。
  • 根据所使用的 fsync 策略，AOF 的速度可能会慢于 RDB 。 在一般情况下， 每秒 fsync 的性能依然非常高， 而关闭 fsync 可以让 AOF 的速度和 RDB 一样快， 即使在高负荷之下也是如此。 不过在处理巨大的写入载入时，RDB 可以提供更有保证的最大延迟时间（latency）。

如何选择使用哪种持久化方式？

一般来说， 如果想达到足以媲美 PostgreSQL 的数据安全性， 你应该同时使用两种持久化功能。

如果你非常关心你的数据， 但仍然可以承受数分钟以内的数据丢失， 那么你可以只使用 RDB 持久化。

有很多用户都只使用 AOF 持久化， 但我们并不推荐这种方式： 因为定时生成 RDB 快照（snapshot）非常便于进行数据库备份， 并且 RDB 恢复数据集的速度也要比 AOF 恢复的速度要快， 除此之外， 使用 RDB 还可以避免之前提到的 AOF 程序的 bug 。

RDB（Redis DataBase）

在默认情况下， Redis 将数据库快照保存在名字为 dump.rdb的二进制文件中。你可以对 Redis 进行设置， 让它在“ N 秒内数据集至少有 M 个改动”这一条件被满足时， 自动保存一次数据集。你也可以通过调用 SAVE或者 BGSAVE ， 手动让 Redis 进行数据集保存操作。(在redis.conf中配置文件名称，默认为dump.rdb)

比如说， 以下设置会让 Redis 在满足“ 60 秒内有至少有 1000 个键被改动”这一条件时， 自动保存一次数据集:

save 60 1000

save VS bgsave

save ：save时只管保存，其它不管，全部阻塞。手动保存。不建议。

bgsave：Redis会在后台异步进行快照操作，快照同时还可以响应客户端请求。

可以通过lastsave命令获取最后一次成功执行快照的时间

工作方式

当 Redis 需要保存 dump.rdb 文件时， 服务器执行以下操作:

  • Redis 调用forks. 同时拥有父进程和子进程。
  • 子进程将数据集写入到一个临时 RDB 文件中。
  • 当子进程完成对新 RDB 文件的写入时，Redis 用新 RDB 文件替换原来的 RDB 文件，并删除旧的 RDB 文件。

image-20220302150418531

总结

image-20220302151040166

AOF（Append Of File）

以日志的形式来记录每个写操作（增量保存），将Redis执行过的所有写指令记录下来(读操作不记录)， 只许追加文件但不可以改写文件，redis启动之初会读取该文件重新构建数据，换言之，redis 重启的话就根据日志文件的内容将写指令从前到后执行一次以完成数据的恢复工作

工作方式

  1. 客户端的请求写命令会被append追加到AOF缓冲区内；

  2. AOF缓冲区根据AOF持久化策略[always,everysec,no]将操作sync同步到磁盘的AOF文件中；

  3. AOF文件大小超过重写策略或手动重写时，会对AOF文件rewrite重写，压缩AOF文件容量；

  4. Redis服务重启时，会重新load加载AOF文件中的写操作达到数据恢复的目的；

image-20220302181738144

开启aof

aof默认是不开启的,需要在您这边的redis.conf文件设置,默认文件名为appendonly.aof

修改默认的appendonly no，改为yes,即开启aof

修复aof文件

遇到AOF文件损坏，通过redis-check-aof --fix appendonly.aof进行恢复

AOF同步频率设置

appendfsync always 始终同步，每次Redis的写入都会立刻记入日志；性能较差但数据完整性比较好

appendfsync everysec 每秒同步，每秒记入日志一次，如果宕机，本秒的数据可能丢失。

appendfsync no redis不主动进行同步，把同步时机交给操作系统。

总结

image-20220302183321974

Redis_主从复制

主机数据更新后根据配置和策略， 自动同步到备机的master/slaver机制，Master以写为主，Slave以读为主

image-20220303235154295

实操主从复制

1.创建6379-81的文件夹

cd /etc/redis
mkdir 63{79..81}

2.复制redis.conf文件到6379-81的文件夹,并修改对应端口

cp redis.conf ./6379 
cp redis.conf ./6380 
cp redis.conf ./6381

# 修改对应端口
sed -i 's/6379/6380/g' ./6380/redis.conf
sed -i 's/6379/6381/g' ./6381/redis.conf

3.设置当本机为slave服务时，设置master服务的IP地址及端口，在Redis启动时，它会自动从master进行数据同步

可修改配置文件:

echo "slaveof 127.0.0.1 6379" >> ./6380/redis.conf

也可在从机命令行输入:slaveof <ip> <port>

4.启动服务

[root@gzbsc001 redis]# redis-server ./6379/redis.conf
[root@gzbsc001 redis]# redis-server ./6380/redis.conf
[root@gzbsc001 redis]# redis-server ./6381/redis.conf

5.查看状态

info replication

127.0.0.1:6379> info replication
# Replication
role:master
connected_slaves:2
slave0:ip=127.0.0.1,port=6380,state=online,offset=42,lag=1
slave1:ip=127.0.0.1,port=6381,state=online,offset=42,lag=0
master_failover_state:no-failover
master_replid:46af5e8112a766addca79a01c13bca39947955eb
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:42
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:1
repl_backlog_histlen:42

127.0.0.1:6380> info replication
# Replication
role:slave
master_host:127.0.0.1
master_port:6379
master_link_status:up
master_last_io_seconds_ago:9
master_sync_in_progress:0
slave_read_repl_offset:42
slave_repl_offset:42
slave_priority:100
slave_read_only:1
replica_announced:1
connected_slaves:0
master_failover_state:no-failover
master_replid:46af5e8112a766addca79a01c13bca39947955eb
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:42
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:1
repl_backlog_histlen:42

127.0.0.1:6381> info replication
# Replication
role:slave
master_host:127.0.0.1
master_port:6379
master_link_status:up
master_last_io_seconds_ago:9
master_sync_in_progress:0
slave_read_repl_offset:42
slave_repl_offset:42
slave_priority:100
slave_read_only:1
replica_announced:1
connected_slaves:0
master_failover_state:no-failover
master_replid:46af5e8112a766addca79a01c13bca39947955eb
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:42
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:1
repl_backlog_histlen:42

常用2招

薪火相传

上一个Slave可以是下一个slave的Master，Slave同样可以接收其他 slaves的连接和同步请求，那么该slave作为了链条中下一个的master, 可以有效减轻master的写压力,去中心化降低风险。

用 slaveof <ip> <port>

中途变更转向:会清除之前的数据，重新建立拷贝最新的

风险是一旦某个slave宕机，后面的slave都没法备份

主机挂了，从机还是从机，无法写数据了

反客为主

当一个master宕机后，后面的slave可以立刻升为master，其后面的slave不用做任何修改。

用slaveof no one 将从机变为主机。

复制原理

  1. Slave启动成功连接到master后会发送一个sync命令
  2. Master接到命令启动后台的存盘进程，同时收集所有接收到的用于修改数据集命令， 在后台进程执行完毕之后，master将传送整个数据文件到slave,以完成一次完全同步
  3. 全量复制：而slave服务在接收到数据库文件数据后，将其存盘并加载到内存中。
  4. 增量复制：Master继续将新的所有收集到的修改命令依次传给slave,完成同步
  5. 但是只要是重新连接master,一次完全同步（全量复制)将被自动执行

image-20220304001601001

添加哨兵模式

反客为主的自动版，能够后台监控主机是否故障，如果故障了根据投票数自动将从库转换为主库

image-20220304001647901

实操添加哨兵模式

1.新建一个redis_sentinel文件夹,并新建sentinel.conf文件
mkdir redis_sentinel
cd redis_sentinel/
touch sentinel.conf

sentinel.conf文件配置内容

# Example sentinel.conf

# port <sentinel-port>
port 26379

# 守护进程模式
daemonize yes

# 指明日志文件名
logfile "/etc/redis/redis_sentinel/sentinel.log"

# 工作路径，sentinel一般指定/tmp比较简单
dir "/etc/redis/redis_sentinel"

# 哨兵监控这个master，在至少quorum个哨兵实例都认为master down后把master标记为odown
# （objective down客观down；相对应的存在sdown，subjective down，主观down）状态。
# slaves是自动发现，所以你没必要明确指定slaves。
sentinel monitor MyMaster 127.0.0.1 6379 1

# master或slave多长时间（默认30秒）不能使用后标记为s_down状态。
sentinel down-after-milliseconds MyMaster 1500

#主备切换时，最多有多少个slave同时对新的master进行同步，这里设置为默认的1。
#snetinel parallel-syncs MyMaster 1

# 若sentinel在该配置值内未能完成failover操作（即故障时master/slave自动切换），则认为本次failover失败。
sentinel down-after-milliseconds MyMaster 3000

# 设置master和slaves验证密码
#sentinel failover-timeout MyMaster 18000
#sentinel auth-pass MyMaster testmaster123
2.启动哨兵模式
cd /etc/redis
redis-sentinel ./redis_sentinel/sentinel.conf
3.查看是否启动成功,并且在sentinel中查看所监控的master 和slave 连接哨兵：
[root@gzbsc001 redis_sentinel]# redis-sentinel ./sentinel.conf 
[root@gzbsc001 redis_sentinel]# 
[root@gzbsc001 redis_sentinel]# ps -ef |grep redis
root       663     1  0 00:06 ?        00:00:01 redis-server 0.0.0.0:6379
root      1078 29636  0 00:07 pts/0    00:00:00 redis-cli -p 6379
root      1732     1  0 00:10 ?        00:00:01 redis-server 0.0.0.0:6380
root      1743     1  0 00:10 ?        00:00:01 redis-server 0.0.0.0:6381
root      1759  1400  0 00:10 pts/1    00:00:00 redis-cli -p 6381
root      1997  1949  0 00:10 pts/2    00:00:00 redis-cli -p 6380
root      6085     1  0 00:25 ?        00:00:00 redis-sentinel *:26379 [sentinel]
root      6093  4290  0 00:25 pts/3    00:00:00 grep --color=auto redis
#-- 连接哨兵
redis-cli -p 26379
#查看master:  
SENTINEL masters

Redis注意事项：
1.最大内存问题：要设置好最大内存，以防不停的申请内存，造成系统内存都被用完。
2.密码问题：需要设置复杂一些，防止暴力破解。

Redis6集群(redis-cluster)

Redis 集群实现了对Redis的水平扩容，即启动N个redis节点，将整个数据库分布存储在这N个节点中，每个节点存储总数据的1/N。

Redis 集群通过分区（partition）来提供一定程度的可用性（availability）： 即使集群中有一部分节点失效或者无法进行通讯， 集群也可以继续处理命令请求。

实操搭建redis集群(redis-cluster)

制作6个实例，6379,6380,6381,6389,6390,6391

1.新增对应6个端口的文件夹

mkdir 63{79..81}
mkdir 63{89..91}

2.创建每个端口对应的redis.conf配置文件

include /usr/local/redis/conf/redis.conf
# 守护进程模式
daemonize yes 
# 监听端口
port 6379
# pid file 修改pidfile指向路径
pidfile "/usr/local/redis/conf/6379/redis_6379.pid"
# 持久化数据库的文件名
dbfilename "dump6379.rdb"
# 工作目录
dir "/usr/local/redis/conf/6379"
# 指明日志文件名
logfile "/usr/local/redis/conf/6379/redis_err_6379.log"
#如果是yes，表示启用集群，否则以单例模式启动
cluster-enabled yes
#请注意，尽管有此选项的名称
cluster-config-file nodes-6379.conf
#Redis群集节点超过不可用的最长时间
cluster-node-timeout 15000

其他端口安装这个去修改对应端口即可

sed -i 's/6379/6380/g' 6380/redis.conf

...

3.启动各种redis服务

/usr/local/redis/bin/redis-server /usr/local/redis/conf/6379/redis.conf 
/usr/local/redis/bin/redis-server /usr/local/redis/conf/6380/redis.conf 
/usr/local/redis/bin/redis-server /usr/local/redis/conf/6381/redis.conf 
/usr/local/redis/bin/redis-server /usr/local/redis/conf/6389/redis.conf 
/usr/local/redis/bin/redis-server /usr/local/redis/conf/6390/redis.conf 
/usr/local/redis/bin/redis-server /usr/local/redis/conf/6391/redis.conf

4.将六个节点合成一个集群

组合之前，请确保所有redis实例启动后，nodes-xxxx.conf文件都生成正常。

/usr/local/redis/bin/redis-cli --cluster create --cluster-replicas 1 192.168.1.88:6379 192.168.1.88:6380 192.168.1.88:6381 192.168.1.88:6389 192.168.1.88:6390 192.168.1.88:6391

此处不要用127.0.0.1， 请用真实IP地址!

--replicas 1 采用最简单的方式配置集群，一台主机，一台从机，正好三组。

如果加了密码，那么创建集群时就需要加-a passwd参数

不然会报以下错误:

[ERR] Node 10.209.17.10:7001 NOAUTH Authentication required.

5.通过 cluster nodes 命令查看集群信息

/usr/local/redis/bin/redis-cli -p 6380 -h 127.0.0.1 -c cluster nodes

[root@sdw88 conf]# /usr/local/redis/bin/redis-cli -p 6380 -h 127.0.0.1 -c cluster nodes
be276ac478812a8d972d56f03cbf96efa58f8c60 192.168.1.88:6391@16391 slave 6a23d938e20642ba97c88dc9c0b250be9687837a 0 1646405200544 3 connected
564f9e44a95756f2d6f48b657d806cef851846bb 192.168.1.88:6380@16380 myself,master - 0 1646405198000 2 connected 5461-10922
6a23d938e20642ba97c88dc9c0b250be9687837a 192.168.1.88:6381@16381 master - 0 1646405200000 3 connected 10923-16383
5e19bbd4b38d49fad458f1148f4c5520be1a51ff 192.168.1.88:6389@16389 slave 0234fab73f450f204892fa06de15c03405147bc5 0 1646405199000 1 connected
0234fab73f450f204892fa06de15c03405147bc5 192.168.1.88:6379@16379 master - 0 1646405199541 1 connected 0-5460
a0f204bb7fcbcbbe61ceea2c92f3251f5b46fd50 192.168.1.88:6390@16390 slave 564f9e44a95756f2d6f48b657d806cef851846bb 0 1646405198538 2 connected

-c 采用集群策略连接，设置数据会自动切换到相应的写主机

6.测试

/usr/local/redis/bin/redis-cli -p 6380 -h 127.0.0.1 -c

[root@sdw88 conf]# /usr/local/redis/bin/redis-cli -p 6380 -h 127.0.0.1 -c
127.0.0.1:6380> 
127.0.0.1:6380> keys *
(empty array)
127.0.0.1:6380> 
127.0.0.1:6380> set k1 v1
-> Redirected to slot [12706] located at 192.168.1.88:6381
OK
192.168.1.88:6381> get k1
"v1"
192.168.1.88:6381> 
192.168.1.88:6381> set k2 v2
-> Redirected to slot [449] located at 192.168.1.88:6379
OK
192.168.1.88:6379>

Redis应用常见的问题解决

缓存穿透

简单的讲缓存穿透：缓存层不命中，存储层不命中。

处理方式:

  1. 缓存空对象，不过此时会占用更多内存空间，所以根据大家业务特性去设置超时时间来控制内存占用的问题。
  2. 布隆过滤器。
  3. 实时监控

缓存击穿

key对应的数据存在，但在redis中过期，此时若有大量并发请求过来，这些请求发现缓存过期一般都会从后端DB加载数据并回设到缓存中(对应1个key过期)

处理方式:

1.预先设置热门数据

2.实时调整

3.使用锁

缓存雪崩

缓存雪崩与缓存击穿的区别在于这里针对很多key缓存，前者则是某一个key

处理方式:

1.构建多级缓存架构：nginx缓存 + redis缓存 +其他缓存（ehcache等）

2.使用锁或队列

3.设置过期标志更新缓存

4.将缓存失效时间分散开

分布式锁

分布式锁主流的实现方案：

  1. 基于数据库实现分布式锁

  2. 基于缓存（Redis等）

  3. 基于Zookeeper

每一种分布式锁解决方案都有各自的优缺点：

  1. 性能：redis最高

  2. 可靠性：zookeeper最高

使用redis实现分布式锁

setnx (set if not exists),如果创建成功则表示获取到锁。

setnx lock true 创建锁
del lock 释放锁
QA:如果中途崩溃，无法释放锁？

此时需要考虑到超时时间的问题。比如 :expire lock 300

QA:由于命令是非原子的，所以还是会死锁,如何解决？

Redis 支持 set 并设置超时时间的功能。

比如: set lock true ex 30 nx

image-20220304233657133

© 2012 - 2024 记忆角落 网站统计    由又拍云CDN赞助 提供云储存服务
Theme by Wing 粤ICP备14056850号