redis基本认识

1.Redis为什么是单线程

在Redis4.0之前，Redis是 单线程运行的。redis 单线程指的是网络请求模块使用了一个线程，即一个线程处理所有网络请求，其他模块仍用了多个线程。对于Redis来说，主要的性能瓶颈是内存或者网络带宽，而并非CPU。

2.Redis为什么这么快

1、完全基于内存，绝大部分请求是纯粹的内存操作，非常快速。数据存在内存中，类似于HashMap，HashMap的优势就是查找和操作的时间复杂度都是O(1)；

2、数据结构简单，对数据操作也简单，Redis中的数据结构是专门进行设计的；

3、采用单线程，避免了不必要的上下文切换和竞争条件，也不存在多进程或者多线程导致的切换而消耗 CPU，不用去考虑各种锁的问题，不存在加锁释放锁操作，没有因为可能出现死锁而导致的性能消耗；

4、使用多路I/O复用模型，非阻塞IO；

5、使用底层模型不同，它们之间底层实现方式以及与客户端之间通信的应用协议不一样，Redis直接自己构建了VM 机制 ，因为一般的系统调用系统函数的话，会浪费一定的时间去移动和请求；

redis key值操作

1、列出所有的key

1
redis> keys *

2、列出匹配的key

1
2
3
redis>keys apple*
1) apple1
2) apple2

3、查找key

1
get key

4、删除key

1
del key

5、添加值

1
set [key] [value]

6、清屏

1
2
// 功能：清除屏幕中的信息
clear

7、退出客户端命令行模式

1
2
3
quit
exit
<ESC>

8、多数据操作

1
2
3
4
// 修改多个数据 --> Multiple
mset key1 value1 key2 value2
// 获取多个数据  
mget key1 key2

9、string 类型数据的基本操作

  • 获取字符长度(string类型)
1
strlen key
  • 追加信息到原始信息后部（如果原始信息存在就追加，否则新建）
1
append key value
  • 数据库中的热点数据key命名惯例
1
表名:主键名:主键值:字段名

redis 清空缓存命令

flushall ——> 清空整个 Redis 服务器的数据(删除所有数据库的所有 key )

flushdb ——> 清空当前数据库中的所有 key

Linux下redis启动命令

1、启动 redis-server

1
2
# 注意配置文件路径
redis-server ./redis.conf

2、带配置文件启动

1
2
3
4
# 注意配置文件路径
redis-server ./redis.conf 
# 准确来说是：
/usr/local/bin/redis-server  /home/data/redis-3.2.1/redis.conf

3、停止 redis 命令 

1
redis-cli shutdown

4、倘若不知道redis-server文件位置输入如下命令查询位置

1
find / -name redis-server

5、查看是否启动成功

1
netstat -nplt

注意：默认端口号是6379

redis的基本类型

String字符串

Redis的字符串是动态字符串，是可以修改的字符串，内部结构实现上类似于Java的ArrayList，采用预分配冗余空间的方式来减少内存的频繁分配，

如图中所示，内部为当前字符串实际分配的空间capacity一般要高于实际字符串长度len。当字符串长度小于1M时，扩容都是加倍现有的空间，如果超过1M，扩容时一次只会多扩1M的空间。需要注意的是字符串最大长度为512M。

命令 表达式 说明
SET SET key “value” [EX / PX] 或 [NX / XX] 将字符串值 value 关联到 key。
SETNX SETNX key value 只在键 key 不存在的情况下， 将键 key 的值设置为 value 。
SETEX SETEX key seconds value 将键 key 的值设置为 value ， 并将键 key 的生存时间设置为 seconds 秒钟。
PSETEX PSETEX key milliseconds value 将键 key 的值设置为 value ， 并将键 key 的生存时间设置为 milliseconds毫秒。
GET GET key 返回与键 key 相关联的字符串值。
GETSET GETSET key value 将键 key 的值设为 value ， 并返回键 key 在被设置之前的旧值。
STRLEN STRLEN key 返回键 key 储存的字符串值的长度。
APPEND APPEND key value 如果键 key 已经存在并且它的值是一个字符串， APPEND 命令将把 value 追加到键 key 现有值的末尾。存在，就像执行 SET key value 一样。
SETRANGE SETRANGE key offset value 从偏移量 offset 开始， 用 value 参数覆写键 key 储存的字符串对应长度部分成set的value值。
GETRANGE GETRANGE key start end 返回键 key 储存的字符串值的指定部分， 字符串的截取范围由 start 和 end 两个偏移量决定 (包括 start 和 end 在内)。-1 表示最后一个字符， -2 表示倒数第二个字符， 以此类推。
INCR INCR key 为键 key 储存的数字值加上一。如果键 key 不存在， 那么它的值会先被初始化为 0 ， 然后再执行 INCR 命令。要求整型
INCRBY INCRBY key increment 为键 key 储存的数字值加上增量 increment 。其余同INCR
DECR DECR key 为键 key 储存的数字值减去一。
DECRBY DECRBY key decrement 将键 key 储存的整数值减去减量 decrement 。
INCRBYFLOAT INCRBYFLOAT key increment 为键 key 储存的值加上浮点数增量 increment 。
MSET MSET key value [key value key1 value1 …] 同时为多个键设置值。MSET 是一个原子性(atomic)操作， 所有给定键都会在同一时间内被设置， 不会出现某些键被设置了但是另一些键没有被设置的情况。
MSETNX MSETNX key value [key value …] 当且仅当所有给定键都不存在时， 为所有给定键设置值。MSETNX 是一个原子性(atomic)操作， 所有给定键要么就全部都被设置， 要么就全部都不设置， 不可能出现第三种状态。
MGET MGET key [key …] 返回给定的一个或多个字符串键的值。如果给定的字符串键里面， 有某个键不存在， 那么这个键的值将以特殊值 nil 表示。

1、EX seconds：将过期时间设置成多少秒

1
2
3
4
5
6
7
8
redis> SET key-with-expire-time "hello" EX 10086
OK

redis> GET key-with-expire-time
"hello"

redis> TTL key-with-expire-time
(integer) 10069

同理，PX milliseconds 是 将键的过期时间设置为 milliseconds 毫秒。

TTL key

以秒为单位，返回给定 key 的剩余生存时间(TTL, time to live)。

返回值：

当 key 不存在时，返回 -2 。
当 key 存在但没有设置剩余生存时间时，返回 -1 。
否则，以秒为单位，返回 key 的剩余生存时间。

2、SETNX 和 SETEX 的区别

SETNX ：
只在键 key 不存在的情况下， 将键 key 的值设置为 value 。
若键 key 已经存在， 则 SETNX 命令不做任何动作。
SETNX 是『SET if Not eXists』(如果不存在，则 SET)的简写。

SETEX：
将键 key 的值设置为 value ， 并将键 key 的生存时间设置为 seconds 秒钟。
如果键 key 已经存在， 那么 SETEX 命令将覆盖已有的值。
SETEX 命令的效果和以下两个命令的效果类似：

3、SETEX 对比 SET 设置过期时间

1
2
3
SETEX命令的效果和以下两个命令的效果类似：
SET key value
EXPIRE key seconds  # 设置生存时间

SETEX 和这两个命令的不同之处在于 SETEX 是一个 原子（atomic）操作， 它可以在同一时间内完成设置值和设置过期时间这两个操作， 因此 SETEX 命令在储存缓存的时候非常实用。

4、INCR、INCRBY、DECR、DECRBY的key值需要是整型

哈希表Hash

哈希等价于Java语言的HashMap或者是Python语言的dict，在实现结构上它使用二维结构，第一维是数组，第二维是链表，hash的内容key和value存放在链表中，数组里存放的是链表的头指针。通过key查找元素时，先计算key的hashcode，然后用hashcode对数组的长度进行取模定位到链表的表头，再对链表进行遍历获取到相应的value值，链表的作用就是用来将产生了「hash碰撞」的元素串起来。

命令 表达式 说明
HSET HSET hash field value 将哈希表 hash 中域 field 的值设置为 value 。
HSETNX HSETNX hash field value 当且仅当域 field 尚未存在于哈希表的情况下， 将它的值设置为 value 。如果给定域已经存在于哈希表当中， 那么命令将放弃执行设置操作。
HGET HGET hash field 返回哈希表中给定域的值。
HEXISTS HEXISTS hash field 检查给定域 field 是否存在于哈希表 hash 当中。
HDEL HDEL key field [field …] 删除哈希表 key 中的一个或多个指定域，不存在的域将被忽略。
HLEN HLEN key 返回哈希表 key 中域的数量。当 key 不存在时，返回 0 。
HSTRLEN HSTRLEN key field 返回哈希表 key 中， 与给定域 field 相关联的值的字符串长度（string length）。(VERSION\>= 3.2.0)
HINCRBY HINCRBY key field increment 为哈希表 key 中的域 field 的值加上增量 increment 。
HINCRBYFLOAT HINCRBYFLOAT key field increment 为哈希表 key 中的域 field 加上浮点数增量 increment 。
HMSET HMSET key field value [field value …] 同时将多个 field-value (域-值)对设置到哈希表 key 中。
HMGET HMGET key field [field …] 返回哈希表 key 中，一个或多个给定域的值。
HKEYS HKEYS key 返回哈希表 key 中的所有域。
HVALS HVALS key 返回哈希表 key 中所有域的值。
HGETALL HGETALL key 返回哈希表 key 中，所有的域和值。

1、添加多个值到Hash表中

1
2
127.0.0.1:6379> HMSET db redis "redis.com" mysql "mysql.com" key "value" key1 value1
OK

2、查看此Hash表数据

1
2
3
4
5
6
7
8
9
10
11
127.0.0.1:6379> HGETALL db
 1) "mysql"
 2) "mysql.com"
 3) "ac"
 4) "bc"
 5) "redis"
 6) "redis.com"
 7) "key"
 8) "value"
 9) "key1"
10) "value1"

3、删除Hash表db中key1的域

1
2
127.0.0.1:6379> HDEL db key1
(integer) 1

4、获取Hash表db中所有域值

1
2
3
4
5
6
7
8
9
127.0.0.1:6379> HGETALL db
1) "mysql"
2) "mysql.com"
3) "ac"
4) "bc"
5) "redis"
6) "redis.com"
7) "key"
8) "value"

由此可见，redis的hash表中键值对存放是奇偶邻近存放，删除了哈希表中为key1的域后，对应域的值value1也没了，所以受影响的行数一条，但是查看所有的时候记录行数减了2.

列表List

列表的存储结构用的是链表而不是数组，而且链表还是双向链表。

命令 表达式 说明
LPUSH LPUSH key value [value …] 将一个或多个值 value 插入到列表 key 的表头
LPUSHX LPUSHX key value 将值 value 插入到列表 key 的表头，当且仅当 key 存在并且是一个列表。当 key 不存在时， LPUSHX 命令什么也不做。
RPUSH RPUSH key value [value …] 将一个或多个值 value 插入到列表 key 的表尾(最右边)。不存在，一个空列表会被创建并执行 RPUSH 操作。
RPUSHX RPUSHX key value 将值 value 插入到列表 key 的表尾，当且仅当 key 存在并且是一个列表。
LPOP LPOP key 移除并返回列表 key 的头元素。当 key 不存在时，返回 nil 。
RPOP RPOP key 移除并返回列表 key 的尾元素。
RPOPLPUSH RPOPLPUSH source destination 将列表 source 中的最后一个元素(尾元素)弹出，并返回给客户端。 将 source 弹出的元素插入到列表 destination ，作为 destination 列表的的头元素。
LREM LREM key count value 根据参数 count 的值，移除列表中与参数 value 相等的元素。
LLEN LLEN key 返回列表 key 的长度。如果 key 不存在，则 key 被解释为一个空列表，返回 0 .
LINDEX LINDEX key index 返回列表 key 中，下标为 index 的元素，-1 表示列表的最后一个元素。
LINSERT LINSERT key BEFORE|AFTER pivot value 将值 value 插入到列表 key 当中，位于值 pivot 之前或之后。命令执行成功，返回插入操作完成之后，列表的长度
LSET LSET key index value 将列表 key 下标为 index 的元素的值设置为 value
LRANGE LRANGE key start stop 返回列表 key 中指定区间内的元素，区间以偏移量 start 和 stop 指定。
LTRIM LTRIM key start stop 对一个列表进行修剪(trim)，就是说，让列表只保留指定区间内的元素，不在指定区间之内的元素都将被删除。

1、LPUSH添加元素操作

1
2
3
4
5
6
7
8
9
10
11
# 加入单个元素
redis> LPUSH languages python
(integer) 1

# 加入重复元素
redis> LPUSH languages python
(integer) 2

# 加入多个元素
redis> LPUSH mylist a b c
(integer) 3
1
2
3
4
redis> LRANGE mylist 0 -1
1) "c"
2) "b"
3) "a"

2、RPOPLPUSH特殊情况 - 旋转(rotation)操作

如果 source 和 destination 相同，则列表中的表尾元素被移动到表头，并返回该元素，可以把这种特殊情况视作列表的旋转(rotation)操作。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# source 和 destination 相同
redis> LRANGE number 0 -1
1) "1"
2) "2"
3) "3"
4) "4"

redis> RPOPLPUSH number number
"4"

redis> LRANGE number 0 -1           # 4 被旋转到了表头
1) "4"
2) "1"
3) "2"
4) "3"

redis> RPOPLPUSH number number
"3"

redis> LRANGE number 0 -1           # 这次是 3 被旋转到了表头
1) "3"
2) "4"
3) "1"
4) "2"

3、LREM参数 count 的值说明

count 的值可以是以下几种：

  • count > 0 : 从表头开始向表尾搜索，移除与 value 相等的元素，数量为 count 。
  • count < 0 : 从表尾开始向表头搜索，移除与 value 相等的元素，数量为 count 的绝对值。
  • count = 0 : 移除表中所有与 value 相等的值。

4、LINSERT 的详解

1
2
3
4
5
6
7
8
9
10
11
12
13
14
redis> RPUSH mylist "Hello"
(integer) 1

redis> RPUSH mylist "World"
(integer) 2

# 在key为“mylist”列表中的pivot为“world”之前插入"There"，并返回操作完成之后列表的长度
redis> LINSERT mylist BEFORE "World" "There"
(integer) 3

redis> LRANGE mylist 0 -1
1) "Hello"
2) "There"
3) "World"

集合set

命令 表达式 说明
SADD SADD key member [member …] 将一个或多个 member 元素加入到集合 key 当中，已经存在于集合的 member 元素将被忽略。假如 key 不存在，则创建一个只包含 member 元素作成员的集合。
SISMEMBER SISMEMBER key member 判断 member 元素是否集合 key 的成员。
SPOP SPOP key 移除并返回集合中的一个随机元素。
SRANDMEMBER SRANDMEMBER key [count] 只提供 key 参数时，返回一个元素；如果还提供了 count 参数，那么返回一个数组；如果集合为空，返回空数组。
SREM SREM key member [member …] 移除集合 key 中的一个或多个 member 元素，不存在的 member 元素会被忽略。
SMOVE SMOVE source destination member 如果 source 集合不存在或不包含指定的 member 元素，则 SMOVE 命令不执行任何操作，仅返回 0 。否则， member 元素从 source 集合中被移除，并添加到 destination 集合中去。
SCARD SCARD key 返回集合 key 的基数(集合中元素的数量)。
SMEMBERS SMEMBERS key 返回集合 key 中的所有成员。
DEL DEL key 清空集合key
SINTER SINTER key [key …] 返回一个集合的全部成员，该集合是所有给定集合的交集。
SINTERSTORE SINTERSTORE destination key [key …] 返回结果集是一个或者多个集合中交集的成员数量。
SUNION SUNION key [key …] 返回一个集合的全部成员，该集合是所有给定集合的并集。
SUNIONSTORE SUNIONSTORE destination key [key …] 返回结果集是一个或者多个集合中并集的成员数量。
SDIFF SDIFF key [key …] 返回一个集合的全部成员，该集合是所有给定集合之间的差集。
SDIFFSTORE SDIFFSTORE destination key [key …] 返回结果集是一个或者多个集合中差集的成员数量。

1、SRANDMEMBER可选的参数

可选的 count 参数

  • 如果 count 为正数，且小于集合基数，那么命令返回一个包含 count 个元素的数组，数组中的元素各不相同。如果 count 大于等于集合基数，那么返回整个集合。
  • 如果 count 为负数，那么命令返回一个数组，数组中的元素可能会重复出现多次，而数组的长度为 count 的绝对值。

2、SMOVE 是原子性操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
redis> SMEMBERS songs
1) "Billie Jean"
2) "Believe Me"

redis> SMEMBERS my_songs
(empty list or set)

redis> SMOVE songs my_songs "Believe Me"
(integer) 1

redis> SMEMBERS songs
1) "Billie Jean"

redis> SMEMBERS my_songs
1) "Believe Me"

有序集合zset

zset底层实现使用了两个数据结构，第一个是hash，第二个是跳跃列表，hash的作用就是关联元素value和权重score，保障元素value的唯一性，可以通过元素value找到相应的score值。跳跃列表的目的在于给元素value排序，根据score的范围获取元素列表。

命令 表达式 说明
ZADD ZADD key score member [[score member] [score member] …] 将一个或多个 member 元素及其 score 值加入到有序集 key 当中。如果某个 member 已经是有序集的成员，那么更新这个 member 的 score 值，并通过重新插入这个 member 元素，来保证该 member 在正确有序的位置上。
ZSCORE ZSCORE key member 返回有序集 key 中，成员 member 的 score 值。
ZINCRBY ZINCRBY key increment member 为有序集 key 的成员 member 的 score 值加上增量 increment 。
ZCARD ZCARD key 当 key 存在且是有序集类型时，返回有序集的基数。 当 key 不存在时，返回 0
ZCOUNT ZCOUNT key min max 返回有序集 key 中， score 值在 min 和 max 之间(默认包括 score 值等于 min 或 max )的成员的数量。
ZRANGE ZRANGE key start stop [WITHSCORES] 返回有序集 key 中，指定区间内的成员。其中成员的位置按 score 值递增(从小到大)来排序。
ZREVRANGE ZREVRANGE key start stop [WITHSCORES] 返回有序集 key 中，指定区间内的成员。其中成员的位置按 score 值递减(从大到小)来排列。 具有相同 score 值的成员按字典序的逆序(reverse lexicographical order)排列。
ZRANGEBYSCORE ZRANGEBYSCORE key min max [WITHSCORES] [LIMIT offset count] 返回有序集 key 中，所有 score 值介于 min 和 max 之间(包括等于 min 或 max )的成员。有序集成员按 score 值递增(从小到大)次序排列。
ZREVRANGEBYSCORE ZREVRANGEBYSCORE key max min [WITHSCORES] [LIMIT offset count] 返回有序集 key 中， score 值介于 max 和 min 之间(默认包括等于 max 或 min )的所有的成员。有序集成员按 score 值递减(从大到小)的次序排列。
ZRANK ZRANK key member 返回有序集 key 中成员 member 的
ZREVRANK ZREVRANK key member 返回有序集 key 中成员 member 的排名。其中有序集成员按 score 值递减(从大到小)排序。
ZREM ZREM key member [member …] 排名以 0 为底，也就是说， score 值最大的成员排名为 0 。移除有序集 key 中的一个或多个成员，不存在的成员将被忽略
ZREMRANGEBYRANK ZREMRANGEBYRANK key start stop 移除有序集 key 中，指定排名(rank)区间内的所有成员。
ZREMRANGEBYSCORE ZREMRANGEBYSCORE key min max 移除有序集 key 中，所有 score 值介于 min 和 max 之间(包括等于 min 或 max )的成员。

1、存值方式

1
2
3
4
5
6
7
8
9
10
redis> ZRANGE salary 0 -1 WITHSCORES    # 测试数据
1) "tom"
2) "2000"
3) "peter"
4) "3500"
5) "jack"
6) "5000"

redis> ZSCORE salary peter              # 注意返回值是字符串
"3500"

redis参考文档

过期键删除策略

定时删除

定时删除是指在设置键的过期时间的同时，创建一个定时器，让定时器在键的过期时间来临时，立即执行对键的删除操作。

定时删除策略对内存是最友好的：通过使用定时器，定时删除策略可以保证过期键会尽可能快的被删除，并释放过期键所占用的内存。

定时删除策略的缺点是，他对CPU时间是最不友好的：再过期键比较多的情况下，删除过期键这一行为可能会占用相当一部分CPU时间。

除此之外，创建一个定时器需要用到Redis服务器中的时间事件。而当前时间事件的实现方式—-无序链表，查找一个事件的时间复杂度为O(N)—-并不能高效地处理大量时间事件。

惰性删除

惰性删除是指放任键过期不管，但是每次从键空间中获取键时，都检查取得的键是否过期，如果过期的话就删除该键，如果没有过期就返回该键。

惰性删除策略对CPU时间来说是最友好的，但对内存是最不友好的。如果数据库中有非常多的过期键，而这些过期键又恰好没有被访问到的话，那么他们也许永远也不会被删除。

定期删除

定期删除是指每隔一段时间，程序就对数据库进行一次检查，删除里面的过期键。

定期删除策略是前两种策略的一种整合和折中:

  • 定期删除策略每隔一段时间执行一次删除过期键操作，并通过限制删除操作执行的时长和频率来减少删除操作对CPU时间的影响。
  • 除此之外，通过定期删除过期键，定期删除策略有效地减少了因为过期键带来的内存浪费。

定期删除策略的难点是确定删除操作执行的时长和频率：

  • 如果删除操作执行的太频繁或者执行的时间太长，定期删除策略就会退化成定时删除策略，以至于将CPU时间过多的消耗在删除过期键上面。
  • 如果删除操作执行的太少，或者执行的时间太短，定期删除策略又会和惰性删除策略一样，出现浪费内存的情况。

LRU算法

LRU(least recently used)是一种缓存置换算法。即在缓存有限的情况下，如果有新的数据需要加载进缓存，则需要将最不可能被继续访问的缓存剔除掉。因为缓存是否可能被访问到没法做预测，所以基于如下假设实现该算法:

如果一个key经常被访问，那么该key的idle time应该是最小的。

(但这个假设也是基于概率，并不是充要条件,很明显,idle time最小的,甚至都不一定会被再次访问到)

这也就是LRU的实现思路。首先实现一个双向链表,每次有一个key被访问之后，就把被访问的key放到链表的头部。当缓存不够时,直接从尾部逐个摘除。

巧用LinkedHashMap完成lru算法

1. 认识LinkedHashMap

由于HashMap的迭代顺序并不是HashMap放置的顺序，也就是无序，这一缺点往往会带来困扰。
这个时候，LinkedHashMap就闪亮登场了，它虽然增加了时间和空间上的开销，但是通过维护一个运行于所有条目的双向链表，LinkedHashMap保证了元素迭代的顺序。

在LinkedHashMap中，只有accessOrder为true，即是访问顺序模式，才会put时对更新的Entry进行重新排序，而如果是插入顺序模式时，不会重新排序，这里的排序跟在HashMap中存储没有关系，只是指在双向链表中的顺序。

举个栗子：开始时，HashMap中有Entry1、Entry2、Entry3，并设置LinkedHashMap为访问顺序，则更新Entry1时，会先把Entry1从双向链表中删除，然后再把Entry1加入到双向链表的表尾，而Entry1在HashMap结构中的存储位置没有变化，对比图如下所示：

关 注 点 结 论
LinkedHashMap是否允许键值对为空 Key和Value都允许空
LinkedHashMap是否允许重复数据 Key重复会覆盖、Value允许重复
LinkedHashMap是否有序 有序
LinkedHashMap是否线程安全 非线程安全

2.LinkedHashMap基本数据结构

关于LinkedHashMap，先提两点：

1、LinkedHashMap可以认为是 HashMap+LinkedList，即它既使用HashMap操作数据结构，又使用LinkedList维护插入元素的先后顺序

2、LinkedHashMap的基本实现思想就是多态。

关于LinkedHashMap的定义：

1
2
3
4
5
6
public class LinkedHashMap<K,V>
    extends HashMap<K,V>
    implements Map<K,V>
{
    ...
}

3、基于LinkedHashMap实现LRU算法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
/**
 * @author Lauy
 * @date 2021/2/1
 */
public class LRUCacheJdkDemo<K, V> extends LinkedHashMap<K,V> {
    private int capacity; //缓存坑位

    public LRUCacheJdkDemo(int capacity) {
        super(capacity, 0.75F, false);
        this.capacity = capacity;
    }

    @Override
    protected boolean removeEldestEntry(Map.Entry<K, V> eldest) {
        return super.size() > capacity;
    }
}

测试

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
public static void main(String[] args) {
        LRUCacheJdkDemo lruCacheJdkDemo = new LRUCacheJdkDemo(3);
        lruCacheJdkDemo.put(1, 1);
        lruCacheJdkDemo.put(2, 2);
        lruCacheJdkDemo.put(3, 3);
        System.out.println(lruCacheJdkDemo.keySet());

        lruCacheJdkDemo.put(4, 1);
        System.out.println(lruCacheJdkDemo.keySet());

        lruCacheJdkDemo.put(3, 1);
        System.out.println(lruCacheJdkDemo.keySet());
        lruCacheJdkDemo.put(3, 1);
        System.out.println(lruCacheJdkDemo.keySet());
        lruCacheJdkDemo.put(3, 1);
        System.out.println(lruCacheJdkDemo.keySet());

        lruCacheJdkDemo.put(5, 1);
        System.out.println(lruCacheJdkDemo.keySet());
    }

存在疑问[https://www.jianshu.com/p/8f4f58b4b8ab],代码看是为false排序，源代码看是true重排序

输出：

当accessOrder：true，我们可以看出内部当空间溢出时进行了淘汰

[1, 2, 3]
[2, 3, 4]
[2, 4, 3]
[2, 4, 3]
[2, 4, 3]
[4, 3, 5]

相比上面的，不仅进行了淘汰还内部排序处理了

accessOrder：false
[1, 2, 3]
[2, 3, 4]
[2, 3, 4]
[2, 3, 4]
[2, 3, 4]
[3, 4, 5]

手写LRU

1.定义Node类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class Node<K, V> {
    K key;
    V value;
    Node<K, V> prev;
    Node<K, V> next;

    public Node() {
        this.prev = this.next = null;
    }

    public Node(K key, V value) {
        this.key = key;
        this.value = value;
        this.prev = this.next = null;
    }
}

2.定义双向链表

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
class DoubleLinkedList<K, V> {
    // 头节点
    Node<K, V> head;
    // 尾节点
    Node<K, V> tail;

    // 2.1 构造方法
    public DoubleLinkedList() {
        head = new Node<>();
        tail = new Node<>();
        head.next = tail;
        tail.prev = head;
    }

    // 2.2 添加到头
    public void addHead(Node<K, V> node) {
        node.next = head.next;
        node.prev = head;
        head.next.prev = node;
        head.next = node;
    }

    // 2.3 删除节点
    public void removeNode(Node<K, V> node) {
        node.next.prev = node.prev;
        node.prev.next = node.next;
        node.prev = null;
        node.next = null;
    }

    // 2.3 获得最后一个节点
    public Node getLast() {
        return tail.prev;
    }
}

3.初始化LRU类

1
2
3
4
5
6
7
8
9
10
11
12
13
// 坑位：空间大小
private int cacheSize;
Map<Integer, Node<Integer, Integer>> map;
DoubleLinkedList<Integer, Integer> doubleLinkedList;

public LRUCacheDemo(int cacheSize) {
    // 坑位
    this.cacheSize = cacheSize;
    // map负责查找，构建一个虚拟的双向链表，它里面安装的就是一个个node节点，作为数据载体
    map = new HashMap<>();
    // 双向链表
    doubleLinkedList = new DoubleLinkedList<>();
}

4.定义方法

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
public int get(int key) {
    if (!map.containsKey(key)) {
        return -1;
    }

    Node<Integer, Integer> node = map.get(key);
    doubleLinkedList.removeNode(node);
    doubleLinkedList.addHead(node);

    return node.value;
}

/**
    * 保存/更新方法
    *
    * @param key
    * @param value
    */
public void put(int key, int value) {
    if (map.containsKey(key)) {
        // update
        Node<Integer, Integer> node = map.get(key);
        node.value = value;
        map.put(key, node);

        doubleLinkedList.removeNode(node);
        doubleLinkedList.addHead(node);
    } else {
        if (map.size() == cacheSize) {
            // 坑位满了
            Node<Integer, Integer> lastNode = doubleLinkedList.getLast();
            map.remove(lastNode.key);
            doubleLinkedList.removeNode(lastNode);
        }
        // 新增
        Node<Integer, Integer> newNode = new Node<>(key, value);
        map.put(key, newNode);
        doubleLinkedList.addHead(newNode);
    }
}

5.完整代码

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
/**
 * LRU算法
 *
 * @author Lauy
 * @date 2021/2/1
 */
public class LRUCacheDemo {

    public LRUCacheDemo(int cacheSize) {
        // 坑位
        this.cacheSize = cacheSize;
        // map负责查找，构建一个虚拟的双向链表，它里面安装的就是一个个node节点，作为数据载体
        map = new HashMap<>();
        // 双向链表
        doubleLinkedList = new DoubleLinkedList<>();
    }

    class Node<K, V> {
        K key;
        V value;
        Node<K, V> prev;
        Node<K, V> next;

        public Node() {
            this.prev = this.next = null;
        }

        public Node(K key, V value) {
            this.key = key;
            this.value = value;
            this.prev = this.next = null;
        }
    }

    class DoubleLinkedList<K, V> {
        Node<K, V> head;
        Node<K, V> tail;

        // 2.1 构造方法
        public DoubleLinkedList() {
            head = new Node<>();
            tail = new Node<>();
            head.next = tail;
            tail.prev = head;
        }

        // 2.2 添加到头
        public void addHead(Node<K, V> node) {
            node.next = head.next;
            node.prev = head;
            head.next.prev = node;
            head.next = node;
        }

        // 2.3 删除节点
        public void removeNode(Node<K, V> node) {
            node.next.prev = node.prev;
            node.prev.next = node.next;
            node.prev = null;
            node.next = null;
        }

        // 2.3 获得最后一个节点
        public Node getLast() {
            return tail.prev;
        }
    }

    private int cacheSize;
    Map<Integer, Node<Integer, Integer>> map;
    DoubleLinkedList<Integer, Integer> doubleLinkedList;


    public int get(int key) {
        if (!map.containsKey(key)) {
            return -1;
        }

        Node<Integer, Integer> node = map.get(key);
        doubleLinkedList.removeNode(node);
        doubleLinkedList.addHead(node);

        return node.value;
    }

    /**
     * 保存/更新方法
     *
     * @param key
     * @param value
     */
    public void put(int key, int value) {
        if (map.containsKey(key)) {
            // update
            Node<Integer, Integer> node = map.get(key);
            node.value = value;
            map.put(key, node);

            doubleLinkedList.removeNode(node);
            doubleLinkedList.addHead(node);
        } else {
            if (map.size() == cacheSize) {
                // 坑位满了
                Node<Integer, Integer> lastNode = doubleLinkedList.getLast();
                map.remove(lastNode.key);
                doubleLinkedList.removeNode(lastNode);
            }
            // 新增
            Node<Integer, Integer> newNode = new Node<>(key, value);
            map.put(key, newNode);
            doubleLinkedList.addHead(newNode);
        }
    }
}

6.测试

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public static void main(String[] args) {
    LRUCacheDemo lruCacheDemo = new LRUCacheDemo(3);

    lruCacheDemo.put(1, 1);
    lruCacheDemo.put(2, 2);
    lruCacheDemo.put(3, 3);
    System.out.println(lruCacheDemo.map.keySet());

    lruCacheDemo.put(4, 1);
    System.out.println(lruCacheDemo.map.keySet());

    lruCacheDemo.put(3, 1);
    System.out.println(lruCacheDemo.map.keySet());
    lruCacheDemo.put(3, 1);
    System.out.println(lruCacheDemo.map.keySet());
    lruCacheDemo.put(3, 1);
    System.out.println(lruCacheDemo.map.keySet());

    lruCacheDemo.put(5, 1);
    System.out.println(lruCacheDemo.map.keySet());
}

7.输出结果

1
2
3
4
5
6
[1, 2, 3]
[2, 3, 4]
[2, 3, 4]
[2, 3, 4]
[2, 3, 4]
[3, 4, 5]

redis分布式锁

一般服务也是多实例部署，如果多个实例需要操作同一份数据库的数据的时候，自然而然的引入了分布式锁。

先定义两个所需的redis操作类和常量

1
2
3
4
5
6
7
8
9
@Resource
private StringRedisTemplate stringRedisTemplate;
@Resource
private RedissonClient redissonClient;

// 锁
private final String LOCK_KEY = "test-redis-lock";
// redis的key
private final String REDIS_LOCK = "redisLock";

基本配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
/**
 * @author coderblue
 */
@Configuration
public class RedissonConfig {

    @Bean
    public RedissonClient redissonClient() {
        // Cluster集群
        // Config config = new Config();
        // config.useClusterServers()
        //         .setScanInterval(2000)
        //         .addNodeAddress("redis://127.0.0.1:6379");
        //
        // RedissonClient redisson = Redisson.create(config);
        // return redisson;

        // 单机模式，不然就会启动报错：非集群
        Config config = new Config();
        config.useSingleServer().setAddress("redis://127.0.0.1:6379").setDatabase(0);
        return  Redisson.create(config);
    }
}

引入POM依赖

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
<!-- 引入操作redis -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
 <!-- spring2.X集成redis所需common-pool2-->
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-pool2</artifactId>
    <version>2.6.0</version>
</dependency>
<!-- 引入redisson -->
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.10.6</version>
</dependency>

使用synchronized实现(仅限单机下)

1
2
3
4
5
6
7
8
9
10
11
12
13
// 锁住此class实例
synchronized (this) {
    // 将redis存的key当成锁
    Integer count = Integer.parseInt(stringRedisTemplate.opsForValue().get(this.REDIS_LOCK));
    if (count > 0) {
        int realCount = count - 1;
        // jedis.set(key, value)
        stringRedisTemplate.opsForValue().set(this.REDIS_LOCK, realCount + "");
        System.out.println("扣减成功，剩余库存：" + realCount);
    } else {
        System.out.println("扣减失败，库存不够");
    }
}

synchronized 和 ReentrantLock的选择：
synchronized属于JVM层面，
ReentrantLock属于类，而且可以通过 tryLock() 设置抢占锁的时间，避免因为无法抢到锁导致的故障。

redis实现分布式锁

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
// 版本号
String clientId = UUID.randomUUID().toString().replace("-", "");
// 没有不存在此key就进行操作返回true，否则不操作，返回false。jedis.setnx(key, value)
// 设置10s过期时间
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(this.LOCK_KEY, clientId, 10, TimeUnit.SECONDS);
if (!result) {
    return "error";
}
try {
    Integer count = Integer.parseInt(stringRedisTemplate.opsForValue().get(this.REDIS_LOCK));
    if (count > 0) {
        int realCount = count - 1;
        stringRedisTemplate.opsForValue().set(this.REDIS_LOCK, realCount + "");
        System.out.println("扣减成功，剩余库存：" + realCount);
    } else {
        System.out.println("扣减失败，库存不够");
    }
} finally {
    if (Objects.equals(clientId, stringRedisTemplate.opsForValue().get(this.LOCK_KEY))) {
        // 解决redis主从架构锁失效问题：
        // 符合对应的版本号才可以释放锁,不然很容易因为设置的过期时间比较短，
        // 还没进行到这锁就被释放了。然后第二个已经进来获取锁后才调用这个方法。这样无限循环下去，
        // 导致永远都是前一个的把后面进来加的锁释放掉了，就永久失效！
        // =========================
        // 当主节点挂掉时，从节点会取而代之，但客户端无明显感知。当客户端 A 成功加锁，指令还未同步，
        // 此时主节点挂掉，从节点提升为主节点，新的主节点没有锁的数据，当客户端 B 加锁时就会成功。
        stringRedisTemplate.delete(this.LOCK_KEY);
    }
}

这个基本可以满足我们需求，但是希望在设置过期时间的时候，我们可以判断下如果快要过期了，但是还没结束，那么我们可以延长时间，比如延长1/3的时间。

使用redisson实现分布式锁

1.redisson基本介绍

2.主要代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
RLock redissonLock = redissonClient.getLock(this.LOCK_KEY);

try {
    // 给redis加把锁，过期时间30 S
    redissonLock.lock(30, TimeUnit.SECONDS);
    Integer count = Integer.parseInt(stringRedisTemplate.opsForValue().get(this.REDIS_LOCK));
    if (count > 0) {
        int realCount = count - 1;
        stringRedisTemplate.opsForValue().set(this.REDIS_LOCK, realCount + "");
        System.out.println("扣减成功，剩余库存：" + realCount);
    } else {
        System.out.println("扣减失败，库存不够");
    }
} finally {
    redissonLock.unlock();
}

但是在超高并发情况下，redisson调用unlock()方法时可能出现如下异常：

我们可以通过增加判断来规避

1
2
3
4
// 是否还是锁定状态 和 是否是当前执行线程的锁
if(redissonLock.isLocked() && redissonLock.isHeldByCurrentThread()) {
    redissonLock.unlock();
}

这边我再延伸一个，使用lua脚本编写：

1
2
3
4
5
6
7
8
9
10
11
// 获取锁
Jedis jedis = RedisUtils.getJedis();

String script = "if redis.call("get",KEYS[1]) == ARGV[1] " +
    "then " +
    "return redis.call("del",KEYS[1]) " +
    "else" +
    " return 0" +
    " end";
// 执行的时候，需要把前面的随机数作为argv[1] 的值传进去，把cache_key作为keys[1]的值传进去。
Object obj = jedis.eval(script, Collections.singletonList(REDIS_LOCK), Collections.singletonList(value));
1
2
// 释放锁
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";

补充：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
for (int i = 0; i < 3; i++) {
    new Thread(() -> {
        RLock lock = redissonClient.getLock("POINT_ACCOUNT_LOCK" + -112);;
        try {
            // 第三个线程只尝试等6秒中，所以拿不到锁，然后调用unlock()方法报错
            // java.lang.IllegalMonitorStateException: attempt to unlock lock, not locked by current thread by node id
            // 可以通过下述finally中判断规避
            // boolean b = lock.tryLock(6L, TimeUnit.SECONDS); // 尝试获取锁，时间为6秒钟
            // lock.lock(); // 直到此线程释放锁资源
            lock.lock(3, TimeUnit.SECONDS); //锁三秒，到期自动释放
            // if (b) {
                LocalDateTime now = LocalDateTime.now();
                System.out.println(Thread.currentThread().getName() + ":释放锁了 - " + now);
                TimeUnit.SECONDS.sleep(5L);
                System.out.println(Thread.currentThread().getName() + "执行耗时：" + now.until(LocalDateTime.now(), ChronoUnit.SECONDS));
            // }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (lock.isLocked() && lock.isHeldByCurrentThread()){
                lock.unlock();
            }
        }
}).start();

redis主从分布

主从模式

1.基本原理

主从复制模式中包含一个主数据库实例（master）与一个或多个从数据库实例（slave）。主负责写，并且将数据复制到其它的 slave 节点，从节点负责读。所有的读请求全部走从节点。这样也可以很轻松实现水平扩容，支撑读高并发。

2.工作机制

  • slave启动后，向master发送SYNC命令，master接收到SYNC命令后通过bgsave保存快照（即上文所介绍的RDB持久化），并使用缓冲区记录保存快照这段时间内执行的写命令
  • master将保存的快照文件发送给slave，并继续记录执行的写命令
  • slave接收到快照文件后，加载快照文件，载入数据
  • master快照发送完后开始向slave发送缓冲区的写命令，slave接收命令并执行，完成复制初始化
  • 此后master每次执行一个写命令都会同步发送给slave，保持master与slave之间数据的一致性

3.主从复制的优缺点

优点：

  • master能自动将数据同步到slave，可以进行读写分离，分担master的读压力
  • master、slave之间的同步是以非阻塞的方式进行的，同步期间，客户端仍然可以提交查询或更新请求

缺点：

  • 不具备自动容错与恢复功能，master或slave的宕机都可能导致客户端请求失败，需要等待机器重启或手动切换客户端IP才能恢复
  • master宕机，如果宕机前数据没有同步完，则切换IP后会存在数据不一致的问题

哨兵模式

1.基本原理

主从模式下，当主服务器宕机后，需要手动把一台从服务器切换为主服务器。然后在哨兵模式下，master宕机，哨兵会 自动选举master并将其他的slave指向新的master。

在主从模式下，redis同时提供了哨兵命令 redis-sentinel，哨兵是一个独立的进程，作为进程，它会独立运行。其原理是哨兵进程向所有的redis机器发送命令，等待Redis服务器响应，从而监控运行的多个Redis实例。

哨兵可以有多个，一般为了便于决策选举，使用奇数个哨兵。哨兵可以和redis机器部署在一起，也可以部署在其他的机器上。多个哨兵构成一个哨兵集群，哨兵直接也会相互通信，检查哨兵是否正常运行，同时发现master宕机哨兵之间会进行决策选举新的master

2.作用

哨兵模式的作用:

  • 通过发送命令，让Redis服务器返回监控其运行状态，包括主服务器和从服务器;

  • 当哨兵监测到master宕机，会自动将slave切换到master，然后通过发布订阅模式通过其他的从服务器，修改配置文件，让它们切换主机;

  • 然而一个哨兵进程对Redis服务器进行监控，也可能会出现问题，为此，我们可以使用多个哨兵进行监控。各个哨兵之间还会进行监控，这样就形成了多哨兵模式。

假设master宕机，sentinel 1先检测到这个结果，系统并不会马上进行 failover(故障转移)选出新的master，仅仅是sentinel 1主观的认为master不可用，这个现象成为 主观下线。当后面的哨兵也检测到主服务器不可用，并且数量达到一定值时，那么哨兵之间就会进行一次投票，投票的结果由sentinel 1发起，进行 failover 操作。切换成功后，就会通过 发布订阅模式，让各个哨兵把自己监控的从服务器实现切换主机，这个过程称为 客观下线。这样对于客户端而言，一切都是透明的。

优缺点

优点

  • 哨兵模式是 基于主从模式的，所有主从的优点，哨兵模式都具有。
  • 主从可以 自动切换，系统更健壮，可用性更高。

缺点

  • 具有主从模式的缺点，每台机器上的数据是一样的，内存的可用性较低。
  • Redis较难支持在线扩容，在集群容量达到上限时在线扩容会变得很复杂。

Cluster集群模式

1.基本介绍

Redis 的哨兵模式基本已经可以实现高可用，读写分离 ，但是在这种模式下每台 Redis 服务器都存储相同的数据，很浪费内存，所以在redis3.0上加入了 Cluster 集群模式，实现了 Redis 的分布式存储，对数据进行分片，也就是说每台 Redis 节点上存储不同的内容；

  • 自动将数据进行分片，每个 master 上放一部分数据
  • 提供内置的高可用支持，部分 master 不可用时，还是可以继续工作的

在 redis cluster 架构下，每个 redis 要放开 两个端口号，比如一个是 6379，另外一个就是 加1w 的端口号，比如 16379。

16379 端口号是用来进行节点间通信的，也就是 cluster bus 的东西，cluster bus 的通信，用来进行故障检测、配置更新、故障转移授权。cluster bus 用了另外一种二进制的协议，gossip 协议，用于节点间进行高效的数据交换，占用更少的网络带宽和处理时间。

对客户端来说，整个cluster被看做是一个整体，客户端可以连接任意一个node进行操作，就像操作单一Redis实例一样，当客户端操作的key没有分配到该node上时，Redis会返回转向指令，指向正确的node，这有点儿像浏览器页面的302 redirect跳转。

3.总结

数据量比较大，QPS要求较高的时候使用。 Redis Cluster是Redis 3.0以后才正式推出，时间较晚，目前能证明在大规模生产环境下成功的案例还不是很多，需要时间检验。

redis发布订阅模式

1.基本介绍

Redis 发布订阅(pub/sub)是一种消息通信模式：发送者(pub)发送消息，订阅者(sub)接收消息。一个redis客户端可以订阅任意多个频道channel，一个频道也可以被多个客户端订阅。

2.代码实现

引入POM依赖

1
2
3
4
5
6
7
8
9
10
11
12
13
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <optional>true</optional>
</dependency>

定义通道名

1
2
3
4
5
6
7
8
9
10
11
/**
 * @author coderblue
 */
public interface Const {

    /**
     * 通道名称
     */
    String CHANNEL = "rides_channel";

}

发布者

定义向通道发送消息方法

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
/**
 * @author coderblue
 */
public interface PublisherService {
    /**
     * 发布消息
     * @param params
     * @return
     */
    String pushMsg(String params);
}

------------------------------------------

@Slf4j
@Service
public class PublisherServiceImpl implements PublisherService {
    @Autowired
    private RedisService redisService;

    @Override
    public String pushMsg(String params) {
        log.info(" 又开始发布消息 .......... ");
        // 直接使用convertAndSend方法即可向指定的通道发布消息
        new Thread(() -> {
            for (int i = 0; i < 5; i++) {
                redisService.sendChannelMess(Const.CHANNEL, System.currentTimeMillis() + "：" + "redis消息队列-线程一");
            }
        }, "线程一测试").start();

        new Thread(() -> {
            for (int i = 0; i < 5; i++) {
                redisService.sendChannelMess(Const.CHANNEL, System.currentTimeMillis() + "：" + "redis消息队列-线程二");
            }
        }, "线程二测试").start();
        return "success";
    }
}

实际通道发送消息方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@Component
public class RedisService {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    /**
     * 向通道发送消息的方法
     * @param channel
     * @param message
     */
    public void sendChannelMess(String channel, String message) {
        stringRedisTemplate.convertAndSend(channel, message);
    }
}

订阅者

redis 监听配置

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
/***
 *  redis 监听配置
 * @author coderblue
 */
@Configuration
public class RedisSubListenerConfig {

    /**
     * 初始化监听器
     *
     * @param connectionFactory
     * @param listenerAdapter
     * @return
     */
    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
                                            MessageListenerAdapter listenerAdapter) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        // new PatternTopic("这里是监听的通道的名字") 通道要和发布者发布消息的通道一致
        container.addMessageListener(listenerAdapter, new PatternTopic(Const.CHANNEL));
        return container;
    }

    /**
     * 绑定消息监听者和接收监听的方法
     *
     * @param redisReceiver
     * @return
     */
    @Bean
    MessageListenerAdapter listenerAdapter(RedisReceiver redisReceiver) {
        // redisReceiver 消息接收者
        // receiveMessage 消息接收后的方法
        return new MessageListenerAdapter(redisReceiver, "receiveMessage");
    }


    @Bean
    StringRedisTemplate template(RedisConnectionFactory connectionFactory) {
        return new StringRedisTemplate(connectionFactory);
    }

    /**
     * 注册订阅者
     *
     * @param latch
     * @return
     */
    @Bean
    RedisReceiver receiver(CountDownLatch latch) {
        return new RedisReceiver(latch);
    }

    /**
     * 计数器，用来控制线程
     *
     * @return
     */
    @Bean
    CountDownLatch latch() {
        // 指定了计数的次数 1
        return new CountDownLatch(1);
    }

}

消息接收方法

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
/***
 * 消息接收者（订阅者）  需要注入到springboot中
 * @author coderblue
 */
@Slf4j
@Component
public class RedisReceiver {

    private CountDownLatch latch;

    @Autowired
    public RedisReceiver(CountDownLatch latch) {
        this.latch = latch;
    }

    /**
     *  收到通道的消息之后执行的方法
     * @param message
     */
    public void receiveMessage(String message) {
        //这里是收到通道的消息之后执行的方法
        log.info("我收到通道里你发的的消息了....." + message);

        latch.countDown();
    }
}

Springboot使用缓存

使用redis存储

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@Resource
private StringRedisTemplate stringRedisTemplate;
private final String REDIS_LOCK = "redisLock";

// jedis.set(key, value)
stringRedisTemplate.opsForValue().set(this.REDIS_LOCK, realCount + "");

// 没有不存在此key就进行操作返回true，否则不操作，返回false。jedis.setnx(key, value)
stringRedisTemplate.opsForValue().setIfAbsent(this.REDIS_LOCK, value, 10, TimeUnit.MINUTES);

// 根据键名获取值
stringRedisTemplate.opsForValue().get(this.REDIS_LOCK)

// 释放锁
stringRedisTemplate.delete(this.LOCK_KEY);

redis存储List类型

1
2
3
4
5
6
7
8
// 根据key删除左边第一个元素
redisTemplate.opsForList().leftPop(POINT_ACTION_VALID_PERIOD);

// 在此key的list中添加一个元素
redisTemplate.opsForList().rightPush(POINT_ACTION_VALID_PERIOD, creditActions);

// 获得此key的list第一个元素 
redisTemplate.opsForList().index(POINT_ACTION_VALID_PERIOD, 0)

使用redis + cache缓存

1.添加POM文件依赖

1
2
3
4
5
6
7
8
9
10
11
12
<!-- redis -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
    <version>2.2.1.RELEASE</version>
</dependency>
<!-- spring2.X集成redis所需common-pool2-->
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-pool2</artifactId>
    <version>2.6.0</version>
</dependency>

2.yml文件配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
spring:
  # 缓存
  redis:
    database: 0
    host: 127.0.0.1
    lettuce:
      pool:
        max-active: 8   #最大连接数据库连接数,设 0 为没有限制
        max-idle: 8     #最大等待连接中的数量,设 0 为没有限制
        max-wait: -1ms  #最大建立连接等待时间。如果超过此时间将接到异常。设为-1表示无限制。
        min-idle: 0     #最小等待连接中的数量,设 0 为没有限制
      shutdown-timeout: 100ms
    password: ''
    port: 6379

3.redis序列化配置类

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
@EnableCaching //开启缓存
@Configuration  //配置类
public class RedisConfig extends CachingConfigurerSupport {

    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        StringRedisSerializer 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);
        // 取值多双引号问题: value序列化用的是jackson2JsonRedisSerializer，改成StringRedisSerializer就行。
        //key序列化方式
        template.setKeySerializer(redisSerializer);
        //value序列化
        template.setValueSerializer(redisSerializer);
        //value hashmap序列化
        template.setHashValueSerializer(redisSerializer);
        template.setHashValueSerializer(redisSerializer);
        template.afterPropertiesSet();
        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;
    }
}

4.常用注解缓存

@Cacheable：对其返回结果进行缓存，下次请求时，如果缓存存在，则直接读取缓存数据返回；如果缓存不存在，则执行方法，并把返回的结果存入缓存中。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@GetMapping
@Cacheable(value = "UserInfo", key = "'selectUserInfo'")
@ApiOperation(value = "查询", notes = "mongo查询")
public Result find() {
    List<UserInfo> all = mongoTemplate.findAll(UserInfo.class);
    log.info("用户信息：" + all);
    return Result.success().data("findAll", all);
}

注意 key = "'selectUserInfo'"，需要加单引号成字符串，不然就会寻找有无 selectUserInfo 此值，没有就报错！！！
======================================================
存储如下：缓存存放的命名空间UserInfo + "::" + key
127.0.0.1:6379> keys *
1) "UserInfo::selectUserInfo"
127.0.0.1:6379>

@CachePut：每次都会执行，并将结果存入指定的缓存中。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@PutMapping("/update")
@ApiOperation("更新对象")
@CachePut(value = "UserInfo", key = "'selectUserInfo'")
public Result update(UserInfo userInfo) {
    Query query = new Query(Criteria.where("id").is(userInfo.getId()));
    Update update = new Update().set("age", userInfo.getAge()).set("name", userInfo.getName());
    //更新查询返回结果集的第一条
    UpdateResult result = mongoTemplate.updateFirst(query, update, UserInfo.class);
    // 将修改后的返回值作为value
    UserInfo info = this.findByName(userInfo.getName());
    return Result.success().data("data", info);
    // 更新查询返回结果集的所有
    // mongoTemplate.updateMulti(query,update,UserInfo.class);
}

@CacheEvict：使用该注解标志的方法，会 清空指定的缓存。一般用在 更新或者删除方法上

1
2
3
4
5
6
7
@ApiOperation("根据id删除对象")
@DeleteMapping("deleteById")
@CacheEvict(value = "UserInfo", key = "#id")
public void deleteById(Integer id) {
    Query query = new Query(Criteria.where("id").is(id));
    mongoTemplate.remove(query, UserInfo.class);
}

使用ehcache缓存

1.引入POM文件依赖

1
2
3
4
5
6
7
8
9
10
11
<!--开启 cache 缓存 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-cache</artifactId>
</dependency>
<!-- ehcache缓存 -->
<dependency>
    <groupId>net.sf.ehcache</groupId>
    <artifactId>ehcache</artifactId>
    <version>2.10.6</version>
</dependency>

2.yml文件配置

1
2
3
4
5
6
spring:
  # 缓存
  cache:
    type: ehcache
    ehcache:
      config: classpath:ehcache.xml

3.ehcache.xml文件配置

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
<?xml version="1.0" encoding="UTF-8"?>
<ehcache xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:noNamespaceSchemaLocation="http://ehcache.org/ehcache.xsd">
	<!-- <diskStore path="java.io.tmpdir/ehcache" /> -->
	<defaultCache maxEntriesLocalHeap="10000" eternal="false"
		timeToIdleSeconds="120" timeToLiveSeconds="120"
		memoryStoreEvictionPolicy="LRU" />

        <defaultCache
            name="后续需删除"
            eternal="false" <!--意味着该缓存会死亡-->
            maxElementsInMemory="900"<!--缓存的最大数目-->
            overflowToDisk="false" <!--内存不足时,是否启用磁盘缓存,如果为true则表示启动磁盘来存储，如果为false则表示不启动磁盘-->
            diskPersistent="false"
            timeToIdleSeconds="0"  <!--当缓存的内容闲置多少时间销毁-->
            timeToLiveSeconds="60" <!--当缓存存活多少时间销毁（单位是秒，如果我们想设置2分钟的缓存存活时间，那么这个值我们需要设置120）-->
            memoryStoreEvictionPolicy="LRU" /> <!--自动销毁策略-->

	<!-- 登录记录缓存 锁定10分钟 -->
	<cache name="passwordRetryCache" maxEntriesLocalHeap="2000"
		eternal="false" timeToIdleSeconds="3600" timeToLiveSeconds="0"
		overflowToDisk="false" statistics="true">
	</cache>
	<!-- session缓存 -->
	<cache name="sessionCache" maxEntriesLocalHeap="10000"
		overflowToDisk="false" eternal="false" diskPersistent="false"
		timeToLiveSeconds="0" timeToIdleSeconds="0" statistics="true" />
	<!-- 认证缓存 -->
	<cache name="AuthenticationCache" maxElementsInMemory="10000"
		eternal="false" timeToIdleSeconds="600" timeToLiveSeconds="1200"
		overflowToDisk="true" />
	<!-- 授权缓存 -->
	<cache name="AuthorizationCache" maxElementsInMemory="10000"
		eternal="false" timeToIdleSeconds="600" timeToLiveSeconds="1200"
		overflowToDisk="true" />

	<!-- 对应查询的UserInfo的返回结果集 -->
	<cache name="UserInfo" maxElementsInMemory="1000"
		   eternal="false" timeToIdleSeconds="600" timeToLiveSeconds="1200"
		   overflowToDisk="true" />

</ehcache>

后续缓存同redis操作雷同,注意 value = “UserInfo” 是需要 ehcache.xml 中有对应的 name

1
2
3
4
5
6
7
8
@GetMapping
@Cacheable(value = "UserInfo", key = "'selectUserInfo'")
@ApiOperation(value = "查询", notes = "mongo查询")
public Result find() {
    List<UserInfo> all = mongoTemplate.findAll(UserInfo.class);
    log.info("用户信息：" + all);
    return Result.success().data("findAll", all);
}

问题

RedisTemplate取值多双引号问题

原因是value序列化用的是jackson2JsonRedisSerializer，改成StringRedisSerializer就行。因为它会自动为String类型的键和值添加双引号，这也是Jackson2JsonRedisSerializer特性，

两种序列化方式：

  1. Jackson2JsonRedisSerializer
  2. StringRedisSerializer （推荐）

修改后

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
RedisTemplate<String, Object> template = new RedisTemplate<String, Object>();
template.setConnectionFactory(factory);
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);
StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
// key采用String的序列化方式
template.setKeySerializer(stringRedisSerializer);
// hash的key也采用String的序列化方式
template.setHashKeySerializer(stringRedisSerializer);
// value序列化方式采用jackson
template.setValueSerializer(stringRedisSerializer);
// hash的value序列化方式采用jackson
template.setHashValueSerializer(stringRedisSerializer);
template.afterPropertiesSet();
return template;

链接

Github仓库传送门

redis参考文档