【狂神說Java】Redis筆記以及拓展

一、Redis 入門

Redis為什么單線程還這么快？

誤區1：高性能的服務器一定是多線程的？
誤區2：多線程（CPU上下文會切換！）一定比單線程效率高！
核心：Redis是將所有的數據放在內存中的，所以說使用單線程去操作效率就是最高的，多線程（CPU上下文會切換：耗時的操作！），對于內存系統來說，如果沒有上下文切換效率就是最高的，多次讀寫都是在一個CPU上的，在內存存儲數據情況下，單線程就是最佳的方案。

（1）完全基于內存，數據存在內存中，絕大部分請求是純粹的內存操作，非常快速，跟傳統的磁盤文件數據存儲相比，避免了通過磁盤IO讀取到內存這部分的開銷。

（2）數據結構簡單，對數據操作也簡單。Redis中的數據結構是專門進行設計的，每種數據結構都有一種或多種數據結構來支持。Redis正是依賴這些靈活的數據結構，來提升讀取和寫入的性能。

（3）采用單線程，省去了很多上下文切換的時間以及CPU消耗，不存在競爭條件，不用去考慮各種鎖的問題，不存在加鎖釋放鎖操作，也不會出現死鎖而導致的性能消耗。

（4）使用基于IO多路復用機制的線程模型，可以處理并發的鏈接。

多個 Socket 可能會產生不同的操作，每個操作對應不同的文件事件，但是IO多路復用程序會監聽多個Socket，將Socket產生的事件放入隊列中排隊，事件分派器每次從隊列中取出一個事件，把該事件交給對應的事件處理器進行處理。

Redis客戶端對服務端的每次調用都經歷了發送命令，執行命令，返回結果三個過程。其中執行命令階段，由于Redis是單線程來處理命令的，所有每一條到達服務端的命令不會立刻執行，所有的命令都會進入一個隊列中，然后逐個被執行。并且多個客戶端發送的命令的執行順序是不確定的。但是可以確定的是不會有兩條命令被同時執行，不會產生并發問題，這就是Redis的單線程基本模型。

多路I/O復用模型是利用 select、poll、epoll 可以同時監察多個流的 I/O 事件的能力，在空閑的時候，會把當前線程阻塞掉，當有一個或多個流有 I/O 事件時，就從阻塞態中喚醒，然后程序就會輪詢一遍所有的流（epoll 是只輪詢那些真正發出了事件的流），并且依次順序的處理就緒的流，這種做法就避免了大量的無用操作。
這里“多路”指的是多個網絡連接，“復用”指的是復用同一個線程。采用多路 I/O 復用技術可以讓單個線程高效的處理多個客戶端的網絡IO連接請求（盡量減少網絡 IO 的時間消耗）

為什么Redis是單線程？

這里我們強調的單線程，指的是網絡請求模塊使用一個線程來處理，即一個線程處理所有網絡請求，其他模塊仍用了多個線程。

那為什么使用單線程呢？官方答案是：因為CPU不是Redis的瓶頸，Redis的瓶頸最有可能是機器內存或者網絡帶寬。既然單線程容易實現，而且CPU不會成為瓶頸，那就順理成章地采用單線程的方案了。

但是，我們使用單線程的方式是無法發揮多核CPU 性能，不過我們可以通過在單機開多個Redis 實例來解決這個問題

二、五種數據類型 + 三種特殊數據類型

Redis是一個開源（BSD許可），內存存儲的數據結構服務器，可用作數據庫，高速緩存和消息隊列代理。它支持字符串、哈希表、列表、集合、有序集合，位圖，hyperloglogs等數據類型。內置復制、Lua腳本、LRU收回、事務以及不同級別磁盤持久化功能，同時通過Redis Sentinel提供高可用，通過Redis Cluster提供自動分區。

Redis-key

在redis中無論什么數據類型，在數據庫中都是以key-value形式保存，通過進行對Redis-key的操作，來完成對數據庫中數據的操作。

下面學習的命令：

keys * : 查看當前數據庫所有key

set key value: 設置插入的鍵值對

exists key：判斷鍵是否存在

del key：刪除鍵值對

move key db：將鍵值對移動到指定數據庫

expire key second：設置鍵值對的過期時間

type key：查看value的數據類型

ttl key: key的過期時間

當前key沒有設置過期時間，所以會返回-1.
當前key有設置過期時間，而且key已經過期，所以會返回-2.
當前key有設置過期時間，且key還沒有過期，故會返回key的正常剩余時間.

RENAME key newkey修改 key 的名稱

RENAMENX key newkey僅當 newkey 不存在時，將 key 改名為 newkey

更多命令：redis命令手冊

String(字符串)

在這里插入圖片描述

127.0.0.1:6379> set key1 vl  #設置值
OK
127.0.0.1:6379> get key1   #獲取值
"vl"
127.0.0.1:6379> keys *   #獲得所有的key
1) "key1"
2) "name"
3) "age"
127.0.0.1:6379> exists key1  #判斷某一個key是否存在
(integer) 1
127.0.0.1:6379> APPEND key1 "hello"  #追加字符串，如果當前key不存在就相當于setkey
(integer) 7
127.0.0.1:6379> get key1
"vlhello"
127.0.0.1:6379> strlen key1  #獲取字符串的長度
(integer) 7
127.0.0.1:6379> append key1 ",zhangsan"
(integer) 16
127.0.0.1:6379> get key1
"vlhello,zhangsan"
127.0.0.1:6379> strlen key1
(integer) 16
127.0.0.1:6379> 
##########################################################
#步長 i+=
127.0.0.1:6379> set views 0  #初始瀏覽量為0
OK
127.0.0.1:6379> get views
"0"
127.0.0.1:6379> incr views  #使值加1（自增1） 
(integer) 1
127.0.0.1:6379> incr views
(integer) 2
127.0.0.1:6379> decr views  #使值減一（自減1）
(integer) 1
127.0.0.1:6379> decr views
(integer) 0
127.0.0.1:6379> decr views
(integer) -1
127.0.0.1:6379> get views
"-1"
127.0.0.1:6379> incrby views 10  #設置步長，指定增量
(integer) 9
127.0.0.1:6379> incrby views 10
(integer) 19
127.0.0.1:6379> decrby views 5  #設置步長，減少增量
(integer) 14
#############################################################
#字符串范圍 ： range
127.0.0.1:6379> get key1
"vlhello,zhangsan"
127.0.0.1:6379> getrange key1 2 5  #截取字符串 [2,5]
"hell"
127.0.0.1:6379> getrange key1 2 -1  #從第二個字符開始，獲取全部字符串
"hello,zhangsan"
127.0.0.1:6379> getrange key1 0 -1  #獲取全部字符串 和 get key是一樣的
"vlhello,zhangsan"#替換
127.0.0.1:6379> set key2 asfzxc
OK
127.0.0.1:6379> get key2
"asfzxc"
127.0.0.1:6379> setrange key2 2 www  #替換指定位置開始的字符串
(integer) 6
127.0.0.1:6379> get key2
"aswwwc"
#############################################################
# setex (set with expire)  #設置過期時間
# setnx (set if not exist)  #不存在在設置 (在分布式鎖中會常常使用)
127.0.0.1:6379> get key1
"vlhello,zhangsan"
127.0.0.1:6379> setex key1 10 "zhangsan" #設置key1的值為 zhangsan，10秒后過期    
OK
127.0.0.1:6379> get key1
"zhangsan"
127.0.0.1:6379> ttl key1
(integer) -2
127.0.0.1:6379> get key1
(nil)
127.0.0.1:6379> setnx key3 "redis"  #如果key3不存在，創建key3
(integer) 1
127.0.0.1:6379> setnx key3 "monkey" #如果key3存在，創建失敗
(integer) 0
127.0.0.1:6379> get key3
"redis"
#############################################################
mset
mget
127.0.0.1:6379> mset k1 v1 k2 v2 k3 v3  #同時設置多個值
OK
127.0.0.1:6379> keys *
1) "k3"
2) "k2"
3) "k1"
127.0.0.1:6379> mget k1 k2 k3  #同時獲取多個值
1) "v1"
2) "v2"
3) "v3"
127.0.0.1:6379>  msetnx k1 v1 k5 v5 #msetnx 是一個原子性操作，要么都成功要么都失敗
(integer) 0
127.0.0.1:6379> get k5
(nil)#對象
set user:1 {name:zhangsan,age:30}  #設置一個user：1對象值為json字符來保存一個對象。
127.0.0.1:6379> set user:1 {name:zhangsan,age:30}
OK
127.0.0.1:6379> get user:1
"{name:zhangsan,age:30}"#這里的key是一個巧妙的設計： user:{id}:{filed}，如此設計在redis中是可以的。
127.0.0.1:6379> mset user:1:name lisi user:1:age 2
OK
127.0.0.1:6379> mget user:1:name user:1:age
1) "lisi"
2) "2"
127.0.0.1:6379> get user:1
(nil)
#############################################################
getset  #先get然后再set127.0.0.1:6379> get db
(nil)
127.0.0.1:6379> getset db redis
(nil)
127.0.0.1:6379> get db
"redis"
127.0.0.1:6379> getset db monkey #如果存在值，獲取原來的值，并設置新的值
"redis"
127.0.0.1:6379> get db
"monkey"

String類似的使用場景：value除了是字符串還可以是數字，用途舉例：

計數器
統計多單位的數量：uid:123666：follow 0
粉絲數
對象存儲緩存

List(列表)

Redis列表是簡單的字符串列表，按照插入順序排序。你可以添加一個元素到列表的頭部（左邊）或者尾部（右邊）

一個列表最多可以包含 $2^{32} - 1$ 個元素 (4294967295, 每個列表超過40億個元素)。

首先我們列表，可以經過規則定義將其變為隊列、棧、雙端隊列等

正如圖Redis中List是可以進行雙端操作的，所以命令也就分為了LXXX和RXXX兩類，有時候L也表示List例如LLEN

在這里插入圖片描述

####################################################
127.0.0.1:6379> lpush list one  #將一個值或者多個值，插入到列表頭部 (左邊插入)
(integer) 1
127.0.0.1:6379> lpush list two
(integer) 2
127.0.0.1:6379> lpush list three
(integer) 3
127.0.0.1:6379> lrange list 0 -1  #獲取list中的值
1) "three"
2) "two"
3) "one"
127.0.0.1:6379> lrange list 0 1  #通過區間獲取具體的值
1) "three"
2) "two"
127.0.0.1:6379> rpush list right    #將一個值或者多個值，插入到列表尾部 (右邊插入)
(integer) 4
127.0.0.1:6379> lrange list 0 -1
1) "three"
2) "two"
3) "one"
4) "right"
####################################################
LPOP
RPOP
127.0.0.1:6379> lpop list  #移除list的第一個元素
"three"
127.0.0.1:6379> lpop list 2  #移除list的前兩個元素
1) "two"
2) "one"
127.0.0.1:6379> lrange list  0 -1  #獲取list中的值
1) "right"
127.0.0.1:6379> lrange list 0 -1
1) "right"
2) "li1"
3) "li"
127.0.0.1:6379> rpop list 1 #移除list的最后一個元素
1) "li"
127.0.0.1:6379> lrange list 0 -1
1) "right"
2) "li1"
####################################################
Lindex
127.0.0.1:6379> lrange list 0 -1
1) "right"
2) "li1"
127.0.0.1:6379> lindex list 2 #通過下標獲得list中的某一個值
(nil)
127.0.0.1:6379> lindex list 0
"right"
####################################################
Llen
127.0.0.1:6379> llen list  #返回列表的長度
(integer) 2
####################################################
移除指定的值  Lrem
取關用到： uid
127.0.0.1:6379> lrange list  0 -1
1) "one"
2) "one"
3) "right"
4) "li1"
127.0.0.1:6379> lrem list 1 one  #移除list集合中指定個數的value，精確匹配
(integer) 1
127.0.0.1:6379> lrange list  0 -1
1) "one"
2) "right"
3) "li1"
127.0.0.1:6379> lpush list one
(integer) 4
127.0.0.1:6379> lpush list one
(integer) 5
127.0.0.1:6379> lrange list 0 -1
1) "one"
2) "one"
3) "one"
4) "right"
5) "li1"
127.0.0.1:6379> lrem list 3 one  #移除三個指定的值
(integer) 3
127.0.0.1:6379> lrange list 0 -1
1) "right"
2) "li1"
####################################################
trim  修剪 。 list截斷
127.0.0.1:6379> rpush mulist "hello1"
(integer) 1
127.0.0.1:6379> rpush mulist "hello2"
(integer) 2
127.0.0.1:6379> rpush mulist "hello3"
(integer) 3
127.0.0.1:6379> rpush mulist "hello4"
(integer) 4
127.0.0.1:6379> lrange mulist  0 -1
1) "hello1"
2) "hello2"
3) "hello3"
4) "hello4"
127.0.0.1:6379> ltrim mulist 1 2  #通過下標截取指定的長度，這個list已經被改變了，截斷了只剩下截取的元素。
OK
127.0.0.1:6379> lrange mulist  0 -1
1) "hello2"
2) "hello3"
####################################################
rpoplpush #移除列表的最后一個元素，將它移動到新的列表當中
127.0.0.1:6379> lrange mulist 0 -1
1) "hello2"
2) "hello3"
3) "hello4"
4) "hello5"
127.0.0.1:6379> rpoplpush mulist myotherlist #移除列表的最后一個元素，將它移動到新的列表當中
"hello5"
127.0.0.1:6379> lrange mulist 0 -1   #查看原來的列表
1) "hello2"
2) "hello3"
3) "hello4"
127.0.0.1:6379> lrange myotherlist 0 -1  #查看目標列表中，確實存在該值
1) "hello5"
####################################################
lset  將列表中指定下標的值替換為另一個值，相當于更新操作
127.0.0.1:6379> exists lit  #判斷這個列表是否存在
(integer) 0
127.0.0.1:6379> lset list 0 item  #如果不存在列表 我們去更新就會報錯
(error) ERR no such key
127.0.0.1:6379> lpush list value1
(integer) 1
127.0.0.1:6379> lrange list 0 0
1) "value1"
127.0.0.1:6379> lset list 0 item  #如果存在，更新當前下標的值
OK
127.0.0.1:6379> lrange list 0 0
1) "item"
127.0.0.1:6379> lset list 1 other  #如果不存在 則會報錯
(error) ERR index out of range
####################################################
linsert  #將某個具體的value值插入到列表中某個元素的前面或者后面127.0.0.1:6379> lrange mulist 0 -1
1) "hello2"
2) "hello3"
3) "hello4"
127.0.0.1:6379> linsert mulist before hello3 zhangsan #將zhangsan值插入到mulist列表中hello3的前面
(integer) 4
127.0.0.1:6379> lrange mulist 0 -1
1) "hello2"
2) "zhangsan"
3) "hello3"
4) "hello4"
127.0.0.1:6379> linsert mulist after hello3 zhangsha #將zhangsha值插入到mulist列表中hello3的后面
(integer) 5
127.0.0.1:6379> lrange mulist 0 -1
1) "hello2"
2) "zhangsan"
3) "hello3"
4) "zhangsha"
5) "hello4"###########################################
blpop/brpop: blocked pop
127.0.0.1:6379> lrange list 0 -1
1) "1"
127.0.0.1:6379> blpop list 15
1) "list"
2) "1"
127.0.0.1:6379> blpop list 15
1) "list"
2) "11"
(3.11s)
127.0.0.1:6379> lrange list 0 -1
(empty list or set)
127.0.0.1:6379> blpop list 5
(nil)
(5.05s)
############################################
# 刪除指定方向指定數量的元素--lrem key count value: count > 0; count < 0; count = 0.
127.0.0.1:6379> lrange list 0 -1
1) "one"
2) "one"
3) "two"
4) "one"
5) "one"
6) "one"
7) "three"
8) "one"
9) "one"
127.0.0.1:6379> lrem list 1 one
(integer) 1
127.0.0.1:6379> lrange list 0 -1
1) "one"
2) "two"
3) "one"
4) "one"
5) "one"
6) "three"
7) "one"
8) "one"
127.0.0.1:6379> lrem list -1 one
(integer) 1
127.0.0.1:6379> lrange list 0 -1
1) "one"
2) "two"
3) "one"
4) "one"
5) "one"
6) "three"
7) "one"
127.0.0.1:6379> lrem list 0 one
(integer) 5
127.0.0.1:6379> lrange list 0 -1
1) "two"
2) "three"
127.0.0.1:6379> lrem list 10 two
(integer) 1
127.0.0.1:6379> lrange list 0 -1
1) "three"
######################################################
# lindex: 根據 index 獲取元素
127.0.0.1:6379> lindex list 2
"zhangsan"
######################################################
# lpushx/rpushx: push if list eXists
127.0.0.1:6379> lrange list 0 -1
1) "one"
2) "zhangsan"
3) "11"
4) "lisi"
5) "three"
6) "four"
127.0.0.1:6379> lpushx list1 val
(integer) 0
127.0.0.1:6379> lpushx list val
(integer) 7
127.0.0.1:6379> rpushx list val
(integer) 8
127.0.0.1:6379> lrange list 0 -1
1) "val"
2) "one"
3) "zhangsan"
4) "11"
5) "lisi"
6) "three"
7) "four"
8) "val"

小結

它實際上是一個鏈表，before node after ，left , right 都可以插入值
如果key不存在，創建新的鏈表
如果key存在，新增內容
如果移除了所有值，空鏈表，也代表不存在。
在兩邊插入或者改動值，效率最高，中間元素相對來說效率會低一點

可以做消息排隊消息隊列( Lpush Rpop) ，棧（Lpush Lpop）

Set(集合)

Redis的Set是string類型的無序集合。集合成員是唯一的，這就意味著集合中不能出現重復的數據。

Redis 中集合是通過哈希表實現的，所以添加，刪除，查找的復雜度都是O(1)。

集合中最大的成員數為 $2^{32} - 1$ (4294967295, 每個集合可存儲40多億個成員)。

在這里插入圖片描述

####################################################
127.0.0.1:6379> sadd myset hello  #set集合中添加元素
(integer) 1
127.0.0.1:6379> sadd myset zhangsan
(integer) 1
127.0.0.1:6379> sadd myset lisi
(integer) 1
127.0.0.1:6379> smembers myset  #查看指定set的所有值
1) "hello"
2) "lisi"
3) "zhangsan"
127.0.0.1:6379> sismember myset lisi #判斷某一個值是不是在set集合中
(integer) 1
127.0.0.1:6379> sismember myset world
(integer) 0####################################################
scard
127.0.0.1:6379> scard myset #獲取set集合中內容的元素個數！
(integer) 3
127.0.0.1:6379> sadd myset hello
(integer) 0
####################################################
rem
127.0.0.1:6379> srem myset hello #移除set集合中的指定元素
(integer) 1
127.0.0.1:6379> SMEMBERS myset
1) "lisi"
2) "zhangsan"
127.0.0.1:6379> scard myset
(integer) 2####################################################
set 無序不重復集合。比如：隨機抽獎
127.0.0.1:6379> srandmember myset   #隨機抽選出一個元素
"lisi"
127.0.0.1:6379> srandmember myset 
"zhangsan"
127.0.0.1:6379> srandmember myset 
"zhangsan"
127.0.0.1:6379> srandmember myset 
"lisi"
127.0.0.1:6379> SMEMBERS myset
1) "lisi"
2) "zhangsan"
127.0.0.1:6379> srandmember myset 2  #隨機抽選出兩個元素
1) "lisi"
2) "zhangsan"####################################################
刪除定的key，刪除隨機的key127.0.0.1:6379> SMEMBERS myset
1) "111"
2) "blue"
3) "lisi"
4) "zhangsan"
127.0.0.1:6379> spop myset  #隨機刪除一些set集合中的元素
"lisi"
127.0.0.1:6379> spop myset
"111"
127.0.0.1:6379> SMEMBERS myset
1) "blue"
2) "zhangsan"####################################################
smove 將一個指定的值，移動到另外一個set集合
127.0.0.1:6379> sadd myset world zhangsan 
(integer) 2
127.0.0.1:6379> SMEMBERS myset
1) "hello"
2) "world"
3) "hello2"
4) "hello3"
5) "zhangsan"
127.0.0.1:6379> smove myset myset2 zhangsan   #將myset中的zhangsan移動到myset2集合中去
(integer) 1
127.0.0.1:6379> SMEMBERS myset
1) "world"
2) "hello2"
3) "hello3"
4) "hello"
127.0.0.1:6379> SMEMBERS myset2
1) "zhangsan"####################################################
微博，B站中的共同關注！(并集)
- 差集  sdiff
- 交集  sinter
- 并集  sunion
127.0.0.1:6379> sadd key1 a b c d 
(integer) 4
127.0.0.1:6379> sadd key2 c d e f
(integer) 4
127.0.0.1:6379> SDIFF key1 key2   #差集
1) "b"
2) "a"
127.0.0.1:6379> SINTER key1 key2  #交集   共同好友就可以這樣實現
1) "d"
2) "c"
127.0.0.1:6379> SUNION key1 key2  #并集
1) "b"
2) "c"
3) "e"
4) "a"
5) "d"
6) "f"
127.0.0.1:6379> sadd set1 a b c d
(integer) 4
127.0.0.1:6379> sadd set2 c d e f
(integer) 4
127.0.0.1:6379> sdiffstore set3 set1 set2
(integer) 2
127.0.0.1:6379> smembers set3
1) "a"
2) "b"

Hash（哈希）

Redis hash 是一個string類型的field和value的映射表，hash特別適合用于存儲對象。

Set就是一種簡化的Hash,只變動key,而value使用默認值填充。可以將一個Hash表作為一個對象進行存儲，表中存放對象的信息。

在這里插入圖片描述

Map集合， key-map，時候這個值是一個map集合。本質和string類型沒有太大區別，還是一個簡單的key-value！

####################################################
hset myhash field zhangsan
127.0.0.1:6379> hset myhash field1 zhangsan  #set一個具體的 key-value
(integer) 1
127.0.0.1:6379> hget myhash field1 #獲取一個字段值
"zhangsan"
127.0.0.1:6379> hmset myhash field1 hello field2 world #設置多個key-value
OK
127.0.0.1:6379> hmget myhash field1 field2 #獲取多個字段值
1) "hello"
2) "world"
127.0.0.1:6379> hgetall myhash #獲取全部的數據
1) "field1"
2) "hello"
3) "field2"
4) "world"
127.0.0.1:6379> hdel myhash field1 #刪除hash指定key字段！ 對應的value值也就消失了
(integer) 1
127.0.0.1:6379> HGETALL myhash
1) "field2"
2) "world"
####################################################
hlen #獲取hash表的字段數量
127.0.0.1:6379> hmset myhash field hello field3 wufeng field4 lis 
OK
127.0.0.1:6379> HGETALL myhash
1) "field2"
2) "world"
3) "field"
4) "hello"
5) "field3"
6) "wufeng"
7) "field4"
8) "lis"
127.0.0.1:6379> hget myhash field  
"hello"
127.0.0.1:6379> hlen myhash  #獲取hash表的字段數量
(integer) 4####################################################
127.0.0.1:6379> HEXISTS myhash field #判斷hash中指定字段是否存在
(integer) 1
127.0.0.1:6379> HEXISTS myhash field6
(integer) 0####################################################
#只獲得所有filed   hkeys
#只獲得所有value   hvals
127.0.0.1:6379> HKEYS myhash #獲得所有filed
1) "field2"
2) "field"
3) "field3"
4) "field4" 
127.0.0.1:6379> HVALS myhash  #獲得所有value
1) "world"
2) "hello"
3) "wufeng"
4) "lis"####################################################
hincrby #加1  hdecrby #減1
127.0.0.1:6379> hset myhash field6 5  #指定增量為5
(integer) 1
127.0.0.1:6379> HINCRBY myhash field6 1 #數值加1
(integer) 6
127.0.0.1:6379> HINCRBY myhash field6 -2 #數值減2
(integer) 3
127.0.0.1:6379> HSETNX myhash field7 hello #如果不存在則可以設置
(integer) 1
127.0.0.1:6379> HSETNX myhash field6 hello #如果存在則可以設置
(integer) 0

hash變更的數據 user name age ,尤其是用戶信息之類的，經常變動的信息。 hash更適合于對象的存儲，String更加適合字符串的存儲。

127.0.0.1:6379> hmset myhash user:1:name zhangsan user:1:age 30 #設置user:{id}:{field}
OK
127.0.0.1:6379> hmget myhash user:1:name user:1:age
1) "zhangsan"
2) "30"
127.0.0.1:6379> HGETALL myhash
1) "field7"
2) "hello"
3) "user:1:name"
4) "zhangsan"
5) "user:1:age"
6) "30"

Zset（有序集合）

不同的是每個元素都會關聯一個double類型的分數（score）。redis正是通過分數來為集合中的成員進行從小到大的排序。

score相同：按字典順序排序

有序集合的成員是唯一的,但分數(score)卻可以重復。

在這里插入圖片描述

在set的基礎上，增加一個值，

set: k1 v1
zset: k1 score1 v1

####################################################
127.0.0.1:6379> zadd myset 1 one  #添加一個值
(integer) 1
127.0.0.1:6379> zadd myset 2 two 3 three #添加多個值
(integer) 2
127.0.0.1:6379> zrange myset 0 -1
1) "one"
2) "two"
3) "three"####################################################
排序如何實現127.0.0.1:6379> zadd salary 2500 xiaohong  #添加三個用戶   薪水，名字
(integer) 1
127.0.0.1:6379> zadd salary 5000 zhangsan
(integer) 1
127.0.0.1:6379> zadd salary 1000 lisi
(integer) 1
127.0.0.1:6379> ZRANGEBYSCORE salary -inf +inf  #顯示全部用戶，按薪水從小到大排名 -inf：表示負無窮
1) "lisi"
2) "xiaohong"
3) "zhangsan"
127.0.0.1:6379> zrevrange salary 0 -1 withscores # 從大到小進行排序
1) "zhangsan"
2) "5000"
3) "xiaohong"
4) "2500"
127.0.0.1:6379> ZRANGEBYSCORE salary -inf +inf withscores #查找所有的數據 ，withscores：帶有的薪水信息
1) "lisi"
2) "1000"
3) "xiaohong"
4) "2500"
5) "zhangsan"
6) "5000"
127.0.0.1:6379> ZRANGEBYSCORE salary -inf 2500 withscores #顯示工資小于2500員工的升序排列
1) "lisi"
2) "1000"
3) "xiaohong"
4) "2500"####################################################
移除：zrem
127.0.0.1:6379> zrange salary 0 -1
1) "lisi"
2) "xiaohong"
3) "zhangsan"
127.0.0.1:6379> zrem salary lisi  #移除有序集合中的指定元素 
(integer) 1
127.0.0.1:6379> zrange salary 0 -1
1) "xiaohong"
2) "zhangsan"
127.0.0.1:6379> zcard salary  #獲取集合中的個數
(integer) 2####################################################
127.0.0.1:6379> zrange myset 0 -1
1) "one"
2) "two"
3) "three"
127.0.0.1:6379> zcount myset 0 2  #獲取指定分數區間的成員數量
(integer) 2
127.0.0.1:6379> zcount myset 0 5
(integer) 3##################################################
127.0.0.1:6379> zrange myzset 0 -1 withscores
1) "one"
2) "1"
3) "two"
4) "2"
5) "three"
6) "3"
127.0.0.1:6379> zcount myzset 0 2
(integer) 2
127.0.0.1:6379> zincrby myzset 2 one
"3"
127.0.0.1:6379> zrange myzset 0 -1 withscores
1) "two"
2) "2"
3) "one"
4) "3"
5) "three"
6) "3"
127.0.0.1:6379> zscore myzset one
"3"
127.0.0.1:6379> zrank myset one
(error) WRONGTYPE Operation against a key holding the wrong kind of value
127.0.0.1:6379> zrank myzset one
(integer) 1

應用案例：

set排序存儲班級成績表工資表排序！
普通消息，1.重要消息 2.帶權重進行判斷
排行榜應用實現，取Top N測試

Geospatial(地理位置)

使用經緯度定位地理坐標并用一個有序集合zset保存，所以zset命令也可以使用

在這里插入圖片描述

有效經緯度

有效的經度從-180度到180度。
有效的緯度從-85.05112878度到85.05112878度。

指定單位的參數 unit 必須是以下單位的其中一個：

m 表示單位為米。

km 表示單位為千米。

mi 表示單位為英里。

ft 表示單位為英尺。

關于GEORADIUS的參數

通過georadius就可以完成附近的人功能

withcoord:帶上坐標

withdist:帶上距離，單位與半徑單位相同

COUNT n : 只顯示前n個(按距離遞增排序)

geoadd

#geoadd 添加地理位置
# 規則： 兩極無法直接添加，我們一般會下載城市數據，直接通過java程序一次性導入
#參數  key 值(緯度、經度、名稱)#有效的經度從-180度到180度。#有效的緯度從-85.05112878度到85.05112878度。#當坐標位置超出上述指定范圍時，該命令將會返回一個錯誤。
127.0.0.1:6379> geoadd china:city  116.40 39.90 beijing
(integer) 1
127.0.0.1:6379> geoadd china:city 121.47 31.23 shanghai
(integer) 1
127.0.0.1:6379> geoadd china:city 106.50 29.53  chongqing
(integer) 1
127.0.0.1:6379> geoadd china:city 114.08 22.54   shenzhen 116.85 38.31 cangzhou
(integer) 2
127.0.0.1:6379> geoadd china:city 120.15 30.28  hangzhou 125.14 42.92 xian
(integer) 2

geopos

獲得當前定位：一定是一個坐標值。
127.0.0.1:6379> geopos china:city beijing #獲取指定的城市的經度和緯度
1) 1) "116.39999896287918091"2) "39.90000009167092543"
127.0.0.1:6379> geopos china:city beijing cangzhou
1) 1) "116.39999896287918091"2) "39.90000009167092543"
2) 1) "116.84999853372573853"2) "38.30999992507150864"

geodist

兩人之間的距離

單位：

m 表示單位為米
km 表示單位為千米
mi 表示單位為英里
ft 表示單位為英尺

127.0.0.1:6379> geodist china:city beijing shanghai km #查看北京到上海的直線距離
"1067.3788"
127.0.0.1:6379> GEODIST china:city beijing chongqing km #查看北京到重慶的直線距離
"1464.0708"

georadius 以給定的經緯度為中心，找出某一半徑內的元素

我附近的人？ (獲得所有附近的人的地址，定位！) 通過半徑來查詢

獲得指定數量的人為200個 count 200

所有數據應該都錄入 china:city ，才會讓結果更加清晰

127.0.0.1:6379> GEORADIUS china:city 110 30 500 km #以110 30這個經度緯度為中心，尋找方圓1000km內的城市
1) "chongqing"
127.0.0.1:6379> GEORADIUS china:city 110 30 1000 km
1) "chongqing"
2) "shenzhen"
3) "hangzhou"
127.0.0.1:6379> GEORADIUS china:city 110 30 1000 km withdist #顯示到中間距離的位置
1) 1) "chongqing"2) "341.9374"
2) 1) "shenzhen"2) "923.9364"
3) 1) "hangzhou"2) "976.4868"
127.0.0.1:6379> GEORADIUS china:city 110 30 1000 km withdist withcoord  #顯示他人的定位信息
1) 1) "chongqing"2) "341.9374"3) 1) "106.49999767541885376"2) "29.52999957900659211"
2) 1) "shenzhen"2) "923.9364"3) 1) "114.08000081777572632"2) "22.53999903789756587"
3) 1) "hangzhou"2) "976.4868"3) 1) "120.15000075101852417"2) "30.2800007575645509"
127.0.0.1:6379> GEORADIUS china:city 110 30 1000 km withdist withcoord  count 1
1) 1) "chongqing"2) "341.9374"3) 1) "106.49999767541885376"2) "29.52999957900659211"
127.0.0.1:6379> GEORADIUS china:city 110 30 1000 km withdist withcoord  count 2 #篩選出指定的結果！
1) 1) "chongqing"2) "341.9374"3) 1) "106.49999767541885376"2) "29.52999957900659211"
2) 1) "shenzhen"2) "923.9364"3) 1) "114.08000081777572632"2) "22.53999903789756587"
127.0.0.1:6379> 127.0.0.1:6379> georadius china:city 110 30 1000 km withcoord withdist withhash
1) 1) "chongqin"2) "341.9374"3) (integer) 40260420916289844) 1) "106.49999767541885376"2) "29.52999957900659211"
2) 1) "shenzhen"2) "923.9364"3) (integer) 40464322968005464) 1) "114.08000081777572632"2) "22.53999903789756587"
3) 1) "hangzhou"2) "976.4868"3) (integer) 40541342573907834) 1) "120.15000075101852417"2) "30.2800007575645509"
127.0.0.1:6379> georadius china:city 110 30 1000 km withcoord withdist count 2
1) 1) "chongqin"2) "341.9374"3) 1) "106.49999767541885376"2) "29.52999957900659211"
2) 1) "shenzhen"2) "923.9364"3) 1) "114.08000081777572632"2) "22.53999903789756587"
127.0.0.1:6379> georadius china:city 110 30 1000 km withcoord withdist count 2 asc
1) 1) "chongqin"2) "341.9374"3) 1) "106.49999767541885376"2) "29.52999957900659211"
2) 1) "shenzhen"2) "923.9364"3) 1) "114.08000081777572632"2) "22.53999903789756587"
127.0.0.1:6379> georadius china:city 110 30 1000 km withcoord withdist count 2 desc
1) 1) "hangzhou"2) "976.4868"3) 1) "120.15000075101852417"2) "30.2800007575645509"
2) 1) "shenzhen"2) "923.9364"3) 1) "114.08000081777572632"2) "22.53999903789756587"

georadiusbymember

#找出位于指定元素(集合中已經存在)周圍的其他元素
127.0.0.1:6379> GEORADIUSBYMEMBER china:city beijing 1000 km
1) "cangzhou"
2) "beijing"
3) "xian"
127.0.0.1:6379> GEORADIUSBYMEMBER china:city shanghai 500km
(error) ERR wrong number of arguments for 'georadiusbymember' command
127.0.0.1:6379> GEORADIUSBYMEMBER china:city shanghai 500 km
1) "hangzhou"
2) "shanghai"

geohash 返回一個或多個位置元素的geohash表示（很少使用到）

該命令將返回11個字符的geohash字符串

#將二維的經緯度轉換為一維的字符串，如果兩個字符串越近，那么則距離越近。
127.0.0.1:6379> GEOHASH china:city beijing chongqing
1) "wx4fbxxfke0"
2) "wm5xzrybty0"

geo 底層的實現原理其實就是Zset 我們可以使用Zset命令來操作geo

127.0.0.1:6379> ZRANGE china:city 0 -1  #查看地圖中全部元素
1) "chongqing"
2) "shenzhen"
3) "hangzhou"
4) "shanghai"
5) "cangzhou"
6) "beijing"
7) "xian"
127.0.0.1:6379> zrem china:city beijing xian  #移除指定元素
(integer) 2
127.0.0.1:6379> ZRANGE china:city 0 -1
1) "chongqing"
2) "shenzhen"
3) "hangzhou"
4) "shanghai"
5) "cangzhou"

Hyperloglog(基數統計)

Redis HyperLogLog 是用來做基數統計的算法，HyperLogLog 的優點是，在輸入元素的數量或者體積非常非常大時，計算基數所需的空間總是固定的、并且是很小的。

花費 12 KB 內存，就可以計算接近 2^64 個不同元素的基數。

因為 HyperLogLog 只會根據輸入元素來計算基數，而不會儲存輸入元素本身，所以 HyperLogLog 不能像集合那樣，返回輸入的各個元素。

其底層使用string數據類型

什么是基數？

數據集中不重復的元素的個數。

即獨立用戶的數量，如：

假設我們有一個網站，需要統計每天訪問網站的獨立用戶數量。通常情況下，我們可以使用傳統的方法，比如將每個用戶的 ID 記錄在一個集合中，然后使用集合的大小來統計獨立用戶數量。但是，當用戶量非常大時，這種方法會占用大量內存空間。

這時，我們可以使用 HyperLogLog 來進行基數統計。假設有一天，我們接收到了以下用戶訪問網站的 ID 數據：

用戶ID：[1001, 2002, 3003, 1001, 4004, 1001, 5005, 6006, 2002]

現在，我們使用 HyperLogLog 結構來統計這些用戶的獨立數量。首先，我們將每個用戶 ID 進行哈希映射，得到一個哈希值，然后根據哈希值來估計基數。

經過哈希映射后，得到的哈希值可能類似于：
[0x32, 0x18, 0x7F, 0x32, 0x45, 0x32, 0x92, 0x11, 0x18]

接著，我們使用位操作來統計零位前導串（zero leading string）的長度。假設經過統計后，得到的零位前導串長度為：
[5, 4, 6, 5, 3, 5, 7, 2, 4]

最后，通過對零位前導串長度的統計結果進行分析，就能夠估計出獨立用戶的數量。
應用場景：
- 網頁的訪問量（UV）：一個用戶多次訪問，也只能算作一個人。
- 傳統實現，存儲用戶的id,然后每次進行比較。當用戶變多之后這種方式及其浪費空間，而我們的目的只是計數，Hyperloglog就能幫助我們利用最小的空間完成。

在這里插入圖片描述

127.0.0.1:6379> PFADD mykey a b c d e f g h k  #創建第一組元素  mykey
(integer) 1
127.0.0.1:6379> PFCOUNT mykey #統計 mykey 元素的基數數量
(integer) 9
127.0.0.1:6379> PFADD mykey2 q w e r t y u i o #創建第二組元素  mykey2
(integer) 1
127.0.0.1:6379> PFCOUNT mykey2
(integer) 9
127.0.0.1:6379> PFMERGE mykey3 mykey mykey2 # 合并兩組  mykey mykey2 => mykey3
OK
127.0.0.1:6379> PFCOUNT mykey3 #看并集的數量  相同的不計
(integer) 16

如果允許容錯，那么一定可以使用Hyperloglog !

如果不允許容錯，就使用set或者自己的數據類型即可！

原因：

HyperLogLog 通過概率統計來估計基數是因為其設計的目的是在犧牲一定的精確度的情況下，換取更小的內存占用和更高的性能。這種犧牲精確度來換取效率的策略被稱為"概率算法"。

HyperLogLog 的近似值原理基于統計學中的概率方法，利用哈希函數將元素映射到一個固定長度的比特串中，然后根據比特串的特征來估計基數。在實際應用中，HyperLogLog 會對零位前導串（zero leading string）的長度進行統計，并根據統計結果來推測獨立元素的數量。

由于 HyperLogLog 使用的是概率方法，因此得到的結果是一個近似值，而不是精確值。這意味著在某些情況下，可能會出現誤差，但通常情況下，這種誤差是可以接受的。通過犧牲一定的精確度，HyperLogLog 能夠以較小的內存消耗來快速估計大型數據集合的基數，適用于需要高效處理大規模數據的場景。

BitMaps(位圖)

使用位存儲，信息狀態只有 0 和 1

Bitmap是一串連續的2進制數字（0或1），每一位所在的位置為偏移(offset)，在bitmap上可執行AND,OR,XOR,NOT以及其它位操作。

應用場景: 簽到統計、狀態統計

在這里插入圖片描述

統計用戶信息,活躍,不活躍!登錄,未登錄!打卡,未打卡! 兩個的狀態,都可以使用bitmaps

Bitmaps位圖,數據結構!都是操作二進制位來進行記錄,就只有0和1兩個狀態.

365天 = 365bit 1字節 = 8bit 46個字節左右!

使用bitmap來記錄周一到周日的打卡

127.0.0.1:6379> setbit sign 0 1  #周一到周日打卡情況，1代表打卡  0代表未打卡
(integer) 0
127.0.0.1:6379> setbit sign 1 1
(integer) 0
127.0.0.1:6379> setbit sign 3 1
(integer) 0
127.0.0.1:6379> setbit sign 4 0
(integer) 0
127.0.0.1:6379> setbit sign 5 0
(integer) 0
127.0.0.1:6379> setbit sign 6 1
(integer) 0127.0.0.1:6379> GETBIT sign 5  #查看某一天是否打卡
(integer) 0
127.0.0.1:6379> GETBIT sign 6
(integer) 1127.0.0.1:6379> bitcount sign #統計這周的打卡記錄
(integer) 4

bitmaps是一串從左到右的二進制串

三、事務

Redis事務本質：一組命令的集合。一個事務中的所有命令都會被序列化，在事務執行過程中，會按照順序執行！

一次性、順序性、排他性 執行一系列的命令

------  隊列  set set set 執行  ---------

Redis事務沒有隔離級別的概念！

所有的命令在事務中，并沒有直接被執行，只有發起執行命令的時候才會執行

redis單條命令是保證原子性的，但是事務不保證原子性

Redis的事務：

開啟事務（multi）
命令入隊（…）
執行事務（exec）

正常執行事務

127.0.0.1:6379> multi  #開啟事務
OK
127.0.0.1:6379(TX)> set k1 v1  #命令入隊
QUEUED
127.0.0.1:6379(TX)> set k2 v2
QUEUED
127.0.0.1:6379(TX)> get k2
QUEUED
127.0.0.1:6379(TX)> set k3 v3
QUEUED
127.0.0.1:6379(TX)> exec  #執行事務
1) OK
2) OK
3) "v2"
4) OK

放棄事務 ( discard )

127.0.0.1:6379> multi #開啟事務
OK
127.0.0.1:6379(TX)> set k1 v1
QUEUED
127.0.0.1:6379(TX)> set k2 v2
QUEUED
127.0.0.1:6379(TX)> DISCARD  #取消事務
OK
127.0.0.1:6379> get k2  #事務隊列中命令都不會被執行
(nil)
127.0.0.1:6379> get k1
(nil)

編譯型異常(代碼有問題，及命令有錯) ，事務中所有的命令都不會被執行

127.0.0.1:6379> multi
OK
127.0.0.1:6379(TX)> set k1 v1 
QUEUED
127.0.0.1:6379(TX)> set k2 v2
QUEUED
127.0.0.1:6379(TX)> getset k2  #錯誤的命令
(error) ERR wrong number of arguments for 'getset' command
127.0.0.1:6379(TX)> set k3 v3
QUEUED
127.0.0.1:6379(TX)> exec  #執行事務報錯
(error) EXECABORT Transaction discarded because of previous errors.
127.0.0.1:6379> get k3   #所有的命令都不執行
(nil)
127.0.0.1:6379> get k1
(nil)

運行時異常(1/0)，如果事務隊列中存在語法性，那么執行命令的時候，其他命令是可以正常執行的，錯誤命令拋出異常

127.0.0.1:6379> set k1 v1
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379(TX)> set k2 v2
QUEUED
127.0.0.1:6379(TX)> set k3 v3
QUEUED
127.0.0.1:6379(TX)> incr k1  #執行的時候會失敗
QUEUED
127.0.0.1:6379(TX)> get k2
QUEUED
127.0.0.1:6379(TX)> exec
1) OK
2) OK
3) (error) ERR value is not an integer or out of range #雖然第一條命令報錯了，但是依舊正常執行成功了。
4) "v2"
127.0.0.1:6379> get k3
"v3"

悲觀鎖、樂觀鎖

監控 watch （面試常問）

悲觀鎖：

很悲觀，認為什么時候都會出現問題，無論做什么都會加鎖

樂觀鎖：

很樂觀，認為什么時候都不會出現問題，所以不會上鎖！更新數據的時候去判斷一下，在此期間是否有人修改過這個數據
獲取version
更新的時候比較version

redis監視測試

正常執行成功

127.0.0.1:6379> set money 100  # 設置余額:100
OK
127.0.0.1:6379> set out 0  # 支出使用:0
OK
127.0.0.1:6379> watch money #監視money 對象
OK
127.0.0.1:6379> multi  
OK
127.0.0.1:6379(TX)> DECRBY money 20
QUEUED
127.0.0.1:6379(TX)> INCRBY out 20
QUEUED
127.0.0.1:6379(TX)> exec  #事務正常結束，數據期間沒有發生變動，這個時候就正常執行成功
1) (integer) 80
2) (integer) 20

測試多線程修改值，使用watch 可以當做redis的樂觀鎖操作

#第一個線程
127.0.0.1:6379> watch money  #監控 money
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379(TX)> DECRBY money 20
QUEUED
127.0.0.1:6379(TX)> INCRBY out 20
QUEUED
127.0.0.1:6379(TX)> exec  #執行之前，另外一個線程，修改了我們的值，這個時候就會導致食物執行失敗。
(nil)#第二個線程
127.0.0.1:6379> get money
"80"
127.0.0.1:6379> INCRby money 100
(integer) 180
127.0.0.1:6379> get money
"180"

? unwatch進行解鎖。

如果修改失敗，獲取最新的值就好

注意：每次提交執行exec后都會自動釋放鎖，不管是否成功

四、Jedis

使用Java來操作Redis，Jedis是Redis官方推薦使用的Java連接redis的客戶端的開發工具，使用java操作redis中間件。

測試

1.導入對應的依賴

  <dependencies><!--導入jedis的包--><dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId><version>3.2.0</version></dependency><!--fastjson 存一些數據用的--><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.48</version></dependency></dependencies>

2.編碼測試：

連接數據庫
操作命令
斷開連接

若連接本地：

public class TestPing {public static void main(String[] args) {Jedis jedis = new Jedis("127.0.0.1", 6379);System.out.println(jedis.ping());}
}

若連接阿里云服務器：

public class TestPing {public static void main(String[] args) {Jedis jedis = new Jedis("47.120.xxx.xxx", 6379);jedis.auth("19991001");System.out.println(jedis.ping());}
}

package com.jihu;
import redis.clients.jedis.Jedis;
public class TestPing {public static void main(String[] args) {//1. new jedis 對象即可Jedis jedis = new Jedis("192.168.56.130",6379);  //自己虛擬機的ip地址jedis.auth("123456");// jedis 所有的命令就是我們之前學習的所有指令System.out.println(jedis.ping());}
}

輸出結果:

常用的API

String 、list、set 、hash、Zset等所有的api命令，就是我們對應的上面學習的指令，一個都沒有變化！

判斷keys * 以及一些基本方法

package com.jihu.base;import redis.clients.jedis.Jedis;import java.util.Set;public class TestKey {public static void main(String[] args) {Jedis jedis = new Jedis("192.168.56.130",6379);jedis.auth("123456");System.out.println("清空數據："+jedis.flushDB());System.out.println("判斷某個鍵是否存在："+jedis.exists("username"));System.out.println("新增<'username','kuangshen'>的鍵值對："+jedis.set("username", "kuangshen"));System.out.println("新增<'password','password'>的鍵值對："+jedis.set("password", "password"));System.out.print("系統中所有的鍵如下：");Set<String> keys = jedis.keys("*");System.out.println(keys);System.out.println("刪除鍵password:"+jedis.del("password"));System.out.println("判斷鍵password是否存在："+jedis.exists("password"));System.out.println("查看鍵username所存儲的值的類型："+jedis.type("username"));System.out.println("隨機返回key空間的一個："+jedis.randomKey());System.out.println("重命名key："+jedis.rename("username","name"));System.out.println("取出改后的name："+jedis.get("name"));System.out.println("按索引查詢："+jedis.select(0));System.out.println("刪除當前選擇數據庫中的所有key："+jedis.flushDB());System.out.println("返回當前數據庫中key的數目："+jedis.dbSize());System.out.println("刪除所有數據庫中的所有key："+jedis.flushAll());}
}

對String操作的命令

package com.jihu.base;
import redis.clients.jedis.Jedis;
import java.util.concurrent.TimeUnit;public class TestString {public static void main(String[] args) {Jedis jedis = new Jedis("192.168.56.130",6379);jedis.auth("123456");jedis.flushDB();System.out.println("===========增加數據===========");System.out.println(jedis.set("key1","value1"));System.out.println(jedis.set("key2","value2"));System.out.println(jedis.set("key3", "value3"));System.out.println("刪除鍵key2:"+jedis.del("key2"));System.out.println("獲取鍵key2:"+jedis.get("key2"));System.out.println("修改key1:"+jedis.set("key1", "value1Changed"));System.out.println("獲取key1的值："+jedis.get("key1"));System.out.println("在key3后面加入值："+jedis.append("key3", "End"));System.out.println("key3的值："+jedis.get("key3"));System.out.println("增加多個鍵值對："+jedis.mset("key01","value01","key02","value02","key03","value03"));System.out.println("獲取多個鍵值對："+jedis.mget("key01","key02","key03"));System.out.println("獲取多個鍵值對："+jedis.mget("key01","key02","key03","key04"));System.out.println("刪除多個鍵值對："+jedis.del("key01","key02"));System.out.println("獲取多個鍵值對："+jedis.mget("key01","key02","key03"));jedis.flushDB();System.out.println("===========新增鍵值對防止覆蓋原先值==============");System.out.println(jedis.setnx("key1", "value1"));System.out.println(jedis.setnx("key2", "value2"));System.out.println(jedis.setnx("key2", "value2-new"));System.out.println(jedis.get("key1"));System.out.println(jedis.get("key2"));System.out.println("===========新增鍵值對并設置有效時間=============");System.out.println(jedis.setex("key3", 2, "value3"));System.out.println(jedis.get("key3"));try {TimeUnit.SECONDS.sleep(3);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(jedis.get("key3"));System.out.println("===========獲取原值，更新為新值==========");System.out.println(jedis.getSet("key2", "key2GetSet"));System.out.println(jedis.get("key2"));System.out.println("獲得key2的值的字串："+jedis.getrange("key2", 2, 4));}
}

對List操作的命令

package com.jihu.base;
import redis.clients.jedis.Jedis;
public class TestList {public static void main(String[] args) {Jedis jedis = new Jedis("192.168.56.130",6379);jedis.auth("123456");jedis.flushDB();System.out.println("===========添加一個list===========");jedis.lpush("collections", "ArrayList", "Vector", "Stack", "HashMap", "WeakHashMap", "LinkedHashMap");jedis.lpush("collections", "HashSet");jedis.lpush("collections", "TreeSet");jedis.lpush("collections", "TreeMap");System.out.println("collections的內容："+jedis.lrange("collections", 0, -1));//-1代表倒數第一個元素，-2代表倒數第二個元素,end為-1表示查詢全部System.out.println("collections區間0-3的元素："+jedis.lrange("collections",0,3));System.out.println("===============================");// 刪除列表指定的值 ，第二個參數為刪除的個數（有重復時），后add進去的值先被刪，類似于出棧System.out.println("刪除指定元素個數："+jedis.lrem("collections", 2, "HashMap"));System.out.println("collections的內容："+jedis.lrange("collections", 0, -1));System.out.println("刪除下表0-3區間之外的元素："+jedis.ltrim("collections", 0, 3));System.out.println("collections的內容："+jedis.lrange("collections", 0, -1));System.out.println("collections列表出棧（左端）："+jedis.lpop("collections"));System.out.println("collections的內容："+jedis.lrange("collections", 0, -1));System.out.println("collections添加元素，從列表右端，與lpush相對應："+jedis.rpush("collections", "EnumMap"));System.out.println("collections的內容："+jedis.lrange("collections", 0, -1));System.out.println("collections列表出棧（右端）："+jedis.rpop("collections"));System.out.println("collections的內容："+jedis.lrange("collections", 0, -1));System.out.println("修改collections指定下標1的內容："+jedis.lset("collections", 1, "LinkedArrayList"));System.out.println("collections的內容："+jedis.lrange("collections", 0, -1));System.out.println("===============================");System.out.println("collections的長度："+jedis.llen("collections"));System.out.println("獲取collections下標為2的元素："+jedis.lindex("collections", 2));System.out.println("===============================");jedis.lpush("sortedList", "3","6","2","0","7","4");System.out.println("sortedList排序前："+jedis.lrange("sortedList", 0, -1));System.out.println(jedis.sort("sortedList"));System.out.println("sortedList排序后："+jedis.lrange("sortedList", 0, -1));}
}

對set操作的命令

package com.jihu.base;import redis.clients.jedis.Jedis;public class TestSet {public static void main(String[] args) {Jedis jedis = new Jedis("192.168.56.130",6379);jedis.auth("123456");jedis.flushDB();System.out.println("============向集合中添加元素（不重復）============");System.out.println(jedis.sadd("eleSet", "e1","e2","e4","e3","e0","e8","e7","e5"));System.out.println(jedis.sadd("eleSet", "e6"));System.out.println(jedis.sadd("eleSet", "e6"));System.out.println("eleSet的所有元素為："+jedis.smembers("eleSet"));System.out.println("刪除一個元素e0："+jedis.srem("eleSet", "e0"));System.out.println("eleSet的所有元素為："+jedis.smembers("eleSet"));System.out.println("刪除兩個元素e7和e6："+jedis.srem("eleSet", "e7","e6"));System.out.println("eleSet的所有元素為："+jedis.smembers("eleSet"));System.out.println("隨機的移除集合中的一個元素："+jedis.spop("eleSet"));System.out.println("隨機的移除集合中的一個元素："+jedis.spop("eleSet"));System.out.println("eleSet的所有元素為："+jedis.smembers("eleSet"));System.out.println("eleSet中包含元素的個數："+jedis.scard("eleSet"));System.out.println("e3是否在eleSet中："+jedis.sismember("eleSet", "e3"));System.out.println("e1是否在eleSet中："+jedis.sismember("eleSet", "e1"));System.out.println("e1是否在eleSet中："+jedis.sismember("eleSet", "e5"));System.out.println("=================================");System.out.println(jedis.sadd("eleSet1", "e1","e2","e4","e3","e0","e8","e7","e5"));System.out.println(jedis.sadd("eleSet2", "e1","e2","e4","e3","e0","e8"));System.out.println("將eleSet1中刪除e1并存入eleSet3中："+jedis.smove("eleSet1", "eleSet3", "e1"));//移到集合元素System.out.println("將eleSet1中刪除e2并存入eleSet3中："+jedis.smove("eleSet1", "eleSet3", "e2"));System.out.println("eleSet1中的元素："+jedis.smembers("eleSet1"));System.out.println("eleSet3中的元素："+jedis.smembers("eleSet3"));System.out.println("============集合運算=================");System.out.println("eleSet1中的元素："+jedis.smembers("eleSet1"));System.out.println("eleSet2中的元素："+jedis.smembers("eleSet2"));System.out.println("eleSet1和eleSet2的交集:"+jedis.sinter("eleSet1","eleSet2"));System.out.println("eleSet1和eleSet2的并集:"+jedis.sunion("eleSet1","eleSet2"));System.out.println("eleSet1和eleSet2的差集:"+jedis.sdiff("eleSet1","eleSet2"));//eleSet1中有，eleSet2中沒有jedis.sinterstore("eleSet4","eleSet1","eleSet2");//求交集并將交集保存到dstkey的集合System.out.println("eleSet4中的元素："+jedis.smembers("eleSet4"));}
}

對hash操作的命令

package com.jihu.base;import redis.clients.jedis.Jedis;import java.util.HashMap;
import java.util.Map;public class TestHash {public static void main(String[] args) {Jedis jedis = new Jedis("192.168.56.130",6379);jedis.auth("123456");jedis.flushDB();Map<String,String> map = new HashMap<String,String>();map.put("key1","value1");map.put("key2","value2");map.put("key3","value3");map.put("key4","value4");//添加名稱為hash（key）的hash元素jedis.hmset("hash",map);//向名稱為hash的hash中添加key為key5，value為value5元素jedis.hset("hash", "key5", "value5");System.out.println("散列hash的所有鍵值對為："+jedis.hgetAll("hash"));//return Map<String,String>System.out.println("散列hash的所有鍵為："+jedis.hkeys("hash"));//return Set<String>System.out.println("散列hash的所有值為："+jedis.hvals("hash"));//return List<String>System.out.println("將key6保存的值加上一個整數，如果key6不存在則添加key6："+jedis.hincrBy("hash", "key6", 6));System.out.println("散列hash的所有鍵值對為："+jedis.hgetAll("hash"));System.out.println("將key6保存的值加上一個整數，如果key6不存在則添加key6："+jedis.hincrBy("hash", "key6", 3));System.out.println("散列hash的所有鍵值對為："+jedis.hgetAll("hash"));System.out.println("刪除一個或者多個鍵值對："+jedis.hdel("hash", "key2"));System.out.println("散列hash的所有鍵值對為："+jedis.hgetAll("hash"));System.out.println("散列hash中鍵值對的個數："+jedis.hlen("hash"));System.out.println("判斷hash中是否存在key2："+jedis.hexists("hash","key2"));System.out.println("判斷hash中是否存在key3："+jedis.hexists("hash","key3"));System.out.println("獲取hash中的值："+jedis.hmget("hash","key3"));System.out.println("獲取hash中的值："+jedis.hmget("hash","key3","key4"));}
}

對Zset操作的命令

public class TestZSet {public static void main(String[] args) {Jedis jedis = JedisEnumSingleton.INSTANCE.getJedis();System.out.println("jedis.flushDB(): " + jedis.flushDB());System.out.println("==================向集合中添加元素==================");System.out.println(jedis.zadd("myset", 1, "one"));HashMap<String, Double> scoreMembers = new HashMap<String, Double>();scoreMembers.put("two", 2.0);scoreMembers.put("three", 3.0);System.out.println(jedis.zadd("myset", scoreMembers));System.out.println("集合中所有元素: " + jedis.zrange("myset", 0, -1));System.out.println("===================排序============================");jedis.zadd("salary", 2500, "zhangsan");jedis.zadd("salary", 1500, "lisi");jedis.zadd("salary", 4500, "wangwu");System.out.println("jedis.zrangeWithScores：" + jedis.zrangeWithScores("salary",0, -1));System.out.println("jedis.zrangeByScore：" + jedis.zrangeByScore("salary",Integer.MIN_VALUE, Integer.MAX_VALUE));System.out.println("jedis.zrangeByScore：" + jedis.zrangeByScoreWithScores("salary",Integer.MIN_VALUE, Integer.MAX_VALUE));System.out.println("jedis.zrevrange：" + jedis.zrevrange("salary",0, -1));System.out.println("jedis.zrevrangeByScore：" + jedis.zrevrangeByScore("salary",Integer.MAX_VALUE, Integer.MIN_VALUE));System.out.println("jedis.zrevrangeByScoreWithScores：" + jedis.zrevrangeByScoreWithScores("salary",Integer.MAX_VALUE, Integer.MIN_VALUE));}
}

事務

package com.jihu;
import com.alibaba.fastjson.JSONObject;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Transaction;public class TestTX {public static void main(String[] args) {Jedis jedis = new Jedis("192.168.56.130",6379);jedis.auth("123456");jedis.flushDB();JSONObject jsonObject = new JSONObject();jsonObject.put("hello","world");jsonObject.put("name","jihu");//開啟事務Transaction multi = jedis.multi();String result = jsonObject.toJSONString();//jedis.watch(result) //給result加樂觀鎖try {multi.set("user1",result);multi.set("user2",result);int i = 1/0; //代碼拋出異常，事務執行失敗multi.exec();  //執行事務} catch (Exception exception) {multi.discard();  //放棄事務exception.printStackTrace();}finally {System.out.println(jedis.get("user1"));System.out.println(jedis.get("user2"));jedis.close(); //關閉連接}}
}

在 Multi 中時無法使用 Jedis:

Exception in thread "main" redis.clients.jedis.exceptions.JedisDataException: Cannot use Jedis when in Multi. Please use Transaction or reset jedis state.

五、SpringBoot整合

SpringBoot 操作數據：spring-data jpa jdbc mongodb redis！

SpringData 也是和 SpringBoot 齊名的項目！

說明：在 SpringBoot2.x 之后，原來使用的jedis 被替換為了 lettuce?

jedis : 采用的直連，多個線程操作的話，是不安全的，如果想要避免不安全的，使用 jedis pool 連接池！更像 BIO 模式

lettuce : 采用netty，實例可以再多個線程中進行共享，不存在線程不安全的情況！可以減少線程數據了，更像 NIO 模式

源碼分析

springboot所有的配置類都有一個自動配置類：RedisAutoConfiguration：

@Bean // 我們可以自己定義一個redisTemplate來替換這個默認的！
@ConditionalOnMissingBean(name = "redisTemplate") // 不存在才生效
public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) throws UnknownHostException {
// 默認的 RedisTemplate 沒有過多的設置，redis 對象都是需要序列化！
// 兩個泛型都是 Object, Object 的類型，我們后使用需要強制轉換 <String, Object>RedisTemplate<Object, Object> template = new RedisTemplate<>();template.setConnectionFactory(redisConnectionFactory);return template;
}@Bean
@ConditionalOnMissingBean // 由于 String 是redis中最常使用的類型，所以說單獨提出來了一個bean！
public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory redisConnectionFactory) throws UnknownHostException {StringRedisTemplate template = new StringRedisTemplate();template.setConnectionFactory(redisConnectionFactory);return template;
}

整合測試

1.導入依賴

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

2.配置連接

#springboot所有的配置類都有一個自動配置類 ：RedisAutoConfiguration
# （RedisAutoConfiguration它在RedisAutoConfiguration/META-INF/spring.factories文件中）
#自動配置類都會綁定一個 properties配置文件 RedisProperties#配置Redis
spring.redis.host=192.168.56.130
spring.redis.port=6379
spring.redis.password=123456

3.測試

package com.jihu;import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisTemplate;@SpringBootTest
class Redis02SpringbootApplicationTests {@Autowiredprivate RedisTemplate redisTemplate;@Testvoid contextLoads() {//redisTemplate  操作不同的數據類型，api和我們的指令是一樣的//redisTemplate.opsForValue()  表示操作字符串 類似String//redisTemplate.opsForList()  表示操作List ,類似list//等等
//       redisTemplate.opsForValue();
//        redisTemplate.opsForList();
//        redisTemplate.opsForSet();
//        redisTemplate.opsForZSet();
//        redisTemplate.opsForHash();
//        redisTemplate.opsForGeo();
//        redisTemplate.opsForHyperLogLog();redisTemplate.opsForList();//除了基本的操作，我們常用的方法都可以直接通過redisTemplate操作，比如事務和基本的CRUD//獲取redis的連接對象//RedisConnection connection = redisTemplate.getConnectionFactory().getConnection();//connection.flushDb();//connection.flushAll();redisTemplate.opsForValue().set("mykey","jihu");System.out.println(redisTemplate.opsForValue().get("mykey"));}
}

4.測試結果

此時我們回到Redis查看數據時候，驚奇發現全是亂碼，可是程序中可以正常輸出。這時候就關系到存儲對象的序列化問題，在網絡中傳輸的對象也是一樣需要序列化，否者就全是亂碼。

RedisTemplate內部的序列化配置是這樣的

默認的序列化器是采用JDK序列化器

對呀對象的保存

如果不序列化這樣會報錯

這里序列化之后就好了

自定義RedisTemplate 序列化配置

1.自定義RedisTemplate序列化配置

我們來編寫一個自己的 RedisTemplete

在這里插入圖片描述

package com.fatfish.config;import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;@Configuration
public class RedisConfig {// 這是我給大家寫好的一個固定模板，大家在企業中，拿去就可以直接使用！// 自己定義了一個 RedisTemplate@Bean@SuppressWarnings("all")public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {// 我們為了自己開發方便，一般直接使用 <String, Object>RedisTemplate<String, Object> template = new RedisTemplate<String,Object>();template.setConnectionFactory(factory);// Json序列化配置Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = newJackson2JsonRedisSerializer(Object.class);ObjectMapper om = new ObjectMapper();om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);jackson2JsonRedisSerializer.setObjectMapper(om);// String 的序列化StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();// key采用String的序列化方式template.setKeySerializer(stringRedisSerializer);// hash的key也采用String的序列化方式template.setHashKeySerializer(stringRedisSerializer);// value序列化方式采用jacksontemplate.setValueSerializer(jackson2JsonRedisSerializer);// hash的value序列化方式采用jacksontemplate.setHashValueSerializer(jackson2JsonRedisSerializer);template.afterPropertiesSet();return template;}
}

再查看就不是亂碼了

    @Resource@Qualifier("redisTemplate")private RedisTemplate redisTemplate;

@Qualifier("redisTemplate": 區分多個同名字的 Bean

2.RedisUtil配置(CRUD操作string，map,list,set)

package com.jihu.config.utils;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;//在我們真實的開發中，或者你們的公司，一般都可以看到一個公司自己封裝的RedisUtil
@Component
public final class RedisUtil {@Autowiredprivate RedisTemplate<String, Object> redisTemplate;// =============================common============================/*** 指定緩存失效時間* @param key  鍵* @param time 時間(秒)*/public boolean expire(String key, long time) {try {if (time > 0) {redisTemplate.expire(key, time, TimeUnit.SECONDS);}return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 根據key 獲取過期時間* @param key 鍵 不能為null* @return 時間(秒) 返回0代表為永久有效*/public long getExpire(String key) {return redisTemplate.getExpire(key, TimeUnit.SECONDS);}/*** 判斷key是否存在* @param key 鍵* @return true 存在 false不存在*/public boolean hasKey(String key) {try {return redisTemplate.hasKey(key);} catch (Exception e) {e.printStackTrace();return false;}}/*** 刪除緩存* @param key 可以傳一個值 或多個*/@SuppressWarnings("unchecked")public void del(String... key) {if (key != null && key.length > 0) {if (key.length == 1) {redisTemplate.delete(key[0]);} else {redisTemplate.delete((Collection<String>) CollectionUtils.arrayToList(key));}}}// ============================String=============================/*** 普通緩存獲取* @param key 鍵* @return 值*/public Object get(String key) {return key == null ? null : redisTemplate.opsForValue().get(key);}/*** 普通緩存放入* @param key   鍵* @param value 值* @return true成功 false失敗*/public boolean set(String key, Object value) {try {redisTemplate.opsForValue().set(key, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 普通緩存放入并設置時間* @param key   鍵* @param value 值* @param time  時間(秒) time要大于0 如果time小于等于0 將設置無限期* @return true成功 false 失敗*/public boolean set(String key, Object value, long time) {try {if (time > 0) {redisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS);} else {set(key, value);}return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 遞增* @param key   鍵* @param delta 要增加幾(大于0)*/public long incr(String key, long delta) {if (delta < 0) {throw new RuntimeException("遞增因子必須大于0");}return redisTemplate.opsForValue().increment(key, delta);}/*** 遞減* @param key   鍵* @param delta 要減少幾(小于0)*/public long decr(String key, long delta) {if (delta < 0) {throw new RuntimeException("遞減因子必須大于0");}return redisTemplate.opsForValue().increment(key, -delta);}// ================================Map=================================/*** HashGet* @param key  鍵 不能為null* @param item 項 不能為null*/public Object hget(String key, String item) {return redisTemplate.opsForHash().get(key, item);}/*** 獲取hashKey對應的所有鍵值* @param key 鍵* @return 對應的多個鍵值*/public Map<Object, Object> hmget(String key) {return redisTemplate.opsForHash().entries(key);}/*** HashSet* @param key 鍵* @param map 對應多個鍵值*/public boolean hmset(String key, Map<String, Object> map) {try {redisTemplate.opsForHash().putAll(key, map);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** HashSet 并設置時間* @param key  鍵* @param map  對應多個鍵值* @param time 時間(秒)* @return true成功 false失敗*/public boolean hmset(String key, Map<String, Object> map, long time) {try {redisTemplate.opsForHash().putAll(key, map);if (time > 0) {expire(key, time);}return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 向一張hash表中放入數據,如果不存在將創建** @param key   鍵* @param item  項* @param value 值* @return true 成功 false失敗*/public boolean hset(String key, String item, Object value) {try {redisTemplate.opsForHash().put(key, item, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 向一張hash表中放入數據,如果不存在將創建** @param key   鍵* @param item  項* @param value 值* @param time  時間(秒) 注意:如果已存在的hash表有時間,這里將會替換原有的時間* @return true 成功 false失敗*/public boolean hset(String key, String item, Object value, long time) {try {redisTemplate.opsForHash().put(key, item, value);if (time > 0) {expire(key, time);}return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 刪除hash表中的值** @param key  鍵 不能為null* @param item 項 可以使多個 不能為null*/public void hdel(String key, Object... item) {redisTemplate.opsForHash().delete(key, item);}/*** 判斷hash表中是否有該項的值** @param key  鍵 不能為null* @param item 項 不能為null* @return true 存在 false不存在*/public boolean hHasKey(String key, String item) {return redisTemplate.opsForHash().hasKey(key, item);}/*** hash遞增 如果不存在,就會創建一個 并把新增后的值返回** @param key  鍵* @param item 項* @param by   要增加幾(大于0)*/public double hincr(String key, String item, double by) {return redisTemplate.opsForHash().increment(key, item, by);}/*** hash遞減** @param key  鍵* @param item 項* @param by   要減少記(小于0)*/public double hdecr(String key, String item, double by) {return redisTemplate.opsForHash().increment(key, item, -by);}// ============================set=============================/*** 根據key獲取Set中的所有值* @param key 鍵*/public Set<Object> sGet(String key) {try {return redisTemplate.opsForSet().members(key);} catch (Exception e) {e.printStackTrace();return null;}}/*** 根據value從一個set中查詢,是否存在** @param key   鍵* @param value 值* @return true 存在 false不存在*/public boolean sHasKey(String key, Object value) {try {return redisTemplate.opsForSet().isMember(key, value);} catch (Exception e) {e.printStackTrace();return false;}}/*** 將數據放入set緩存** @param key    鍵* @param values 值 可以是多個* @return 成功個數*/public long sSet(String key, Object... values) {try {return redisTemplate.opsForSet().add(key, values);} catch (Exception e) {e.printStackTrace();return 0;}}/*** 將set數據放入緩存** @param key    鍵* @param time   時間(秒)* @param values 值 可以是多個* @return 成功個數*/public long sSetAndTime(String key, long time, Object... values) {try {Long count = redisTemplate.opsForSet().add(key, values);if (time > 0)expire(key, time);return count;} catch (Exception e) {e.printStackTrace();return 0;}}/*** 獲取set緩存的長度** @param key 鍵*/public long sGetSetSize(String key) {try {return redisTemplate.opsForSet().size(key);} catch (Exception e) {e.printStackTrace();return 0;}}/*** 移除值為value的** @param key    鍵* @param values 值 可以是多個* @return 移除的個數*/public long setRemove(String key, Object... values) {try {Long count = redisTemplate.opsForSet().remove(key, values);return count;} catch (Exception e) {e.printStackTrace();return 0;}}// ===============================list=================================/*** 獲取list緩存的內容** @param key   鍵* @param start 開始* @param end   結束 0 到 -1代表所有值*/public List<Object> lGet(String key, long start, long end) {try {return redisTemplate.opsForList().range(key, start, end);} catch (Exception e) {e.printStackTrace();return null;}}/*** 獲取list緩存的長度** @param key 鍵*/public long lGetListSize(String key) {try {return redisTemplate.opsForList().size(key);} catch (Exception e) {e.printStackTrace();return 0;}}/*** 通過索引 獲取list中的值** @param key   鍵* @param index 索引 index>=0時， 0 表頭，1 第二個元素，依次類推；index<0時，-1，表尾，-2倒數第二個元素，依次類推*/public Object lGetIndex(String key, long index) {try {return redisTemplate.opsForList().index(key, index);} catch (Exception e) {e.printStackTrace();return null;}}/*** 將list放入緩存** @param key   鍵* @param value 值*/public boolean lSet(String key, Object value) {try {redisTemplate.opsForList().rightPush(key, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 將list放入緩存* @param key   鍵* @param value 值* @param time  時間(秒)*/public boolean lSet(String key, Object value, long time) {try {redisTemplate.opsForList().rightPush(key, value);if (time > 0)expire(key, time);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 將list放入緩存** @param key   鍵* @param value 值* @return*/public boolean lSet(String key, List<Object> value) {try {redisTemplate.opsForList().rightPushAll(key, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 將list放入緩存** @param key   鍵* @param value 值* @param time  時間(秒)* @return*/public boolean lSet(String key, List<Object> value, long time) {try {redisTemplate.opsForList().rightPushAll(key, value);if (time > 0)expire(key, time);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 根據索引修改list中的某條數據** @param key   鍵* @param index 索引* @param value 值* @return*/public boolean lUpdateIndex(String key, long index, Object value) {try {redisTemplate.opsForList().set(key, index, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 移除N個值為value** @param key   鍵* @param count 移除多少個* @param value 值* @return 移除的個數*/public long lRemove(String key, long count, Object value) {try {Long remove = redisTemplate.opsForList().remove(key, count, value);return remove;} catch (Exception e) {e.printStackTrace();return 0;}}}

六、Redis.conf 詳解

1.容量單位不區分大小寫，G和GB有區別

2.可以使用 include 組合多個配置文件

3.網絡配置

protected-mode no  #保護模式
port 6379  #端口設置

4.通用配置(GENERAL)

daemonize yes  #以守護進程的方式運行，默認是no ，我們需要自己開啟為yespidfile /var/run/redis_6379.pid #如果以后臺的方式運行，我們就需要指定一個pid文件# Specify the server verbosity level.
# This can be one of:
# debug (a lot of information, useful for development/testing)
# verbose (many rarely useful info, but not a mess like the debug level)
# notice (moderately verbose, what you want in production probably)
# warning (only very important / critical messages are logged)
loglevel notice   #日志級別
logfile ""  #日志的文件位置名
databases 16  #數據庫的數量，默認是16個數據庫
always-show-logo no  #是否總是顯示log

5.快照(SNAPSHOTTING) rdb配置

持久化，在規定的時間內，執行了多少次操作，則會持久化到文件 .rdb .aof

redis是內存數據庫，如果沒有持久化，那么數據斷電及失。

#在3600s內，如果至少有 1 key進行了修改，我們將進行持久化操作save 3600 1#在300s內，如果至少有 100 key進行了修改，我們將進行持久化操作save 300 100#在 60s 內，如果至少有 10000 key進行了修改，我們將進行持久化操作save 60 10000#我們之后學習持久化，會自己定義這個測試stop-writes-on-bgsave-error yes  #持久化如果出錯，是否還需要繼續工作。rdbcompression yes  #是否壓縮rdb文件，需要消耗一些 cpu資源rdbchecksum yes  #保存rdb文件的視覺，進行錯誤的檢查效驗dir ./  #rdb文件保存的目錄

REPLICATION 主從復制，我們后面講解主從復制的時候再進行詳解

在Redis的配置文件（redis.conf）中，可以配置RDB持久化相關的參數：

save m n：設置觸發RDB持久化的條件和頻率。例如，save 900 1表示在900秒（15分鐘）內如果至少有1個key被修改，則執行BGSAVE命令生成快照。
stop-writes-on-bgsave-error yes/no：當BGSAVE命令執行失敗時，是否停止寫入操作。
rdbcompression yes/no：是否對RDB文件進行壓縮存儲，節省磁盤空間但會增加CPU消耗。
rdbchecksum yes/no：在寫入RDB文件時是否進行數據校驗。
dir /path/to/directory：指定RDB文件的存儲目錄。
dbfilename dump.rdb：指定RDB文件的名稱。

6.SECURITY 安全

 1.可以再redis.conf配置文件里面修改密碼requirepass 123456 #可以在這里設置redis的密碼，默認是沒有密碼的2.可以用命令設置密碼
127.0.0.1:6379> ping
(error) NOAUTH Authentication required.  #設置密碼后發現所有的命令都沒有權限了
127.0.0.1:6379> auth 123456   #使用密碼進行登錄
OK
127.0.0.1:6379> config get requirepass  #查看redis的密碼
1) "requirepass"
2) "123456"
127.0.0.1:6379> config set requirepass "1234567"   #用命令設置redis的密碼

7.CLIENTS 限制（客戶端連接相關）

maxclients 10000  #設置能連上redis的最大客戶端的數量maxmemory <bytes> # redis 配置最大的內存容量maxmemory-policy noeviction # 內存到達上限之后的處理策略
6種策略1、volatile-lru：只對設置了過期時間的key進行LRU（默認值）2、allkeys-lru ： 刪除lru算法的key3、volatile-random：隨機刪除即將過期key4、allkeys-random：隨機刪除5、volatile-ttl ： 刪除即將過期的6、noeviction ： 永不過期，返回錯誤# volatile-lru -> 在具有過期設置的鍵中使用近似 LRU 進行驅逐。
# allkeys-lru -> 使用近似 LRU 驅逐任何鍵。
# volatile-lfu -> 在具有過期設置的鍵中使用近似 LFU 進行驅逐。
# allkeys-lfu -> 使用近似 LFU 驅逐任何鍵。
# volatile-random -> 從設置了過期時間的密鑰中刪除一個隨機密鑰。
# allkeys-random -> 刪除隨機密鑰，任何密鑰。
# volatile-ttl -> 刪除過期時間最近的密鑰（次要 TTL）
# noeviction -> 不驅逐任何東西，只是在寫操作上返回錯誤。

8.APPEND ONLY 模式 aof配置

appendonly no # 默認是不開啟aof模式的，默認是使用rdb方式持久化的，在大部分所有的情況下，rdb完全夠用！
appendfilename "appendonly.aof" # 持久化的文件的名字# appendfsync always # 每次修改都會 sync。消耗性能
appendfsync everysec # 每秒執行一次 sync，可能會丟失這1s的數據！
# appendfsync no # 不執行 sync，這個時候操作系統自己同步數據，速度最快！

七、Redis持久化——RDB

面試和工作，持久化都是重點！

Redis 是內存數據庫，如果不將內存中的數據庫狀態保存到磁盤，那么一旦服務器進程退出，服務器中的數據庫狀態也會消失。所以 Redis 提供了持久化功能！

持久化：在指定時間間隔內將內存數據存入磁盤中，斷電也能恢復數據，使用快照文件讀到內存中。

RDB：讀寫文件

RDB(Redis DataBase)

什么是RDB

用在主從復制中，rdb就是備用的，在從機上面。

在指定的時間間隔內將內存中的數據集快照寫入磁盤，也就是行話講的Snapshot快照，它恢復時是將快照文件直接讀到內存里。

Redis會單獨創建（fork）一個子進程來進行持久化，會先將數據寫入到一個臨時文件中，待持久化過程都結束了，再用這個臨時文件替換上次持久化好的文件。整個過程中，主進程是不進行任何IO操作的。這就確保了極高的性能。如果需要進行大規模數據的恢復，且對于數據恢復的完整性不是非常敏感，那 RDB方式要比AOF方式更加的高效。RDB的缺點是最后一次持久化后的數據可能丟失。我們默認的就是 RDB，一般情況下不需要修改這個配置！

有時候在生產環境我們會將這個文件進行備份！

rdb保存的文件是dump.rdb 都是在我們的配置文件中快照中進行配置的！

觸發機制

save的規則滿足的情況下，會自動觸發rdb原則
執行flushall命令，也會觸發我們的rdb原則
退出redis，也會自動產生rdb文件

備份就自動生成一個 dump.rdb

1.save

使用 save 命令，會立刻對當前內存中的數據進行持久化 ,但是會阻塞，也就是不接受其他操作了；

由于 save 命令是同步命令，會占用Redis的主進程。若Redis數據非常多時，save命令執行速度會非常慢，阻塞所有客戶端的請求。

示意圖：

2.flushall命令

flushall 命令也會觸發持久化；

觸發持久化規則
滿足配置條件中的觸發條件；

可以通過配置文件對 Redis 進行設置，讓它在“ N 秒內數據集至少有 M 個改動”這一條件被滿足時，自動進行數據集保存操作。

3.bgsave

bgsave 是異步進行，進行持久化的時候，redis 還可以將繼續響應客戶端請求；

示意圖：

bgsave和save對比

命令	save	bgsave
IO類型	同步	異步
阻塞？	是	是（阻塞發生在fock()，通常非常快）
復雜度	O(n)	O(n)
優點	不會消耗額外的內存	不阻塞客戶端命令
缺點	阻塞客戶端命令	需要fock子進程，消耗內存

SAVE:

描述：SAVE 命令會阻塞 Redis 服務器進程，直到持久化過程完成為止。它會將當前的數據集以同步阻塞方式保存到磁盤上的一個文件中，這個過程會在主進程中進行。
特點：
阻塞：SAVE 命令會阻塞 Redis 服務器進程，直到持久化過程完成。
同步：持久化操作是同步執行的，因此會導致服務器停止服務一段時間。

優點：
簡單：SAVE 是一個簡單的命令，容易理解和使用。
數據一致性：在持久化期間，Redis 不接受任何請求，因此可以保證數據一致性。

缺點：
阻塞：持久化過程會阻塞 Redis 服務器，導致服務器暫時無法處理請求，影響性能。
性能問題：對于大型數據庫來說，保存整個數據集可能會消耗大量時間和資源。

適用場合：適用于小型數據庫，或者在數據量不大且可以容忍服務器停止服務一段時間的情況下。

BGSAVE:

描述：BGSAVE 命令會派生一個子進程，由子進程負責將數據集保存到磁盤上的一個文件中。父進程繼續處理請求，不會被阻塞。
特點：
后臺執行：BGSAVE 是一個后臺執行的命令，不會阻塞 Redis 服務器主進程。
異步：持久化操作是異步執行的，不會影響服務器的響應速度。

優點：
非阻塞：BGSAVE 不會阻塞服務器進程，因此不會影響 Redis 服務器的性能。
異步：持久化過程在后臺執行，不會影響服務器響應請求的能力。

缺點：
內存占用：BGSAVE 使用子進程進行持久化操作，可能會消耗一定的內存資源。
可能會失敗：在某些情況下，BGSAVE 操作可能會因為子進程出錯或者其他原因失敗。

適用場合：適用于大型數據庫，或者在不能容忍服務器停止服務的情況下，因為 BGSAVE 不會阻塞 Redis 服務器的主進程。

總的來說，如果你的數據庫較小，且可以容忍短暫的服務停止，那么使用 SAVE 是一個簡單的選擇。但如果你的數據庫較大，或者不能容忍服務停止，那么最好使用 BGSAVE。

如果恢復rdb文件！

1、只需要將rdb文件放在我們redis啟動目錄就可以，redis啟動的時候會自動檢查dump.rdb 恢復其中的數據！
2、查看需要存在的位置

127.0.0.1:6379> config get dir
1) "dir"
2) "/usr/local/bin" # 如果在這個目錄下存在 dump.rdb 文件，啟動就會自動恢復其中的數據

優缺點

優點：

適合大規模的數據恢復
對數據的完整性要求不高

缺點：

需要一定的時間間隔進行操作，如果redis意外宕機了，這個最后一次修改的數據就沒有了。
fork進程的時候，會占用一定的內容空間。

優點：

RDB持久化生成的快照文件緊湊，適合用于備份和恢復數據。
在數據恢復時，由于快照是全量的，恢復速度相對較快。

缺點：

RDB持久化是全量備份，如果Redis進程意外終止，可能會丟失最后一次快照生成后的數據。
對于大型數據集，生成快照會消耗較多的CPU和IO資源，可能會影響Redis的性能。
RDB持久化無法實現實時備份，只能通過定期生成快照的方式來保證數據的持久化。

總的來說，RDB持久化適合對數據一致性要求不是特別高、需要定期備份的場景。對于要求實時持久化并且能夠容忍一定數據丟失的場景，AOF持久化更為適合。

八、Redis 持久化——AOF

AOF（Append-Only File）(記錄文件)

將我們的所有命令都記錄下來，history，恢復的時候就把這個文件全部在執行一遍！

AOF 是什么

以日志的形式來記錄每個寫操作，將Redis執行過的所有指令記錄下來（讀操作不記錄），只許追加文件但不可以改寫文件，redis啟動之初會讀取該文件重新構建數據，換言之，redis重啟的話就根據日志文件的內容將寫指令從前到后執行一次以完成數據的恢復工作

Aof保存的是 appendonly.aof 文件

append

默認是不開啟的，我們需要手動進行配置！我們只需要將 appendonly 改為yes就開啟了 aof！

重啟，redis 就可以生效了！如果這個 aof 文件有錯位，這時候 redis 是啟動不起來的嗎，我們需要修復這個aof文件

開啟后會生成appendonly.aof文件

在里面寫入數據后

看 appendonly.aof 里面會有寫數據的記錄

如果這個aof文件有錯誤，這時候redis是啟動不起來的，需要我們修復這個aof文件

redis給我們提供了一個工具 redis-check-aof --fix

aof配置文件出錯，連接時會出現這個問題

用redis-check-aof --fix 進行恢復

如果文件正常，重啟就可以直接恢復了

重寫規則說明

aof 默認就是文件的無限追加，文件會越來越大！

如果 aof 文件大于 64m，太大了！ fork一個新的進程來將我們的文件進行重寫！

優點和缺點！

appendonly no # 默認是不開啟aof模式的，默認是使用rdb方式持久化的，在大部分所有的情況下，
rdb完全夠用！
appendfilename "appendonly.aof" # 持久化的文件的名字
# appendfsync always # 每次修改都會 sync。消耗性能
appendfsync everysec # 每秒執行一次 sync，可能會丟失這1s的數據！
# appendfsync no # 不執行 sync，這個時候操作系統自己同步數據，速度最快！
# rewrite 重寫，

優點

1、每一次修改都同步，文件的完整會更加好！

2、每秒同步一次，可能會丟失一秒的數據

3、從不同步，效率最高的！

缺點

1、相對于數據文件來說，aof遠遠大于 rdb，修復的速度也比 rdb慢！

2、Aof 運行效率也要比 rdb 慢，所以我們redis默認的配置就是rdb持久化！

擴展：

1、RDB 持久化方式能夠在指定的時間間隔內對你的數據進行快照存儲

2、AOF 持久化方式記錄每次對服務器寫的操作，當服務器重啟的時候會重新執行這些命令來恢復原始的數據，AOF命令以Redis 協議追加保存每次寫的操作到文件末尾，Redis還能對AOF文件進行后臺重寫，使得AOF文件的體積不至于過大。

3、只做緩存，如果你只希望你的數據在服務器運行的時候存在，你也可以不使用任何持久化

4、同時開啟兩種持久化方式

在這種情況下，當redis重啟的時候會優先載入AOF文件來恢復原始的數據，因為在通常情況下AOF 文件保存的數據集要比RDB文件保存的數據集要完整。
RDB 的數據不實時，同時使用兩者時服務器重啟也只會找AOF文件，那要不要只使用AOF呢？作者建議不要，因為RDB更適合用于備份數據庫（AOF在不斷變化不好備份），快速重啟，而且不會有 AOF可能潛在的Bug，留著作為一個萬一的手段。

5、性能建議

因為RDB文件只用作后備用途，建議只在Slave上持久化RDB文件，而且只要15分鐘備份一次就夠了，只保留 save 900 1 這條規則。
如果Enable AOF ，好處是在最惡劣情況下也只會丟失不超過兩秒數據，啟動腳本較簡單只load自己的AOF文件就可以了，代價一是帶來了持續的IO，二是AOF rewrite 的最后將 rewrite 過程中產生的新數據寫到新文件造成的阻塞幾乎是不可避免的。只要硬盤許可，應該盡量減少AOF rewrite 的頻率，AOF重寫的基礎大小默認值64M太小了，可以設到5G以上，默認超過原大小100%大小重寫可以改到適當的數值。
如果不Enable AOF ，僅靠 Master-Slave Repllcation 實現高可用性也可以，能省掉一大筆IO，也減少了rewrite時帶來的系統波動。代價是如果Master/Slave 同時倒掉，會丟失十幾分鐘的數據，啟動腳本也要比較兩個 Master/Slave 中的 RDB文件，載入較新的那個，微博就是這種架構。

優點：

可讀性高：AOF文件是以文本形式保存的，易于理解和調試。
容災性強：AOF文件記錄了所有寫操作命令，因此可以更可靠地恢復數據。
適用于追加操作：AOF采用追加方式記錄寫操作，因此對于磁盤的IO消耗相對較低。

缺點：

文件較大：由于AOF文件保存了每個寫操作命令，因此可能會比RDB生成的快照文件更大。
恢復速度較慢：由于Redis在啟動時需要重新執行AOF文件中的所有寫操作，因此恢復速度可能比RDB快照方式慢。
重寫可能會耗時：AOF重寫過程可能會耗費一定的時間和資源，尤其是在數據集較大時。

配置AOF持久化

在Redis的配置文件中，可以通過以下參數配置AOF持久化：

appendonly yes/no：是否開啟AOF持久化，默認為no。
appendfilename：AOF文件的文件名，默認為appendonly.aof。
appendfsync always/everysec/no：控制何時將寫入操作同步到磁盤，可選值有always（每個寫入操作都會同步到磁盤）、everysec（每秒同步一次）和no（不同步）。
auto-aof-rewrite-percentage：觸發AOF重寫的百分比，默認為100。
auto-aof-rewrite-min-size：AOF文件大小超過該值時，觸發AOF重寫，默認為64MB。

適用場景

數據可靠性要求高：AOF持久化記錄了每個寫操作命令，因此在數據可靠性要求較高的場景下更為適用。
追加操作頻繁：AOF持久化采用追加方式記錄寫操作，適用于寫操作頻繁的場景。
可讀性要求高：AOF文件是以文本形式保存的，易于理解和調試，適用于需要對持久化數據進行查看和修改的場景。

AOF重寫是Redis中的一項機制，旨在解決AOF持久化過程中可能出現的文件膨脹和性能問題。在AOF持久化過程中，隨著Redis接收到寫操作命令的不斷累積，AOF文件會不斷增長，可能會導致文件體積過大，影響性能和占用過多的磁盤空間。AOF重寫機制的作用是生成一個新的AOF文件，該文件與當前數據集的狀態相同，但是文件體積更小，仍然能夠完整地恢復數據集。

AOF重寫的工作原理

生成新的AOF文件：

AOF重寫過程由后臺線程完成，不會阻塞主進程。該線程會遍歷當前數據集的所有鍵值對，并生成一系列寫命令，這些命令可以完整地重現當前數據集的狀態。

記錄生成過程：

在生成新的AOF文件時，Redis會記錄當前服務器接收到的所有寫命令。這些命令不是原始的客戶端寫命令，而是生成新AOF文件所需的最少的寫命令序列。

保留最近一次完整寫命令序列：

在AOF重寫過程中，Redis會記錄生成新AOF文件所需的寫命令序列。一旦重寫完成，Redis會將這個寫命令序列寫入新的AOF文件中，代替舊的AOF文件。

切換AOF文件：

當新AOF文件生成完畢并寫入完成后，Redis會原子性地將舊的AOF文件替換為新的AOF文件。這樣就完成了AOF重寫。

AOF重寫的優點

降低AOF文件體積：由于AOF重寫只記錄當前數據集的狀態，因此生成的新AOF文件體積通常會小于原始AOF文件。
減少寫操作記錄：AOF重寫只記錄生成新AOF文件所需的最少的寫命令序列，而不是全部歷史寫命令，因此能夠降低AOF文件的體積。
提升性能：AOF重寫過程由后臺線程完成，不會阻塞主進程，因此不會影響Redis的正常響應速度。

auto-aof-rewrite-percentage：觸發AOF重寫的AOF文件體積增長百分比，默認為100%。

這個配置項指定了觸發AOF重寫的AOF文件體積增長的百分比閾值。默認情況下，這個值是100%，意味著當AOF文件的體積增長到原始AOF文件的大小的兩倍時，Redis就會觸發AOF重寫過程。

舉個例子來說，假設初始時AOF文件的大小是100MB。如果設置了默認的100%閾值，那么當AOF文件的大小增長到200MB時，就會觸發AOF重寫。這樣做的目的是為了在AOF文件不斷增長時，及時觸發AOF重寫，以防止AOF文件變得過大影響性能或占用過多的磁盤空間。

九、RDB和AOF選擇

RDB 和 AOF 對比

	RDB	AOF
啟動優先級	低	高
體積	小	大
恢復速度	快	慢
數據安全性	丟數據	根據策略決定

如何選擇使用哪種持久化方式？

一般來說，如果想達到足以媲美 PostgreSQL 的數據安全性，你應該同時使用兩種持久化功能。

如果你非常關心你的數據，但仍然可以承受數分鐘以內的數據丟失，那么你可以只使用 RDB 持久化。

有很多用戶都只使用 AOF 持久化，但并不推薦這種方式：因為定時生成 RDB 快照（snapshot）非常便于進行數據庫備份，并且 RDB 恢復數據集的速度也要比 AOF 恢復的速度要快。

首先，需要創建 RDB 文件的備份，通常是最新的 dump.rdb 文件。這是為了確保數據安全。

將備份文件保存在一個安全的地方，以防意外情況發生時需要還原數據。

通過執行以下兩個命令來完成切換持久化方式：首先，使用 CONFIG SET appendonly yes 命令開啟 AOF 功能；然后，使用 CONFIG SET save "" 命令關閉 RDB 功能。這樣做之后，Redis 就會開始將寫入命令追加到 AOF 文件中，并不再執行 RDB 快照持久化。

確保執行完上述命令后，數據庫中的鍵數量沒有發生變化，以確保數據持久化的完整性。

確保寫入命令會被正確地追加到 AOF 文件的末尾，以確保持久化的有效性。

在步驟3中，第一個命令執行后，Redis 會阻塞直到初始 AOF 文件創建完成為止。這意味著 Redis 會等待 AOF 文件完全初始化之后才會繼續處理其他命令請求。一旦 AOF 文件初始化完成，Redis 就會開始將寫入命令追加到 AOF 文件末尾。

第二個命令則是可選的，用于關閉 RDB 功能。如果你愿意的話，你可以同時使用 RDB 和 AOF 這兩種持久化功能。然而，需要確保在 redis.conf 配置文件中打開 AOF 功能，以防止服務器重啟后之前通過 CONFIG SET 命令設置的配置被遺忘，導致服務器按照原來的配置啟動。

十、Redis訂閱發布

Redis 發布訂閱(pub/sub)是一種消息通信模式：發送者(pub)發送消息，訂閱者(sub)接收消息。如：微信、微博、關注系統！

Redis 客戶端可以訂閱任意數量的頻道。

訂閱/發布消息圖：

第一個：消息發送者，第二個：頻道第三個：消息訂閱者！

下圖展示了頻道 channel1 ，以及訂閱這個頻道的三個客戶端 —— client2 、 client5 和 client1 之間的關系：

當有新消息通過 PUBLISH 命令發送給頻道 channel1 時，這個消息就會被發送給訂閱它的三個客戶端：

命令

這些命令被廣泛用于構建即時通信應用，比如網絡聊天室(chatroom)和實時廣播、實時提醒等。

測試

訂閱端：

127.0.0.1:6379> SUBSCRIBE jihu2  #訂閱一個頻道jihu2
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "jihu2"
3) (integer) 1
#等待讀取推送的信息
1) "message"
2) "jihu2"
3) "hello beauty women"  #自動接收jihu2發布的信息
1) "message"
2) "jihu2"
3) "hello redis"

發送端：

[root@localhost bin]# redis-cli -p 6379 #發布者發布消息到頻道
127.0.0.1:6379> PUBLISH jihu2 "hello beauty women"  #發布者發布消息到頻道
(integer) 1
127.0.0.1:6379> PUBLISH jihu2 "hello redis"
(integer) 1

原理

Redis是使用C實現的，通過分析 Redis 源碼里的 pubsub.c 文件，了解發布和訂閱機制的底層實現，籍此加深對 Redis 的理解。

Redis 通過 PUBLISH 、SUBSCRIBE 和 PSUBSCRIBE 等命令實現發布和訂閱功能。

微信：

通過 SUBSCRIBE 命令訂閱某頻道后，redis-server 里維護了一個字典，字典的鍵就是一個個頻道！，而字典的值則是一個鏈表，鏈表中保存了所有訂閱這個 channel 的客戶端。SUBSCRIBE 命令的關鍵，就是將客戶端添加到給定 channel 的訂閱鏈表中。

通過 PUBLISH 命令向訂閱者發送消息，redis-server 會使用給定的頻道作為鍵，在它所維護的 channel 字典中查找記錄了訂閱這個頻道的所有客戶端的鏈表，遍歷這個鏈表，將消息發布給所有訂閱者。

缺點

如果一個客戶端訂閱了頻道，但自己讀取消息的速度卻不夠快的話，那么不斷積壓的消息會使redis輸出緩沖區的體積變得越來越大，這可能使得redis本身的速度變慢，甚至直接崩潰。
這和數據傳輸可靠性有關，如果在訂閱方斷線，那么他將會丟失所有在短線期間發布者發布的消息。

使用場景

1、實時消息系統！

2、事實聊天！（頻道當做聊天室，將信息回顯給所有人即可！）

3、訂閱，關注系統都是可以的！

稍微復雜的場景，我們就會使用消息中間件MQ處理。

當使用 Java 實現 Redis 的發布訂閱功能時，我們需要使用 Redis 官方提供的 Java 客戶端庫 Jedis。確保你已經將 Jedis 添加到項目依賴中。

下面我將編寫兩個 Java 類，一個作為發布者，另一個作為訂閱者。
發布者（Publisher）
import redis.clients.jedis.Jedis;public class Publisher {public static void main(String[] args) {// 連接到本地 Redis 服務器Jedis jedis = new Jedis("localhost");// 模擬發布消息到頻道for (int i = 0; i < 5; i++) {String message = "News Update " + i;jedis.publish("news", message);System.out.println("Published '" + message + "' to channel 'news'");try {Thread.sleep(1000); // 模擬消息發布間隔} catch (InterruptedException e) {e.printStackTrace();}}// 關閉連接jedis.close();}
}
這段代碼連接到本地 Redis 服務器，然后模擬發布了 5 條消息到名為 ‘news’ 的頻道，并且在控制臺上打印了發布的消息內容。
訂閱者（Subscriber）
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;public class Subscriber {public static void main(String[] args) {// 連接到本地 Redis 服務器Jedis jedis = new Jedis("localhost");// 訂閱消息jedis.subscribe(new JedisPubSub() {@Overridepublic void onMessage(String channel, String message) {System.out.println("Received message: " + message);}}, "news");}
}
這段代碼連接到本地 Redis 服務器，然后訂閱了名為 ‘news’ 的頻道，并且實現了 onMessage 方法來處理接收到的消息，在控制臺上打印出來。
運行代碼
確保 Redis 服務器正在運行，然后在不同的終端中分別運行發布者和訂閱者的 Java 類，你會看到訂閱者接收到發布者發布的消息。
通過這種方式，你可以在 Java 中使用 Redis 的發布訂閱功能，實現消息的發布和訂閱。

十一、Redis主從復制

概念

主從復制，是指將一臺Redis服務器的數據，復制到其他的Redis服務器。前者稱為主節點 (master/leader)，后者稱為從節點(slave/follower)；數據的復制是單向的，只能由主節點到從節點。 Master以寫為主，Slave 以讀為主。

默認情況下，每臺Redis服務器都是主節點;

且一個主節點可以有多個從節點(或沒有從節點)，但一個從節點只能有一個主節點。

主從復制的作用主要包括：

1、數據冗余：主從復制實現了數據的熱備份，是持久化之外的一種數據冗余方式。

2、故障恢復：當主節點出現問題時，可以由從節點提供服務，實現快速的故障恢復；實際上是一種服務的冗余。

3、負載均衡：在主從復制的基礎上，配合讀寫分離，可以由主節點提供寫服務，由從節點提供讀服務（即寫Redis數據時應用連接主節點，讀Redis數據時應用連接從節點），分擔服務器負載；尤其是在寫少讀多的場景下，通過多個從節點分擔讀負載，可以大大提高Redis服務器的并發量。

4、高可用（集群）基石：除了上述作用以外，主從復制還是哨兵和集群能夠實施的基礎，因此說主從復制是Redis高可用的基礎

一般來說，要將Redis運用于工程項目中，只使用一臺Redis是萬萬不能的（宕機），原因如下：

1、從結構上，單個Redis服務器會發生單點故障，并且一臺服務器需要處理所有的請求負載，壓力較大；

2、從容量上，單個Redis服務器內存容量有限，就算一臺Redis服務器內存容量為256G，也不能將所有內存用作Redis存儲內存，一般來說，單臺Redis最大使用內存不應該超過20G。

電商網站上的商品，一般都是一次上傳，無數次瀏覽的，說專業點也就是"多讀少寫"。

對于這種場景，我們可以使如下這種架構：

主從復制，讀寫分離！ 80% 的情況下都是在進行讀操作！減緩服務器的壓力！架構中經常使用！一主二從！

只要在公司中，主從復制就是必須要使用的，因為在真實的項目中不可能單機使用Redis！

環境配置

只配置從庫，不用配置主庫！

127.0.0.1:6379> info replication  #查看當前庫的信息
# Replication
role:master   #角色  master
connected_slaves:0  # 0 表示沒有從機
master_failover_state:no-failover
master_replid:f80a04ead679af98a1c84518b78e529a0a33c986
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:0
second_repl_offset:-1
repl_backlog_active:0
repl_backlog_size:1048576
repl_backlog_first_byte_offset:0
repl_backlog_histlen:0

復制3個配置文件，然后修改對應的信息

要想改一個啟動項

要改以下四步： ( redis80.confshu ）

端口 2.pid名字 3.log文件名字 4.dump.rdb 名字

修改完畢后，啟動我們的3個redis服務器，可以通過進程信息查看

一主二從

默認情況下，每臺Redis服務器都是主節點; 我們一般情況下只用配置從機就好了。

主（79）二從( 80 , 81 )

從機中查看：

127.0.0.1:6380> SLAVEOF 127.0.0.1 6379   #SLAVEOF host 6379  找誰當自己的老的 
OK
127.0.0.1:6380> info repilcation
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:3
master_sync_in_progress:0
slave_read_repl_offset:28
slave_repl_offset:28
slave_priority:100
slave_read_only:1
replica_announced:1
connected_slaves:0
master_failover_state:no-failover
master_replid:a0e8a30728e95d67d6b191795452ea27b934ba67
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:28
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:1
repl_backlog_histlen:28

主機中查看(此時兩個從機都配置完畢了)：

127.0.0.1:6379> info replication
# Replication
role:master
connected_slaves:2   #多了從機的配置
slave0:ip=127.0.0.1,port=6380,state=online,offset=196,lag=0  #多了從機的配置
slave1:ip=127.0.0.1,port=6381,state=online,offset=196,lag=1
master_failover_state:no-failover
master_replid:a0e8a30728e95d67d6b191795452ea27b934ba67
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:196
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:1
repl_backlog_histlen:196

真實的主從配置應該在配置文件中配置，這樣的話是永久的，我們這里使用的是命令是暫時的。

這里是在配置文件中配置主從信息

細節

主機可以寫，從機不能寫只能讀。主機中的所有信息和數據，都會自動被從機保存。

主機寫：

從機只能讀取內容

測試：主機斷開連接，從機依舊連接到主機的，但是沒有寫操作，這個時候，主機如果回來了，從機依舊可以直接獲取到主機寫的信息！

如果是使用命令行，來配置的主從，這個時候如果重啟了，就會變回主機！只要變為從機，立馬就會從主機中獲取值！

賦值原理

Slave 啟動成功連接到 master 后會發送一個sync同步命令

Master 接到命令，啟動后臺的存盤進程，同時收集所有接收到的用于修改數據集命令，在后臺進程執行完畢之后，master將傳送整個數據文件到slave，并完成一次完全同步。

全量復制：而slave服務在接收到數據庫文件數據后，將其存盤并加載到內存中。

增量復制：Master 繼續將新的所有收集到的修改命令依次傳給slave，完成同步

但是只要是重新連接master，一次完全同步（全量復制）將被自動執行！我們的數據一定可以在從機中看到！

層層鏈路

上一個M鏈接下一個 S！

這時候也可以完成我們的主從復制！

如果沒有老大了，這個時候能不能選擇一個老大出來呢？需手動設置！

謀朝篡位

如果主機斷開了連接，我們可以使用SLAVEOF no one 讓自己變成主機！其他的節點就可以手動連接到最新的這個主節點（手動）！如果這個時候老大修復了，那就重新連接！

十二、哨兵模式

（自動選舉老大的模式）

概述

主從切換技術的方法是：當主服務器宕機后，需要手動把一臺從服務器切換為主服務器，這就需要人工干預，費事費力，還會造成一段時間內服務不可用。這不是一種推薦的方式，更多時候，我們優先考慮哨兵模式。Redis從2.8開始正式提供了Sentinel（哨兵）架構來解決這個問題。

謀朝篡位的自動版，能夠后臺監控主機是否故障，如果故障了根據投票數自動將從庫轉換為主庫。

哨兵模式是一種特殊的模式，首先Redis提供了哨兵的命令，哨兵是一個獨立的進程，作為進程，它會獨立運行。其原理是哨兵通過發送命令，等待Redis服務器響應，從而監控運行的多個Redis實例。

Redis中的哨兵模式是一種用于實現高可用性（high availability）的機制。在Redis中，哨兵（Sentinel）是一個獨立的進程，負責監控主數據庫（master）和從數據庫（slave），并在主數據庫失效時自動將一個從數據庫提升為新的主數據庫，以確保系統的持續可用性。

以下是Redis中哨兵模式的主要組成部分和工作原理：

哨兵（Sentinel）：哨兵是一個獨立的進程，負責監控Redis中的主數據庫和從數據庫。哨兵會定期向主數據庫和從數據庫發送PING命令來檢測它們的狀態，并在發現主數據庫不可用時，選舉出一個新的主數據庫。

主數據庫（Master）：主數據庫是Redis中的主要數據存儲節點，負責處理所有寫操作，并將數據同步到從數據庫。

從數據庫（Slave）：從數據庫是主數據庫的備份節點，負責接收來自主數據庫的數據復制，并在主數據庫失效時，通過選舉成為新的主數據庫。

工作流程如下：

初始階段，哨兵監控著系統中的主數據庫和從數據庫。
如果主數據庫失效（比如宕機），哨兵會開始進行故障檢測。
哨兵會根據預設的條件（例如，主數據庫不可用的連續次數）來決定是否宣布主數據庫失效。
一旦主數據庫失效，哨兵會從當前的從數據庫中選舉出一個新的主數據庫。
選舉出的新主數據庫將會向哨兵報告自己的狀態，并成為系統中的主數據庫。
哨兵會更新系統中所有的從數據庫，使它們成為新主數據庫的從數據庫，并開始將數據復制到它們。

在Redis Sentinel（哨兵）中，用于選舉主節點的算法主要是基于哨兵之間的通信和協作。哨兵通過相互通信來達成一致，選擇一個合適的從節點作為新的主節點。主要的選舉算法包括：

投票選舉算法：

在Redis Sentinel中，當哨兵檢測到主節點不可用時，它會發起一次選舉過程。
每個哨兵都會投票給自己認為適合成為新主節點的從節點。
哨兵將投票結果進行匯總，并選擇得到最多票數的從節點作為新的主節點。
如果存在多個從節點獲得相同數量的選票，則根據一定的優先級規則進行進一步的決定，例如根據從節點的復制偏移量（復制進度）或者ID等因素進行排序。

加權投票選舉算法：

類似于投票選舉算法，但是不同哨兵的投票可能具有不同的權重，這可以根據哨兵的配置或性能來調整。
通過給具有更高權重的哨兵賦予更多的投票權，可以在選舉過程中更加重視它們的投票結果。

基于節點健康度的選舉算法：

在進行選舉時，哨兵可能會考慮到每個從節點的健康度。
如果一個從節點被認為更健康或者復制進度更接近于主節點，則它可能會獲得更多的選票，從而有更大的機會成為新的主節點。

這里的哨兵有兩個作用

通過發送命令，讓Redis服務器返回監控其運行狀態，包括主服務器和從服務器。
當哨兵監測到master宕機，會自動將slave切換成master，然后通過發布訂閱模式通知其他的從服務器，修改配置文件，讓它們切換主機。

然而一個哨兵進程對Redis服務器進行監控，可能會出現問題，為此，我們可以使用多個哨兵進行監控。各個哨兵之間還會進行監控，這樣就形成了多哨兵模式。

假設主服務器宕機，哨兵1先檢測到這個結果，系統并不會馬上進行failover過程，僅僅是哨兵1主觀的認為主服務器不可用，這個現象成為主觀下線。當后面的哨兵也檢測到主服務器不可用，并且數量達到一定值時，那么哨兵之間就會進行一次投票，投票的結果由一個哨兵發起，進行failover[故障轉移]操作。切換成功后，就會通過發布訂閱模式，讓各個哨兵把自己監控的從服務器實現切換主機，這個過程稱為 客觀下線。

測試

我們目前的狀態是一主二從！

1、配置哨兵配置文件 sentinel.conf

# sentinel monitor 被監控的名稱 host port 1
sentinel monitor myredis 127.0.0.1 6379 1   
#后面的這個數字1 ，代表主機掛了，從機投票看讓誰接替成為主機，票數最多的，就會成為主機。

sentinel：這是Redis Sentinel的命令行工具。

monitor：這是sentinel命令的一個子命令，用于指示哨兵開始監視一個Redis實例。

myredis：這是要監視的Redis實例的名稱。它是一個用戶定義的標識符，用于在監視多個Redis實例時進行區分。

127.0.0.1：這是要監視的Redis實例的主機地址。在這里，使用的是本地主機（localhost）的IP地址。

6379：這是要監視的Redis實例的端口號。在這里，Redis實例使用的是默認的6379端口。

1：這是指定了要求多少個哨兵同意（包括自身）才將一個節點標記為主節點。在這里，使用的是"1"，表示只需要一個哨兵同意即可。

綜合起來，該命令告訴Redis Sentinel開始監視一個名為 “myredis” 的Redis實例，該實例位于本地主機的6379端口，如果至少有一個哨兵同意，它將標記一個節點為主節點。

2.啟動哨兵

[root@localhost bin]# redis-sentinel kconfig/sentinel.conf

如果Master 節點斷開了，這個時候就會從從機中隨機選擇一個服務器！（這里面有一個投票算法！）

哨兵日志！ (指定6381為主節點)

哨兵模式

如果主機此時回來了，只能歸并到新的主機下，當做從機，這就是哨兵模式的規則！

哨兵模式優缺點

優點：

哨兵集群，基于主從復制模式，所有主從復制的優點，它都有
主從可以切換，故障可以轉移，系統的可用性更好
哨兵模式是主從模式的升級，手動到自動，更加健壯

缺點：

Redis不好在線擴容，集群容量一旦達到上限，在線擴容就十分麻煩
實現哨兵模式的配置其實是很麻煩的，里面有很多配置項

哨兵模式的全部配置

完整的哨兵模式配置文件 sentinel.conf

# Example sentinel.conf# 哨兵sentinel實例運行的端口 默認26379
port 26379# 哨兵sentinel的工作目錄
dir /tmp# 哨兵sentinel監控的redis主節點的 ip port 
# master-name  可以自己命名的主節點名字 只能由字母A-z、數字0-9 、這三個字符".-_"組成。
# quorum 當這些quorum個數sentinel哨兵認為master主節點失聯 那么這時 客觀上認為主節點失聯了
# sentinel monitor <master-name> <ip> <redis-port> <quorum>
sentinel monitor mymaster 127.0.0.1 6379 1# 當在Redis實例中開啟了requirepass foobared 授權密碼 這樣所有連接Redis實例的客戶端都要提供密碼
# 設置哨兵sentinel 連接主從的密碼 注意必須為主從設置一樣的驗證密碼
# sentinel auth-pass <master-name> <password>
sentinel auth-pass mymaster MySUPER--secret-0123passw0rd# 指定多少毫秒之后 主節點沒有應答哨兵sentinel 此時 哨兵主觀上認為主節點下線 默認30秒
# sentinel down-after-milliseconds <master-name> <milliseconds>
sentinel down-after-milliseconds mymaster 30000# 這個配置項指定了在發生failover主備切換時最多可以有多少個slave同時對新的master進行 同步，
這個數字越小，完成failover所需的時間就越長，
但是如果這個數字越大，就意味著越 多的slave因為replication而不可用。
可以通過將這個值設為 1 來保證每次只有一個slave 處于不能處理命令請求的狀態。
# sentinel parallel-syncs <master-name> <numslaves>
sentinel parallel-syncs mymaster 1# 故障轉移的超時時間 failover-timeout 可以用在以下這些方面： 
#1. 同一個sentinel對同一個master兩次failover之間的間隔時間。
#2. 當一個slave從一個錯誤的master那里同步數據開始計算時間。直到slave被糾正為向正確的master那里同步數據時。
#3.當想要取消一個正在進行的failover所需要的時間。  
#4.當進行failover時，配置所有slaves指向新的master所需的最大時間。不過，即使過了這個超時，slaves依然會被正確配置為指向master，但是就不按parallel-syncs所配置的規則來了
# 默認三分鐘
# sentinel failover-timeout <master-name> <milliseconds>
sentinel failover-timeout mymaster 180000# SCRIPTS EXECUTION#配置當某一事件發生時所需要執行的腳本，可以通過腳本來通知管理員，例如當系統運行不正常時發郵件通知相關人員。
#對于腳本的運行結果有以下規則：
#若腳本執行后返回1，那么該腳本稍后將會被再次執行，重復次數目前默認為10
#若腳本執行后返回2，或者比2更高的一個返回值，腳本將不會重復執行。
#如果腳本在執行過程中由于收到系統中斷信號被終止了，則同返回值為1時的行為相同。
#一個腳本的最大執行時間為60s，如果超過這個時間，腳本將會被一個SIGKILL信號終止，之后重新執行。#通知型腳本:當sentinel有任何警告級別的事件發生時（比如說redis實例的主觀失效和客觀失效等等），將會去調用這個腳本，
#這時這個腳本應該通過郵件，SMS等方式去通知系統管理員關于系統不正常運行的信息。調用該腳本時，將傳給腳本兩個參數，
#一個是事件的類型，
#一個是事件的描述。
#如果sentinel.conf配置文件中配置了這個腳本路徑，那么必須保證這個腳本存在于這個路徑，并且是可執行的，否則sentinel無法正常啟動成功。
#通知腳本  用shell編寫
# sentinel notification-script <master-name> <script-path>sentinel notification-script mymaster /var/redis/notify.sh# 客戶端重新配置主節點參數腳本
# 當一個master由于failover而發生改變時，這個腳本將會被調用，通知相關的客戶端關于master地址已經發生改變的信息。
# 以下參數將會在調用腳本時傳給腳本:
# <master-name> <role> <state> <from-ip> <from-port> <to-ip> <to-port>
# 目前<state>總是“failover”,
# <role>是“leader”或者“observer”中的一個。 
# 參數 from-ip, from-port, to-ip, to-port是用來和舊的master和新的master(即舊的slave)通信的
# 這個腳本應該是通用的，能被多次調用，不是針對性的。
# sentinel client-reconfig-script <master-name> <script-path>
sentinel client-reconfig-script mymaster /var/redis/reconfig.sh  #一般都是由運維來配置。

sentinel parallel-syncs mymaster 1:

這條Redis Sentinel配置命令指定了在執行故障轉移期間，要將從節點重新配置為新的主節點所需的并行復制過程的最大數量。這里的參數解釋如下：

sentinel：Redis Sentinel命令前綴。
parallel-syncs：這是配置項的名稱，用于指定并行同步的數量。
mymaster：這是要設置并行同步的Redis主節點的名稱。
1：這是指定了在執行故障轉移期間所允許的最大并行同步數量。

換句話說，當發生故障轉移并將一個從節點提升為新的主節點時，這個參數指定了可以同時對多少個從節點執行同步操作。在這種情況下，配置為 1 表示在進行故障轉移時只允許一個從節點與新的主節點進行并行同步。這可以避免對主節點的負載過大，因為在同步期間會消耗大量的網絡帶寬和主節點的處理能力。

如果需要更高的可用性，可以增加并行同步的數量，但要注意確保主節點的資源足夠支持這些并行同步操作，以免影響主節點的性能。

十三、Redis緩存穿透和雪崩

緩存就是一堵墻

穿透就是越過墻（查不到），

擊穿就是墻破了個洞（熱點過期，點），

雪崩就是墻倒了（緩存集中過期 / Redis宕機，面）

服務的高可用問題！

Redis緩存的使用，極大的提升了應用程序的性能和效率，特別是數據查詢方面。但同時，它也帶來了一些問題。其中，最要害的問題，就是數據的一致性問題，從嚴格意義上講，這個問題無解。如果對數據的一致性要求很高，那么就不能使用緩存。

另外的一些典型問題就是，緩存穿透、緩存雪崩和緩存擊穿。目前，業界也都有比較流行的解決方案。

緩存穿透（查不到）

概念

緩存穿透的概念很簡單，用戶想要查詢一個數據，發現redis內存數據庫沒有，也就是緩存沒有命中，于是向持久層數據庫查詢。發現也沒有，于是本次查詢失敗。當用戶很多的時候，緩存都沒有命中（秒殺！），于是都去請求了持久層數據庫。這會給持久層數據庫造成很大的壓力，這時候就相當于出現了緩存穿透。

解決方案

布隆過濾器

布隆過濾器是一種數據結構，對所有可能查詢的參數以hash形式存儲，在控制層先進行校驗，不符合則丟棄，從而避免了對底層存儲系統的查詢壓力；

緩存空對象

當存儲層不命中后，即使返回的空對象也將其緩存起來，同時會設置一個過期時間，之后再訪問這個數據將會從緩存中獲取，保護了后端數據源；

但是這種方法會存在兩個問題：

1、如果空值能夠被緩存起來，這就意味著緩存需要更多的空間存儲更多的鍵，因為這當中可能會有很多的空值的鍵；

2、即使對空值設置了過期時間，還是會存在緩存層和存儲層的數據會有一段時間窗口的不一致，這對于需要保持一致性的業務會有影響。

緩存擊穿（量太大，緩存過期！）

概述

這里需要注意和緩存擊穿的區別，緩存擊穿，是指一個key非常熱點，在不停的扛著大并發，大并發集中對這一個點進行訪問，當這個key在失效的瞬間，持續的大并發就穿破緩存，直接請求數據庫，就像在一個屏障上鑿開了一個洞。

當某個key在過期的瞬間，有大量的請求并發訪問，這類數據一般是熱點數據，由于緩存過期，會同時訪問數據庫來查詢最新數據，并且回寫緩存，會導使數據庫瞬間壓力過大。

解決方案

設置熱點數據永不過期

從緩存層面來看，沒有設置過期時間，所以不會出現熱點 key 過期后產生的問題。

加互斥鎖 (分布式鎖)

分布式鎖：使用分布式鎖，保證對于每個key同時只有一個線程去查詢后端服務，其他線程沒有獲得分布式鎖的權限，因此只需要等待即可。這種方式將高并發的壓力轉移到了分布式鎖，因此對分布式鎖的考驗很大。

緩存雪崩

概念

緩存雪崩，是指在某一個時間段，緩存集中過期失效。Redis 宕機！

產生雪崩的原因之一，比如在寫本文的時候，馬上就要到雙十二零點，很快就會迎來一波搶購，這波商品時間比較集中的放入了緩存，假設緩存一個小時。那么到了凌晨一點鐘的時候，這批商品的緩存就都過期了。而對這批商品的訪問查詢，都落到了數據庫上，對于數據庫而言，就會產生周期性的壓力波峰。于是所有的請求都會達到存儲層，存儲層的調用量會暴增，造成存儲層也會掛掉的情況。

其實集中過期，倒不是非常致命，比較致命的緩存雪崩，是緩存服務器某個節點宕機或斷網。因為自然形成的緩存雪崩，一定是在某個時間段集中創建緩存，這個時候，數據庫也是可以頂住壓力的。無非就是對數據庫產生周期性的壓力而已。而緩存服務節點的宕機，對數據庫服務器造成的壓力是不可預知的，很有可能瞬間就把數據庫壓垮。

解決方案

redis高可用

這個思想的含義是，既然redis有可能掛掉，那我多增設幾臺redis，這樣一臺掛掉之后其他的還可以繼續工作，其實就是搭建的集群。（異地多活！）

限流降級（在SpringCloud講解過！）

這個解決方案的思想是，在緩存失效后，通過加鎖或者隊列來控制讀數據庫寫緩存的線程數量。比如對某個key只允許一個線程查詢數據和寫緩存，其他線程等待。

數據預熱

數據加熱的含義就是在正式部署之前，我先把可能的數據先預先訪問一遍，這樣部分可能大量訪問的數據就會加載到緩存中。在即將發生大并發訪問前手動觸發加載緩存不同的key，設置不同的過期時間，讓緩存失效的時間點盡量均勻。

十四、布隆過濾器

1. 什么是布隆過濾器

布隆過濾器（Bloom Filter），是1970年，由一個叫布隆的小伙子提出的，距今已經五十年了，和老哥一樣老。

它實際上是一個很長的二進制向量和一系列隨機映射函數，二進制大家應該都清楚，存儲的數據不是0就是1，默認是0。

主要用于判斷一個元素是否在一個集合中，0代表不存在某個數據，1代表存在某個數據。

懂了嗎？作為暖男的老哥在給你們畫張圖來幫助理解：

2. 布隆過濾器用途

解決Redis緩存穿透（今天重點講解）
在爬蟲時，對爬蟲網址進行過濾，已經存在布隆中的網址，不在爬取。
垃圾郵件過濾，對每一個發送郵件的地址進行判斷是否在布隆的黑名單中，如果在就判斷為垃圾郵件。

3. 布隆過濾器原理

存入過程

布隆過濾器上面說了，就是一個二進制數據的集合。當一個數據加入這個集合時，經歷如下洗禮（這里有缺點，下面會講）：

通過K個哈希函數計算該數據，返回K個計算出的hash值
這些K個hash值映射到對應的K個二進制的數組下標
將K個下標對應的二進制數據改成1。

例如，第一個哈希函數返回x，第二個第三個哈希函數返回y與z，那么：X、Y、Z對應的二進制改成1。

查詢過程

布隆過濾器主要作用就是查詢一個數據，在不在這個二進制的集合中，查詢過程如下

通過K個哈希函數計算該數據，對應計算出的K個hash值
通過hash值找到對應的二進制的數組下標
判斷：如果存在一處位置的二進制數據是0，那么該數據不存在。如果都是1，該數據存在集合中。（這里有缺點，下面會講）

刪除過程

一般不能刪除布隆過濾器里的數據，這是一個缺點之一，我們下面會分析

布隆過濾器的優缺點

優點

由于存儲的是二進制數據，所以占用的空間很小
它的插入和查詢速度是非常快的，時間復雜度是O（K），可以聯想一下HashMap的過程
保密性很好，因為本身不存儲任何原始數據，只有二進制數據

缺點

這就要回到我們上面所說的那些缺點了。

添加數據是通過計算數據的hash值，那么很有可能存在這種情況：兩個不同的數據計算得到相同的hash值。

例如圖中的“你好”和“hello”，假如最終算出hash值相同，那么他們會將同一個下標的二進制數據改為1。

這個時候，你就不知道下標為2的二進制，到底是代表“你好”還是“hello”。

由此得出如下缺點

一、存在誤判

假如上面的圖沒有存"hello"，只存了"你好"，那么用"hello"來查詢的時候，會判斷"hello"存在集合中。

因為“你好”和“hello”的hash值是相同的，通過相同的hash值，找到的二進制數據也是一樣的，都是1。

二、刪除困難

到這里我不說大家應該也明白為什么吧，作為你們的暖男老哥，還是講一下吧。

還是用上面的舉例，因為“你好”和“hello”的hash值相同，對應的數組下標也是一樣的。

這時候老哥想去刪除“你好”，將下標為2里的二進制數據，由1改成了0。

那么我們是不是連“hello”都一起刪了呀。（0代表有這個數據，1代表沒有這個數據）

4. 實現布隆過濾器

有很多種實現方式，其中一種就是Guava提供的實現方式。

一、引入Guava pom配置

<dependency><groupId>com.google.guava</groupId><artifactId>guava</artifactId><version>29.0-jre</version>
</dependency>

二、代碼實現

這里我們順便測一下它的誤判率。

package com.fatfish.bloomfilter;import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;public class MyBloomFilter {private static int size = 1000000;private static double fpp = 0.01;private static BloomFilter<Integer> bloomFilter = BloomFilter.create(Funnels.integerFunnel(), size, fpp);public static void main(String[] args) {for (int i = 0; i < size; i++) {bloomFilter.put(i);}int count = 0;for (int i = size; i < size + 1000000; i++) {if (bloomFilter.mightContain(i)) {count++;System.out.println(i + "誤判了");}}System.out.println("總共的誤判數目：" + count);System.out.printf("誤判率：%.3f", 1.0*count/size);}
}

運行結果：

10萬數據里有947個誤判，約等于0.01%，也就是我們代碼里設置的誤判率：fpp = 0.01。

深入分析代碼

核心BloomFilter.create方法

@VisibleForTestingstatic <T> BloomFilter<T> create(      Funnel<? super T> funnel, long expectedInsertions, double fpp, Strategy strategy) {。。。。
}

這里有四個參數：

funnel：數據類型(一般是調用Funnels工具類中的)
expectedInsertions：期望插入的值的個數
fpp：誤判率(默認值為0.03)
strategy：哈希算法

我們重點講一下fpp參數

fpp誤判率（False Positive Probability，假陽性概率）

情景一：`fpp = 0.01`

誤判個數：947
占內存大小：9585058位數

情景二：`fpp = 0.03`（默認參數）

誤判個數：3033
占內存大小：7298440位數

情景總結

誤判率可以通過fpp參數進行調節
fpp越小，需要的內存空間就越大：0.01需要900多萬位數，0.03需要700多萬位數。
fpp越小，集合添加數據時，就需要更多的hash函數運算更多的hash值，去存儲到對應的數組下標里。（忘了去看上面的布隆過濾存入數據的過程）

上面的numBits，表示存一百萬個int類型數字，需要的位數為7298440，700多萬位。理論上存一百萬個數，一個int是4字節32位，需要481000000=3200萬位。如果使用HashMap去存，按HashMap50%的存儲效率，需要6400萬位。可以看出BloomFilter的存儲空間很小，只有HashMap的1/10左右

上面的numHashFunctions表示需要幾個hash函數運算，去映射不同的下標存這些數字是否存在（0 or 1）。

解決Redis緩存穿透

上面使用Guava實現的布隆過濾器是把數據放在了本地內存中。分布式的場景中就不合適了，無法共享內存。

我們還可以用Redis來實現布隆過濾器，這里使用Redis封裝好的客戶端工具Redisson。

其底層是使用數據結構bitMap，大家就把它理解成上面說的二進制結構，由于篇幅原因，bitmap不在這篇文章里講，我們之后寫一篇文章介紹。

代碼實現

pom配置：

<dependency><groupId>org.redisson</groupId><artifactId>redisson-spring-boot-starter</artifactId><version>3.13.4</version>
</dependency>

java代碼：

public class RedissonBloomFilter {public static void main(String[] args) {Config config = new Config();config.useSingleServer().setAddress("redis://127.0.0.1:6379");config.useSingleServer().setPassword("1234");//構造RedissonRedissonClient redisson = Redisson.create(config);RBloomFilter<String> bloomFilter = redisson.getBloomFilter("phoneList");//初始化布隆過濾器：預計元素為100000000L,誤差率為3%bloomFilter.tryInit(100000000L,0.03);//將號碼10086插入到布隆過濾器中bloomFilter.add("10086");//判斷下面號碼是否在布隆過濾器中//輸出falseSystem.out.println(bloomFilter.contains("123456"));//輸出trueSystem.out.println(bloomFilter.contains("10086"));}
}

由于Guava那個版本，我們已經很詳細的講了布隆過濾器的那些參數，這里就不重復贅述了。

Jedis：

Jedis是Redis官方推薦的Java客戶端之一，它是一個直接連接Redis服務器的Java庫，提供了簡單易用的API，使用起來較為直觀。
Jedis通過創建連接池來管理與Redis服務器的連接，通過連接池可以減少連接創建和銷毀的開銷，提高性能。
Jedis的優點是簡單易用，適合于對Redis操作較為簡單的場景。

RedisTemplate：

RedisTemplate是Spring Data Redis提供的一個Redis操作模板，它封裝了Redis連接和操作的細節，提供了更高層次的抽象和功能。
RedisTemplate提供了一種更加靈活和功能豐富的方式來操作Redis，支持多種數據結構和命令，同時與Spring框架集成更為緊密。
RedisTemplate是基于Jedis或Lettuce等底層連接池實現的，可以靈活切換底層連接方式。

Redis連接池：

Redis連接池用于管理與Redis服務器的連接，它可以提高連接的復用率和效率，減少連接創建和銷毀的開銷。
連接池可以通過預先創建一定數量的連接并維護在一個連接池中，客戶端需要時從連接池中獲取連接并在使用完畢后歸還，以實現連接的復用和管理。

Lettuce：

Lettuce是一個基于Netty實現的高性能Redis客戶端，相比于Jedis，它支持異步操作和響應式編程模型，具有更好的并發性能和可擴展性。
Lettuce提供了更加靈活和強大的API，支持連接池、集群、哨兵、SSL等特性，同時提供了更豐富的配置選項和監控功能。

Redisson：

Redisson是一個基于Redis的Java駐內存數據網格（In-Memory Data Grid）和分布式鎖（Distributed Lock）的框架，它提供了豐富的分布式數據結構和分布式服務。
Redisson不僅提供了對Redis的普通操作，還提供了諸如分布式集合、分布式映射、分布式鎖等高級功能，可以簡化分布式系統的開發和管理。
Redisson是一個完整的Redis客戶端，它使用了Lettuce作為底層的Redis連接實現，同時提供了更高級別的功能和抽象。

總的來說，Jedis和Lettuce是直接連接Redis服務器的客戶端，RedisTemplate是Spring Data Redis提供的操作模板，Redisson是一個完整的Redis客戶端框架，而Redis連接池則用于管理客戶端與Redis服務器之間的連接。它們之間的選擇取決于項目的需求和開發者的偏好，可以根據具體場景選擇最適合的客戶端。

package com.fatfish.bloomfilter;import com.fatfish.config.RedisConfig;
import com.fatfish.utils.JedisEnumSingleton;
import com.fatfish.utils.RedisUtil;
import com.google.common.base.Charsets;
import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;@Component
public class RedisBloomFilter {// 布隆過濾器零一串大小private static final long EXPECTED_INSERTIONS = 1000000;// 預設的誤判率，假陽性概率private static final double FALSE_POSITIVE_PROBABILITY = 0.03;// 主機號private static final String HOST = "127.0.0.1";// 端口號private static final int PORT = 6379;// 布隆過濾器private static BloomFilter<CharSequence> bloomFilter;// Jedisprivate static Jedis jedis;// Jedis Poolprivate static JedisPool jedisPool;// RedisTemplate —— 封裝在 RedisUtil 中private static RedisUtil redisUtil;// redissonprivate static RedissonClient redissonClient;// 內存private static Map<String, String> memory;static {// 初始化內存memory = new HashMap<>();memory.put("張三", "北京市");memory.put("李四", "上海市");memory.put("王五", "廣州市");// 初始化布隆過濾器bloomFilter = BloomFilter.create(Funnels.stringFunnel(Charsets.UTF_8), 1, 0.01);// 初始化 Jedis PooljedisPool = new JedisPool(new JedisPoolConfig(), HOST, PORT);// redisUtil 初始化redisUtil = new RedisUtil();// redissonClient 初始化Config config = new Config();config.useSingleServer().setAddress("redis://" + HOST + ":" + PORT);redissonClient = Redisson.create(config);//        bloomFilter = redissonClient.getBloomFilter("myBloomFilter");// 初始化 Jedis// 方式 1 ：Jedis直連jedis = JedisEnumSingleton.INSTANCE.getJedis();// 方式 2 ：Jedis pooljedis = jedisPool.getResource();// 方式 3 ：RedisTemplate —— 基于Jedis或Lettuce等底層連接池實現// 方式 4 ：Redisson}public static void main(String[] args) {// 加載數據庫中的相關數據到布隆過濾器loadDataFromDB2BloomFilter();// 查布隆過濾器，判斷數據是否存在（Redis 或 內存）// 若數據不存在，則返回// 若數據存在，則繼續；String[] querys = new String[]{"張三", "李四", "張三", "王五", "趙六"};for (String query : querys) {System.out.println("=================================");System.out.println("開始查詢" + query);boolean mightContain = bloomFilter.mightContain(query);// 數據不存在，則返回if (!mightContain) {System.out.println(query + "不存在，請檢查后再操作！");return;}// 數據存在，則繼續else {// 查Redis 緩存
//                queryRedisByJedis(query);
//                queryRedisByRedisTemplate(query);queryRedisByRedisson(query);}}if (redissonClient != null && !redissonClient.isShutdown()) {redissonClient.shutdown();}}/*** 查Redis 緩存* Redis 緩存存在，則直接返回* Redis 緩存不存在，則查內存，并將結果也寫到Redis 緩存中** @param query 待查詢的 key*/private static void queryRedisByRedisson(String query) {boolean exists = redissonClient.getBucket(query).isExists();// Redis 緩存存在，則直接返回if (exists) {String result = redissonClient.getBucket(query).get().toString();System.out.println("結果存在于Redis緩存, 查詢結果：" + result);}// Redis 緩存不存在，則查內存，并將結果也寫到Redis 緩存中else {// 查內存if (memory.containsKey(query)) {String result = memory.get(query);// 將結果也寫到Redis 緩存中redissonClient.getBucket(query).set(result);System.out.println("結果存在于DB, 查詢結果：" + result);}}}/*** 查Redis 緩存* Redis 緩存存在，則直接返回* Redis 緩存不存在，則查內存，并將結果也寫到Redis 緩存中** @param query 待查詢的 key*/private static void queryRedisByRedisTemplate(String query) {boolean exists = redisUtil.hasKey(query);// Redis 緩存存在，則直接返回if (exists) {String result = (String) redisUtil.get(query);System.out.println("結果存在于Redis緩存, 查詢結果：" + result);}// Redis 緩存不存在，則查內存，并將結果也寫到Redis 緩存中else {// 查內存if (memory.containsKey(query)) {String result = memory.get(query);// 將結果也寫到Redis 緩存中redisUtil.set(query, result);System.out.println("結果存在于DB, 查詢結果：" + result);}}}/*** 查Redis 緩存* Redis 緩存存在，則直接返回* Redis 緩存不存在，則查內存，并將結果也寫到Redis 緩存中** @param query 待查詢的 key*/private static void queryRedisByJedis(String query) {Boolean exists = jedis.exists(query);// Redis 緩存存在，則直接返回if (exists) {String result = jedis.get(query);System.out.println("結果存在于Redis緩存, 查詢結果：" + result);}// Redis 緩存不存在，則查內存，并將結果也寫到Redis 緩存中else {// 查內存if (memory.containsKey(query)) {String result = memory.get(query);// 將結果也寫到Redis 緩存中jedis.set(query, result);System.out.println("結果存在于DB, 查詢結果：" + result);}}}/*** 加載數據庫中的相關數據到布隆過濾器*/private static void loadDataFromDB2BloomFilter() {System.out.println("開始加載數據庫中的相關數據到布隆過濾器...");for (Map.Entry<String, String> cache : memory.entrySet()) {String name = cache.getKey();bloomFilter.put(name);}System.out.println("數據加載完成！");}
}