[CF418E] Tricky Password 題解
阿新 • • 發佈：2022-05-21
Redis
1、NoSql概述
1.1、為什麼要用Nosql
單機MySQL時代
90年代，一個基本的網站訪問量一般不會太大，單個數據庫完全足夠！
那個時候，更多的去使用靜態網頁 Html~ 伺服器根本沒有太大的壓力
這種情況下：整個網站的瓶頸是什麼？
- 資料量如果太大、一臺機器放不下了！
- 資料的索引 （B+Tree）,一個機器記憶體也放不下
- 訪問量（讀寫混合），一個伺服器承受不了
只要開始出現以上三種情況之一，那麼就必須要晉級！
Memcached(快取)+MySQL+垂直拆分(讀寫分離)
網站80%的情況都是在讀，每次都要去查詢資料庫的話就十分麻煩！所以說我們要減輕資料庫的壓力，可以使用快取來保證效率！
發展過程：優化資料結構和索引 --> 檔案快取（IO）--> Memcached
分庫分表+水平拆分+MySQL叢集
技術和業務在發展的同時，對人的要求也越來越高！
本質：資料庫：(讀 寫)
MyISAM:表鎖 十分影響效率！高併發下就會出現嚴重的鎖問題
InnoDB：行鎖
慢慢的就開始使用分庫分表來解決寫的壓力！Mysql在那個年代推出了表分割槽！這個並沒有多少公司使用！
Mysql叢集，很好的滿足那個年代的需求
如今最近的年代
MySQL等關係型資料庫就不夠用了！資料量很多，變化很快！
MySQL有的使用它來儲存一些較大的檔案，部落格，圖片！資料庫表很大，效率就低了！如果有一種資料庫來專門處理這種資料，MySQL壓力就會變得十分小（研究如何處理這些問題）大資料的IO壓力下，表的結構幾乎是無法更改的
目前一個基本的網際網路專案
為什麼要用NoSQL
使用者的個人資訊，社交網路，地理位置，使用者自己產生的資料，使用者日誌等等爆發式增長！
這個時候我們就需要使用NoSQL資料庫，NoSQL可以很好的處理以上情況！
1.2、什麼是NoSQL
NoSQL = Not Only SQL (不僅僅是SQL)
關係型資料庫：表格：行，列 （POI）
泛指非關係型資料庫，隨著web2.0網際網路的誕生！傳統的關係型資料庫很難對付web2.0時代！尤其是超大規模的高併發的社群！暴露出來很多難以克服的問題，NoSQL在當今大資料環境下發展的十分迅速，Redis是發展最快的，而且是我們當下必須要掌握的。
很多的資料型別使用者的個人資訊，社交網路，地理位置。這些資料型別的儲存不需要一個固定的格式！不需要多餘的操作就可以橫向擴充套件的！ Map<String,Object> 使用鍵值對來控制！
NoSQL特點
解耦！
-
方便擴充套件(資料之間沒有關係，很好擴充套件)
-
大資料量高效能(Redis一秒寫8萬次，讀取11萬次，NoSQL的快取是記錄級的，是一種細粒度的快取，效能會比較高)
-
資料型別是多樣型的！(不需要事先設計資料庫！隨去隨用！如果是資料量十分大的表，很多人就無法設計了)
-
傳統RDBMS和NoSQL
傳統的RDBMS - 結構化組織 - SQL - 資料和關係都存在單獨的表中 - 操作，資料定義語言 - 嚴格的一致性 - ...NoSQL - 不僅僅是資料 - 沒有固定的查詢語言 - 鍵值對儲存，列儲存，文件儲存，圖形資料庫（社交關係） - 最終一致性 - CAP定理 和 BASE (異地多活) - 高效能，高可用，高可擴 - ....
瞭解：3V+3高
大資料時代的3V：主要是描述問題的
- 海量（Volume）
- 多樣（Variety）
- 實時（Velocity）
大資料時代的3高：主要是對程式的要求
- 高併發
- 高可擴
- 高效能
真正在公司中的實踐：NoSQL+RDBMS 一起使用才是最強的，阿里巴巴的架構演進
1.3、NoSQL的四大分類
KV鍵值對：
- 新浪：Redis
- 美團：Redis + Tair
- 阿里、百度：Redis + memecache
文件型資料庫（bson格式和json一樣）：
- MongoDB（必須要掌握）
- MongoDB是一個基於分散式檔案儲存的資料庫，C++編寫，主要用來處理大量的文件
- MongoDB是一個介於關係型資料庫和非關係型資料庫中間的產品！MongoDB是非關係型資料庫中功能最豐富，最像關係型資料庫的！
- ConthDB
列儲存資料庫：
- HBase
- 分散式檔案系統
圖關係資料庫：
- 他不是存圖形，放的是關係，比如：朋友圈社交網路，廣告推薦！
- Neo4J、infoGrid
四者對比！
| 分類 | Examples舉例 | 典型應用場景 | 資料模型 | 優點 | 缺點 |
|---|---|---|---|---|---|
| 鍵值對(key-value) | Tokyo Cabinet/Tyrant,Redis,Voldemort,Oracle BDB | 內容快取，主要用於處理大量資料的高訪問負載，也用於一些日誌系統等等 | Key指向Value的鍵值對，通常用hash table來實現 | 查詢速度塊 | 資料無結構化，通常只被當作字串或者二進位制資料 |
| 列儲存資料庫 | Cassandra,HBase,Riak | 分散式檔案系統 | 以列簇式儲存，將同一列資料存在一起 | 查詢速度塊，可擴充套件性強，更容易進行分散式擴充套件 | 功能相對侷限 |
| 文件型資料庫 | CouchDB,MongoDB | Web應用（與Key-Value類似，Value是結構化的，不同的是資料庫能夠了解Value的內容) | Key-Value對應的鍵值對Value為結構化資料 | 資料結構要求不嚴格，表結構可變，不需要像關係型資料庫一樣需要預先定義表結構 | 查詢效能不高，而且缺乏統一的查詢語法 |
| 圖形(Graph)資料庫 | Neo4J,InfoGrid,Infinite Graph | 社交網路，推薦系統等。專注於構建關係圖譜 | 圖結構 | 利用圖結構相關演算法。比如最短路徑定址，N度關係查詢等 | 很多時候需要對整個圖做計算才能得出需要的資訊，而且這種結構不太好做分散式的叢集方案 |
2、Redis入門
2.1、概述
Redis是什麼
Redis(Remote Dictionary Server),即遠端字典服務
是一個開源的使用ANSI C語言編寫、支援網路、可基於記憶體亦可持久化的日誌型、Key-Value資料庫，並提供多種語言的APi.
區別的是redis會週期性的把更新的資料寫入磁碟或者把修改操作寫入追加的記錄檔案，並且在此基礎上實現了master-slave(主從)同步。
免費和開源！是當下最熱門的NoSQL技術之一！也被人們稱之為結構化資料庫！
Redis能幹嘛？
1、記憶體儲存、持久化（記憶體中是斷電即失）所以說持久化很重要（RDB,aof）
2、效率高，可以用於快取記憶體
3、釋出訂閱系統
4、地圖資訊分析
5、計時器、計數器（瀏覽量）
6、......
特性
- 多樣的資料型別
- 持久化
- 叢集
- 事務
......
學習中需要用到的東西
redis官網：https://redis.io/
Redis推薦都是在Linux伺服器上搭建的，我們都是基於Linux學習
2.2、Window安裝
-
解壓到自己的環境目錄即可
- 開啟Redis 雙擊服務即可
4.使用reids連線客戶端
Window下使用確實簡單，但是Redis推薦使用Linux
2.3、Linux安裝
-
下載安裝包!
-
解壓Redis的安裝包 程式
- 進入解壓後的檔案，可以看到Redis的配置檔案
-
基本的環境安裝
yum install gcc-c++ make make install
- redis的預設安裝路徑
usr/local/bin
-
將redis配置檔案複製
-
redis預設不是後臺啟動的，修改配置檔案！
-
啟動Redis服務！
-
使用Redis-cli連線
-
檢視redis的程序是否開啟！
ps -ef|grep -
如何關閉Redis服務？
-
再次檢視Redis程序
- 後面會使用單機多Redis啟動叢集測試
2.4、測試效能
redis-benchmark是一個壓力測試工具
redis-benchmark 命令引數
| 序號 | 選項 | 描述 | 預設值 |
|---|---|---|---|
| 1 | -h | 指定伺服器主機名 | 127.0.0.1 |
| 2 | -p | 指定伺服器埠 | 6379 |
| 3 | -s | 指定伺服器 socket | |
| 4 | -c | 指定併發連線數 | 50 |
| 5 | -n | 指定請求數 | 10000 |
| 6 | -d | 以位元組的形式指定 SET/GET 值的資料大小 | 2 |
| 7 | -k | 1=keep alive 0=reconnect | 1 |
| 8 | -r | SET/GET/INCR 使用隨機 key, SADD 使用隨機值 | |
| 9 | -P | 通過管道傳輸 |
1 |
| 10 | -q | 強制退出 redis。僅顯示 query/sec 值 | |
| 11 | --csv | 以 CSV 格式輸出 | |
| 12 | *-l*（L 的小寫字母） | 生成迴圈，永久執行測試 | |
| 13 | -t | 僅執行以逗號分隔的測試命令列表。 | |
| 14 | *-I*（i 的大寫字母） | Idle 模式。僅開啟 N 個 idle 連線並等待。 |
# 測試：100併發連線 100000請求
redis-benchmark -h localhost -p 6379 -c 100 -n 100000
2.5、基礎的知識
redis預設有16個數據庫
預設使用的是第0個
可以使用select進行切換資料庫
127.0.0.1:6379> select 3 # 切換資料庫
OK
127.0.0.1:6379[3]> dbsize # 檢視資料庫大小
(integer) 0
127.0.0.1:6379[3]> keys * # 檢視資料庫所以的key
1) "name"
127.0.0.1:6379[3]>
清除當前資料庫flushdb
127.0.0.1:6379[3]> FLUSHDB # 清除當前資料庫
OK
127.0.0.1:6379[3]> KEYS *
(empty array)
清除全部資料庫內容FLUSHALL
Redis是單執行緒的
明白Redis是很快的，官方表示，Redis是基於記憶體操作，CPU不是Redis效能瓶頸，Redis的瓶頸是根據機器的記憶體和網路頻寬，既然可以使用單執行緒來實現，就使用單執行緒了！所以就使用單執行緒了！
Redis是C語言寫的，官方提供的資料為100000+的QPS,完全不比同樣使用key-value的Memecache差！
Redis為什麼單執行緒還這麼塊？
- 誤區1：高效能的伺服器一定是多執行緒的？
- 誤區2：多執行緒（CPU上下文會切換！）一定比單執行緒效率高！
先去CPU>記憶體>硬碟的速度要有所瞭解
核心：redis是將所有資料全部放在記憶體中，所有說使用單執行緒去操作效率就是最高的，多執行緒（CPU上下文會切換：耗時的操作！！！），對於記憶體系統來說，如果沒有上下文切換效率就是最高的！多次讀寫都是在一個CPU上的，在記憶體情況下，這個就是最佳的方案！
3、五大資料型別
官方文件
Rdis是一個開源(BSD許可)的，記憶體中的資料結構儲存系統，它可以用作資料庫、快取和訊息中介軟體MQ。它支援多種型別的資料結構，如字串(strings),雜湊(hashes),列表(lists),集合（sets),有序集合（sorted sets)與範圍查詢bitmaps,hyperloglogs和地理空間(geospatial)索引半徑查詢。Redis內建了複製(replication),LUA指令碼(Luascripting),LRU驅動事件(LRU eviction),事務（transactions)和不同級別的磁碟持久化(persistence),並通過Redis哨兵(Sentinel)和自動分割槽(Cluster)提供高可用性(high availability)。
3.1、Redis-Key
127.0.0.1:6379> set name pengfei
OK
127.0.0.1:6379> KEYS * # 檢視所有的key
1) "name"
127.0.0.1:6379> set age 1 #set key
OK
127.0.0.1:6379> KEYS *
1) "age"
2) "name"
127.0.0.1:6379> EXISTS name #判斷當前的key是否存在
(integer) 1
127.0.0.1:6379> EXISTS name1
(integer) 0
127.0.0.1:6379> move name 1 # 移除當前的key
(integer) 1
127.0.0.1:6379> KEYS *
1) "age"
127.0.0.1:6379> set name pengfei
OK
127.0.0.1:6379> KEYS *
1) "name"
2) "age"
127.0.0.1:6379>
127.0.0.1:6379> KEYS *
1) "name"
2) "age"
127.0.0.1:6379> get name
"pengfei"
127.0.0.1:6379> EXPIRE name 10 # 設定key的過期時間 單位是秒
(integer) 1
127.0.0.1:6379> ttl name # 檢視當前key的剩餘時間 單位是秒
(integer) -2
127.0.0.1:6379> get name
(nil)
127.0.0.1:6379> set name pengfei
OK
127.0.0.1:6379> KEYS *
1) "name"
2) "age"
127.0.0.1:6379> type name # 檢視當前key的型別
string
127.0.0.1:6379> type age
string
127.0.0.1:6379>
3.2、String(字串)
####################################################
127.0.0.1:6379> set key1 v1 # 設定值
OK
127.0.0.1:6379> get key1 # 獲得值
"v1"
127.0.0.1:6379> keys * # 獲得所有的key
1) "key1"
127.0.0.1:6379> EXISTS key1 #判斷某一個key是否存在
(integer) 1
127.0.0.1:6379> append key1 "hello" # 追加字串，如果當前key不存在，就相當於set key
(integer) 7
127.0.0.1:6379> get key1
"v1hello"
127.0.0.1:6379> STRLEN key1 # 獲取字串的長度
(integer) 7
127.0.0.1:6379> append key1 ",pengfei"
(integer) 15
127.0.0.1:6379> get key1
"v1hello,pengfei"
127.0.0.1:6379>
#######################################################
# 步長
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> get views
"2"
127.0.0.1:6379> decr views # 自減1 瀏覽量-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
127.0.0.1:6379> DECRBY views 5
(integer) 9
####################################################
# 字串範圍 range
127.0.0.1:6379> set key1 "hello,pengfei" # 設定key1值
OK
127.0.0.1:6379> get key1
"hello,pengfei"
127.0.0.1:6379> GETRANGE key1 0 3 #擷取字串 【0，3】
"hell"
127.0.0.1:6379> GETRANGE key1 0 -1 # 獲取全部字串 和get key一樣
"hello,pengfei"
# 替換
127.0.0.1:6379> set key2 abcdefg
OK
127.0.0.1:6379> get key2
"abcdefg"
127.0.0.1:6379> SETRANGE key2 1 xx #替換指定位置開始的字串
(integer) 7
127.0.0.1:6379> get key2
"axxdefg"
####################################################
# setex(set with expire) 設定過期時間
# setnx(set if not exist) 不存在設定（在分散式鎖中常常使用）
127.0.0.1:6379> setex key3 30 "hello" # 設定key3值為hello 30 過期
OK
127.0.0.1:6379> ttl key3
(integer) 24
127.0.0.1:6379> get key3
"hello"
127.0.0.1:6379> setnx mykey "redis" # 如果mykey不存在，建立mykey
(integer) 1
127.0.0.1:6379> keys *
1) "mykey"
127.0.0.1:6379> get key3
(nil)
127.0.0.1:6379> ttl key3
(integer) -2
127.0.0.1:6379> setnx mykey "monggoDB" # 如果mykey存在，建立失敗
(integer) 0
127.0.0.1:6379> get mykey
"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) "k1"
3) "k2"
127.0.0.1:6379> mget k1 k2 k3 # 同時獲取多個值
1) "v1"
2) "v2"
3) "v3"
127.0.0.1:6379> msetnx k1 v1 k4 v4 # msetnx 是一個原子性操作，要麼一起成功，要麼一起失敗
(integer) 0
127.0.0.1:6379> get k4
(nil)
# 物件
set user:1{name:zhangsan,age:3} 設定一個user：1 物件 值為json字串儲存一個物件
# 這裡的key是一個巧妙的設計：user:{id}:{filed}
127.0.0.1:6379> mset user:1:name pengfei user:1:age 2
OK
127.0.0.1:6379> mget user:1:name user:1:age
1) "pengfei"
2) "2"
####################################################
getset # 先get然後set
127.0.0.1:6379> getset db redis #如果不存在值，則返回nil
(nil)
127.0.0.1:6379> get db
"redis"
127.0.0.1:6379> getset db mongoDB # 如果存在值，獲取原來的值，並設定新的值
"redis"
127.0.0.1:6379> get db
"mongoDB"
資料結構是相通的，jedis
String型別的使用場景：value除了是我們的字串還可以是我們的數字！
- 計數器
- 統計多單位的數量
- 粉絲數
- 物件快取儲存
3.3、List
基本的資料型別，列表
在Redis裡面，我們可以把list玩成棧，佇列，阻塞佇列！
所有的list命令都是l開頭的，Redis不區分大小寫命令
###################################################
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> LRANGE list 0 -1
1) "three"
2) "two"
3) "one"
4) "right"
127.0.0.1:6379> LPOP list # 移除list的第一個元素
"three"
127.0.0.1:6379> RPOP list # 移除list的最後一個元素
"right"
127.0.0.1:6379> LRANGE list 0 -1
1) "two"
2) "one"
###################################################
Lindex
127.0.0.1:6379> LRANGE list 0 -1
1) "two"
2) "one"
127.0.0.1:6379> lindex list 0 # 通過下標獲得list中的某一個值！
"two"
###################################################
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> LLEN list # 返回列表的長度
(integer) 3
###################################################
移除指定的值！
lrem
127.0.0.1:6379> LRANGE list 0 -1
1) "three"
2) "three"
3) "two"
4) "one"
127.0.0.1:6379> lrem list 1 one # 移除list集合中指定個數的value ,精確匹配
(integer) 1
127.0.0.1:6379> LRANGE list 0 -1
1) "three"
2) "three"
3) "two"
127.0.0.1:6379> lrem list 1 three
(integer) 1
127.0.0.1:6379> LPUSH list three
(integer) 3
127.0.0.1:6379> lrem list 2 three
(integer) 2
127.0.0.1:6379> LRANGE list 0 -1
1) "two"
###################################################
trim 修剪：list 截斷
127.0.0.1:6379> Rpush mylist "hello"
(integer) 1
127.0.0.1:6379> Rpush mylist "hello1"
(integer) 2
127.0.0.1:6379> Rpush mylist "hello2"
(integer) 3
127.0.0.1:6379> Rpush mylist "hello3"
(integer) 4
127.0.0.1:6379> LRANGE mylist 0 -1
1) "hello"
2) "hello1"
3) "hello2"
4) "hello3"
127.0.0.1:6379> ltrim mylist 1 2 # 通過下標擷取指定的長度，這個list已經被改變了，截斷了只剩下擷取的元素
OK
127.0.0.1:6379> LRANGE mylist 0 -1
1) "hello1"
2) "hello2"
###################################################
rpoplpush # 移除列表的最後一個元素 ，並移動到其他的列表中
127.0.0.1:6379> rpush mylist "hello"
(integer) 1
127.0.0.1:6379> rpush mylist "hello1"
(integer) 2
127.0.0.1:6379> rpush mylist "hello2"
(integer) 3
127.0.0.1:6379> rpoplpush mylist myotherlist # 移除列表的最後一個元素 ，並移動到其他的列表中
"hello2"
127.0.0.1:6379> lrange mylist 0 -1 # 檢視原來的列表
1) "hello"
2) "hello1"
127.0.0.1:6379> lrange myotherlist 0 -1 # 檢視目標列表中，確實存在之前列表中的最後一個元素
1) "hello2"
###################################################
lset 將列表中指定下標的值替換為另外一個值
127.0.0.1:6379> EXISTS list # 判斷這個列表是否存在
(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> LRANGE list 1 other #如果不存在，則會報錯
(error) ERR value is not an integer or out of range
###################################################
linsert # 將某個具體的value插入到列表中某個元素的前面或者後面
127.0.0.1:6379> Rpush mylist "hello"
(integer) 1
127.0.0.1:6379> Rpush mylist "world"
(integer) 2
127.0.0.1:6379> linsert mylist before "world" "other"
(integer) 3
127.0.0.1:6379> LRANGE mylist 0 -1
1) "hello"
2) "other"
3) "world"
127.0.0.1:6379> linsert mylist after "world" "new"
(integer) 4
127.0.0.1:6379> LRANGE mylist 0 -1
1) "hello"
2) "other"
3) "world"
4) "new"
小結
- 它實際上是一個連結串列，before Node after,left,right都可以插入值
- 如果key不存在，建立新的連結串列
- 如果key存在，新增內容
- 如果移除了所有的值,空連結串列，也代表不存在！
- 在兩邊插入或者改動值，效率最高！中間元素，相對來說效率會低一點
訊息排隊！訊息佇列 (Lpush Rpop) ,棧 (Lpush Lpop)
3.4、Set（集合）
set中的值是不能重複的!
###############################################################
127.0.0.1:6379> sadd myset "hello" # set集合中新增元素
(integer) 1
127.0.0.1:6379> sadd myset "hello,world"
(integer) 1
127.0.0.1:6379> sadd myset "hello,nimei"
(integer) 1
127.0.0.1:6379> SMEMBERS myset # 檢視指定set的所有值
1) "hello"
2) "hello,nimei"
3) "hello,world"
127.0.0.1:6379> SISMEMBER myset hello #判斷某個值是否在set集合中
(integer) 1
127.0.0.1:6379> SISMEMBER myset world
(integer) 0
###############################################################
127.0.0.1:6379> scard myset # 獲取set集合裡面元素個數
(integer) 3
127.0.0.1:6379> sadd myset "hello,nime2i"
(integer) 1
127.0.0.1:6379> sadd myset "hello,nime2i"
(integer) 0 #新增重複值返回0，不進入集合
127.0.0.1:6379> scard myset
(integer) 4
###############################################################
127.0.0.1:6379> srem myset hello # 移除set集合中的指定元素
(integer) 1
127.0.0.1:6379> scard myset
(integer) 3
127.0.0.1:6379> SMEMBERS myset
1) "hello,nime2i"
2) "hello,nimei"
3) "hello,world"
###############################################################
set 無序不重複集合，抽隨機
127.0.0.1:6379> SMEMBERS myset
1) "hello,nime2i"
2) "hello,nimei"
3) "hello,world"
127.0.0.1:6379> SRANDMEMBER myset # 隨機抽取一個元素
"hello,world"
127.0.0.1:6379> SRANDMEMBER myset
"hello,nimei"
127.0.0.1:6379> SRANDMEMBER myset
"hello,world"
127.0.0.1:6379> SRANDMEMBER myset
"hello,nime2i"
127.0.0.1:6379> SRANDMEMBER myset 2 # 隨機抽取指定個元素個數
1) "hello,nimei"
2) "hello,world"
###############################################################
刪除指定的key ,隨機刪除一個key
127.0.0.1:6379> spop myset #隨即刪除一些set集合中的元素
"hello,nime2i"
127.0.0.1:6379> spop myset
"hello,nimei"
127.0.0.1:6379> spop myset
"hello,world"
127.0.0.1:6379> SMEMBERS myset
(empty array)
###############################################################
將一個指定的值，移動到另一個集合
127.0.0.1:6379> sadd myset "hello"
(integer) 1
127.0.0.1:6379> sadd myset "hello1"
(integer) 1
127.0.0.1:6379> sadd myset "hello2"
(integer) 1
127.0.0.1:6379> sadd myset2 "set2"
(integer) 1
127.0.0.1:6379> smove myset myset2 "hello" # 將一個指定的值，移動到另一個集合
(integer) 1
127.0.0.1:6379> SMEMBERS myset
1) "hello1"
2) "hello2"
127.0.0.1:6379> SMEMBERS myset2
1) "hello"
2) "set2"
###############################################################
微博，B站，共同關注！（並集）
數字集合類：
- 差集
- 交集
- 並集
127.0.0.1:6379> sadd key1 a
(integer) 1
127.0.0.1:6379> sadd key1 b
(integer) 1
127.0.0.1:6379> sadd key1 c
(integer) 1
127.0.0.1:6379> sadd key2 c
(integer) 1
127.0.0.1:6379> sadd key2 d
(integer) 1
127.0.0.1:6379> sadd key2 e
(integer) 1
127.0.0.1:6379> sdiff key1 key2 # 差集
1) "b"
2) "a"
127.0.0.1:6379> SINTER key1 key2 # 交集
1) "c"
127.0.0.1:6379> SUNION key1 key2 # 並集
1) "a"
2) "d"
3) "c"
4) "b"
5) "e"
微博，A使用者將所有關注的人放在一個set集合中！將它的粉絲也放在一個集合中！
共同關注，共同愛好，二度好友，推薦好友！（六度分隔理論）
3.4、Hash
map集合，key-Map集合！這時候這個值是一個map集合！本質和string型別沒有太大區別，還是一個簡單的key-value
set myhash field pengfei
#######################################################
127.0.0.1:6379> hset myhash field1 pengfei #set一個具體 key-value
(integer) 1
127.0.0.1:6379> hget myhash field1 # 獲取一個欄位值
"pengfei"
127.0.0.1:6379> hmset myhash field1 hello field2 world # set多個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
127.0.0.1:6379> hmset myhash field1 hello field2 world
OK
127.0.0.1:6379> hgetall myhash
1) "field2"
2) "world"
3) "field1"
4) "hello"
127.0.0.1:6379> hlen myhash # 獲取hash表的欄位數量
(integer) 2
#######################################################
127.0.0.1:6379> HEXISTS myhash field1 # 判斷hash中指定欄位是否存在
(integer) 1
127.0.0.1:6379> HEXISTS myhash field3
(integer) 0
#######################################################
# 只獲得所有field
# 只獲得所有value
127.0.0.1:6379> hkeys myhash # 只獲得所有field
1) "field2"
2) "field1"
127.0.0.1:6379> hvals myhash # 只獲得所有value
1) "world"
2) "hello"
#######################################################
incr
decr
127.0.0.1:6379> hset myhash field3 5
(integer) 1
127.0.0.1:6379> HINCRBY myhash field3 1 # 指定增量
(integer) 6
127.0.0.1:6379> HINCRBY myhash field3 -1
(integer) 5
127.0.0.1:6379> HSETnx myhash field4 hello # 如果不存在 則可以設定
(integer) 1
127.0.0.1:6379> HSETnx myhash field4 world# 如果存在 則不可以設定
(integer) 0
127.0.0.1:6379> hset user:1 name pengfei
(integer) 1
127.0.0.1:6379> hget user:1 name
"pengfei"
hash變更的資料user name age,尤其是使用者資訊之類的，經常變動的資訊！hash更適合於物件的儲存，string更加適合字串儲存
3.5、Zset（有序集合）
在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 1 two 4 four # 新增兩個值
(integer) 2
127.0.0.1:6379> zadd myset 2 two
(integer) 0
127.0.0.1:6379> zadd myset 3 three
(integer) 1
127.0.0.1:6379> ZRANge myset 0 -1
1) "one"
2) "two"
3) "three"
############################################################
排序
127.0.0.1:6379> zadd sqlary 2500 xiaohong # 新增三個使用者
(integer) 1
127.0.0.1:6379> zadd sqlary 5000 pengfei
(integer) 1
127.0.0.1:6379> zadd sqlary 10000000 ergou
(integer) 1
127.0.0.1:6379> ZRANGEBYSCORE sqlary -inf +inf # 顯示全部的使用者 按照從小到大排序
1) "xiaohong"
2) "pengfei"
3) "ergou"
127.0.0.1:6379> ZREVRANGE sqlary 0 -1 # 從大到小排序
1) "ergou"
2) "pengfei"
127.0.0.1:6379> ZRANGEBYSCORE sqlary -inf +inf withscores #顯示全部的使用者和排序依據 按照從小到大排序
1) "xiaohong"
2) "2500"
3) "pengfei"
4) "5000"
5) "ergou"
6) "10000000"
127.0.0.1:6379> ZRANGEBYSCORE sqlary -inf 5000 withscores #顯示指定範圍的排序 <5000 升序
1) "xiaohong"
2) "2500"
3) "pengfei"
4) "5000"
############################################################
移除rem中的元素
127.0.0.1:6379> zrange sqlary 0 -1
1) "xiaohong"
2) "pengfei"
3) "ergou"
127.0.0.1:6379> zrem sqlary xiaohong # 移除有序集合中的指定元素
(integer) 1
127.0.0.1:6379> zrange sqlary 0 -1
1) "pengfei"
2) "ergou"
127.0.0.1:6379> zcard sqlary # 獲取有序集合中的個數
(integer) 2
############################################################
127.0.0.1:6379> zadd myset 1 hello
(integer) 1
127.0.0.1:6379> zadd myset 2 world 3 pengfei
(integer) 2
127.0.0.1:6379> zcount myset 1 3 # 獲取指定區間的元素數量
(integer) 3
127.0.0.1:6379> zcount myset 1 2
(integer) 2
案例思路：set排序 儲存班級成績表，工資表排序！
普通訊息，1，重要訊息 2，帶權重進行判斷
排行榜應用實現，Top N 測試！
4、三種特殊資料型別
4.1、geospatial(地理位置)
朋友的定位，附近的人，打車距離計算？
Redis的Geo在Redis3.2版本就推出了！這個功能可以推算地理位置的資訊，兩地之間的距離，方圓幾裡的人
可以查詢一些測試資料：http://www.jsons.cn/lngcode/
- GEOADD
- GEOHASH
- GEODIST
- GEOPOS
- GEORADIUS
- GEORADIUSBYMEMBER
geoadd
# geoadd 新增地理位置
# 規則 兩級無法直接新增，我們一般會下載城市資料，直接通過java程式一次性匯入
# 引數 ： key 值（經度，維度，名稱）
127.0.0.1:6379> geoadd china:city 117.06 36.67 jinan
(integer) 1
127.0.0.1:6379> geoadd china:city 115.97 36.44 liaocheng
(integer) 1
127.0.0.1:6379> geoadd china:city 120.36 36.09 qingdao
(integer) 1
127.0.0.1:6379> geoadd china:city 115.54 34.82 caoxian
(integer) 1
geopos
獲得當前定位：一定是個座標值
127.0.0.1:6379> geopos china:city jinan # 獲取指定城市的經緯度
1) 1) "117.05999940633773804"
2) "36.66999984888297348"
127.0.0.1:6379> geopos china:city liaocheng
1) 1) "115.96999794244766235"
2) "36.43999925088456138"
127.0.0.1:6379> geopos china:city liaocheng qingdao
1) 1) "115.96999794244766235"
2) "36.43999925088456138"
2) 1) "120.36000162363052368"
2) "36.08999988376071855"
geodist
兩個人之間的距離！
單位：
-
m表示單位為米
-
km表示單位為千米
-
mi表示單位為英里
-
ft表示單位為英尺
127.0.0.1:6379> GEODIST china:city jinan liaocheng
"100691.0657"
127.0.0.1:6379> GEODIST china:city jinan liaocheng km # 聊城東昌府區到濟南歷城區的距離
"100.6911"
127.0.0.1:6379> GEODIST china:city liaocheng qingdao km # 聊城東昌府區到青島的距離
"395.5831"
georadius
我附近的人？（獲取所有附近的人的地址，定位！）通過半徑來查詢！
獲得指定數量的人，200
所有的資料都應該錄入：china:city
127.0.0.1:6379> GEORADIUS china:city 120.00 38.00 500 km # 以120 38為中心，尋找方圓1000km的城市
1) "liaocheng"
2) "jinan"
3) "qingdao"
127.0.0.1:6379> GEORADIUS china:city 120.00 38.00 500 km withdist # 顯示到中心距離的位置
1) 1) "liaocheng"
2) "396.8311"
2) 1) "jinan"
2) "299.1176"
3) 1) "qingdao"
2) "214.8322"
127.0.0.1:6379> GEORADIUS china:city 120.00 38.00 500 km withdist withcoord # 顯示他人的位置資訊
1) 1) "liaocheng"
2) "396.8311"
3) 1) "115.96999794244766235"
2) "36.43999925088456138"
2) 1) "jinan"
2) "299.1176"
3) 1) "117.05999940633773804"
2) "36.66999984888297348"
3) 1) "qingdao"
2) "214.8322"
3) 1) "120.36000162363052368"
2) "36.08999988376071855"
127.0.0.1:6379> GEORADIUS china:city 120.00 38.00 500 km withdist withcoord count 1 # 篩選出指定的結果
1) 1) "qingdao"
2) "214.8322"
3) 1) "120.36000162363052368"
2) "36.08999988376071855"
127.0.0.1:6379> GEORADIUS china:city 120.00 38.00 500 km withdist withcoord count 2
1) 1) "qingdao"
2) "214.8322"
3) 1) "120.36000162363052368"
2) "36.08999988376071855"
2) 1) "jinan"
2) "299.1176"
3) 1) "117.05999940633773804"
2) "36.66999984888297348"
GEORADIUSBYMEMBER
127.0.0.1:6379> GEORADIUSBYMEMBER china:city jinan 1000 km # 找出指定位置周圍的城市
1) "caoxian"
2) "liaocheng"
3) "jinan"
4) "qingdao"
127.0.0.1:6379> GEORADIUSBYMEMBER china:city jinan 120 km
1) "liaocheng"
2) "jinan"
geohash 返回一個或多個位置元素的geohash表示
該命令將返回11個字元的geohash字串
# 將二維的經緯度轉換為一維的字串，如果兩個字串越接近，兩者之間的距離越近
127.0.0.1:6379> geohash china:city jinan liaocheng
1) "wwe0xg86px0"
2) "ww6rqcwy2u0"
GEO 底層的實現原理其實就是Zset! 我們可以使用Zset命令來操作geo
127.0.0.1:6379> zrange china:city 0 -1 # 檢視地圖中全部元素
1) "caoxian"
2) "liaocheng"
3) "jinan"
4) "qingdao"
127.0.0.1:6379> zrem china:city qingdao # 移除指定元素
(integer) 1
127.0.0.1:6379> zrange china:city 0 -1
1) "caoxian"
2) "liaocheng"
3) "jinan"
4.2、Hyperloglog
什麼是基數
A{1,3,5,7,8,3}
B{1,3,5,7,8}
基數（不重複的元素）= 5 可以接受誤差
簡介
Redis 2.8.9版本就更新了Hyperloglog資料結構
Redis Hyperloglog 基數統計的演算法！
優點：佔用的記憶體是固定的，2^64 不同的元素的基數，只需要廢12kb記憶體，如果要從記憶體角度來比較的化 Hyperloglog首選！
網頁的 UV(一個人訪問一個網站多次，但是還是算作一個人！)
傳統的方式，set儲存使用者的id,然後就可以統計set中元素數量作為標準判斷！
這個方式如果儲存大量的使用者id,就會比較麻煩！我們的目的是為了計數，而不是儲存使用者id;
0.81%錯誤率！統計UV任務，可以忽略不記！
測試使用
127.0.0.1:6379> Pfadd myset a b c d e f g h i j # 建立第一組元素
(integer) 1
127.0.0.1:6379> PFCOUNT myset # 統計myset元素數量
(integer) 10
127.0.0.1:6379> Pfadd myset2 i j z s w a d m k l # 建立第二組元素
(integer) 1
127.0.0.1:6379> PFCOUNT myset2
(integer) 10
127.0.0.1:6379> PFMERGE myset3 myset myset2 # 合併兩組 myset myset2 到 myset3 並集！
OK
127.0.0.1:6379> PFCOUNT myset3 # 檢視並集的數量
(integer) 16
如果允許容錯，那麼一定使用Hyperloglog！
如果不容錯，就使用set或者自己的資料型別即可！
4.3、Bitmaps
位儲存
統計使用者資訊，活躍，不活躍！登入、未登入！打卡！兩個狀態的，都可以使用Bitmaps
Bitmaps點陣圖，資料結構！都是操作二進位制位來進行記錄，就只有0和1兩個狀態！
測試
使用bitmap 來記錄週一到週日的打卡
週一：1 週二：0 週三：0 週四：1
127.0.0.1:6379> setbit sign 0 1
(integer) 0
127.0.0.1:6379> setbit sign 1 0
(integer) 0
127.0.0.1:6379> setbit sign 3 0
(integer) 0
127.0.0.1:6379> setbit sign 4 1
(integer) 0
127.0.0.1:6379> setbit sign 5 1
(integer) 0
127.0.0.1:6379> setbit sign 6 1
(integer) 0
127.0.0.1:6379> setbit sign 2 0
(integer) 0
檢視某天是否打卡！
127.0.0.1:6379> getbit sign 6
(integer) 1
127.0.0.1:6379> getbit sign 2
(integer) 0
統計操作，統計打卡的天數！
127.0.0.1:6379> bitcount sign # 統計這周的打卡記錄，就可以看到是否有全勤
(integer) 4
5、事務
Redis事務本質：一組命令的集合！一個事務中的所有命令都會被序列化，在事務執行過程中，會按照順序執行，一次性、順序性、排他性！執行一系列的命令！
-------- 佇列 set set set 執行 -------
Redis事務沒有隔離級別的概念
所有的命令在事務中，並沒有被執行！只有發起執行命令的時候才會執行！Exec
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
放棄事務
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)> set k4 v4
QUEUED
127.0.0.1:6379(TX)> DISCARD # 取消事務
OK
127.0.0.1:6379> get k4 # 事務佇列中的命令都不會被執行
(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)> set k3 v3
QUEUED
127.0.0.1:6379(TX)> getset k3 # 錯誤的命令
(error) ERR wrong number of arguments for 'getset' command
127.0.0.1:6379(TX)> set k4 v4
QUEUED
127.0.0.1:6379(TX)> set k5 v5
QUEUED
127.0.0.1:6379(TX)> exec # 執行事務報錯
(error) EXECABORT Transaction discarded because of previous errors.
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)> incr k1 # h
QUEUED
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)> get k3
QUEUED
127.0.0.1:6379(TX)> EXEC
1) (error) ERR value is not an integer or out of range # 雖然第一條命令報錯了，但是依舊執行成功了
2) OK
3) OK
4) "v3"
127.0.0.1:6379> get k2
"v2"
127.0.0.1:6379> get k3
"v3"
監控！Watch
悲觀鎖：
- 很悲觀，什麼時候都會出問題，無論做什麼都加鎖！
樂觀鎖：
- 很樂觀，認為什麼時候都不會出問題，所以不會上鎖！更新資料的時候去判斷一下，在此期間是否有人修改過這個資料, version!
- 獲取version
- 更新的時候比較version
Redis監視測試
正常執行成功！
127.0.0.1:6379> set money 100
OK
127.0.0.1:6379> set out 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 # 監視
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379(TX)> Decrby money 10
QUEUED
127.0.0.1:6379(TX)> incrby out 10
QUEUED
127.0.0.1:6379(TX)> exec # 執行之前，另外一個執行緒修改了我們的值，這個時候再執行事務，會執行失敗
(nil)
如果修改失敗獲取最新值即可！
6、Jedis
什麼是jedis是Redis官方推薦的java連線開發工具！使用java操作Redis中介軟體！如果你要使用java操作redis,那麼一定要對jedis十分熟悉！
測試
-
匯入對應的依賴
<!-- 匯入jedis包--> <dependencies> <!-- https://mvnrepository.com/artifact/redis.clients/jedis --> <dependency> <groupId>redis.clients</groupId> <artifactId>jedis</artifactId> <version>4.2.3</version> </dependency> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>2.0.1</version> </dependency> </dependencies> -
編碼測試
- 連線資料庫
- 操作命令
- 斷開連線！
package com.peng; import redis.clients.jedis.Jedis; public class TestPing { public static void main(String[] args) { // new Jedis 物件即可 Jedis jedis = new Jedis("127.0.0.1",6379); //jedis所以的命令就是我們之前學習的所有指令 System.out.println(jedis.ping()); } }輸出pong
6.1、常用API
-
string
package com.peng; import redis.clients.jedis.Jedis; import java.util.Set; public class TestKey { public static void main(String[] args) { Jedis jedis = new Jedis("127.0.0.1",6379); System.out.println("清空資料："+jedis.flushDB()); System.out.println("判斷某個key是否存在："+jedis.exists("user")); System.out.println("新增<‘username’,'pengfei'>的鍵值對："+jedis.set("username","pengfei")); System.out.println("新增<‘password’,'123456'>的鍵值對："+jedis.set("password","123456")); System.out.println("系統中所以的鍵如下："); 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()); } } -
list
-
set
-
hash
-
zset
所有的api命令就是對應上面學習的指令！
事務
package com.peng;
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("127.0.0.1", 6379);
jedis.flushDB();
JSONObject jsonObject = new JSONObject();
jsonObject.put("hello","world");
jsonObject.put("name","pengfei");
//開啟事務
Transaction multi = jedis.multi();
String result = jsonObject.toJSONString();
// jedis.watch("user1");
try {
multi.set("user1",result);
multi.set("user2",result);
int i = 1/0; //程式碼丟擲異常事務，執行失敗
multi.exec(); //執行事務
} catch (Exception e) {
multi.discard(); //放棄事務
e.printStackTrace();
} finally {
System.out.println(jedis.get("user1"));
System.out.println(jedis.get("user2"));
jedis.close();
}
}
}
7、SpringBoot整合
SpringBoot操作資料：spring-data jpa jdbc mongodb redis!
SpringData也是和SpringBoot齊名的專案！
說明：在springboot2.x之後，原來使用的jedis被替換為lettuce
jedis:採用的是直連，多個執行緒操作的話是不安全的，如果要避免不安全的，使用jedis pool連線池！ BIO
lettuce:採用netty,例項可以在多個執行緒中進行共享，不存線上程不安全的情況！可以減少執行緒數量！NIO
原始碼分析：
@Bean
@ConditionalOnMissingBean(
name = {"redisTemplate"}//我們可以自己定義一個redisTemplate來替換它
)
@ConditionalOnSingleCandidate(RedisConnectionFactory.class)
public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
//預設的redisTemplate沒有過多的設定，redis物件都是需要序列化！
//兩個泛型都是Object,的型別，我們後面需要強制轉換<String,Object>
RedisTemplate<Object, Object> template = new RedisTemplate();
template.setConnectionFactory(redisConnectionFactory);
return template;
}
@Bean
@ConditionalOnMissingBean //由於String是redis中最常用的型別，所有單獨提出了一個bean
@ConditionalOnSingleCandidate(RedisConnectionFactory.class)
public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory redisConnectionFactory) {
return new StringRedisTemplate(redisConnectionFactory);
}
整合測試
-
匯入依賴
<dependencies> <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.springframework.boot</groupId> <artifactId>spring-boot-devtools</artifactId> <scope>runtime</scope> <optional>true</optional> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-configuration-processor</artifactId> <optional>true</optional> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <optional>true</optional> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> </dependencies> -
配置連線
# springboot 所有的配置類，都有一個自動配置類 # 自動配置類都會繫結一個properties 配置檔案 spring.redis.host=127.0.0.1 spring.redis.port=6379 -
測試
@SpringBootTest class SpringbootRedisApplicationTests { @Autowired private RedisTemplate redisTemplate; @Test void contextLoads() { /** * RedisTemplate 操作不同的資料型別，api和我們的指令是一樣的 * opsForValue 操作字串 類似String * opsForList 操作List 類似List * opsForSet 操作Set 類似set * opsForHash 操作hash 類似hash * opsForZSet 操作zset 類似zset * opsForGeo * opsForHyperLogLog * */ /*除了基本的操作，我們常用的方法就可以直接通過redisTemplate操作，比如事務，和基本的crud*/ //獲取redis的連線物件 // RedisConnection connection = redisTemplate.getConnectionFactory().getConnection(); // connection.flushDb(); redisTemplate.opsForValue().set("mykey","pengfei"); System.out.println(redisTemplate.opsForValue().get("mykey")); } }
序列化配置！
關於物件的儲存：
我們編寫一個自己的RedisTemplate
package com.peng.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 redisConnectionFactory) {
//我們為了自己開發方便，一般直接使用<String,Object>
RedisTemplate<String, Object> template = new RedisTemplate();
template.setConnectionFactory(redisConnectionFactory);
//json序列化配置
Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<Object>(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序列化方式採用jackson
template.setValueSerializer(jackson2JsonRedisSerializer);
//hash的value序列化方式採用jackson
template.setHashKeySerializer(jackson2JsonRedisSerializer);
template.afterPropertiesSet();
return template;
}
}
@Test
public void test() throws JsonProcessingException {
//真實的開發一般都是用json來傳遞物件
User user = new User("pengfei", 3);
// String jsonuser = new ObjectMapper().writeValueAsString(user);
redisTemplate.opsForValue().set("user",user);
System.out.println(redisTemplate.opsForValue().get("user"));
}
Redis工具類：
package com.peng.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;
//在我們的真實的開發中，獲取在你們的公司，一般都可以看到公司封裝的工具類
@Component
public final class RedisUtils {
@Autowired
private 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;
}
}
}
測試：
@Autowired
private RedisUtils redisUtils;
@Test
public void test1(){
redisUtils.set("name","pengfei");
System.out.println(redisUtils.get("name"));
}
所有的redis操作，其實對於java開發人員來說，十分簡單，更重要是要去理解redis的思想和每一種資料結構的用處和作用場景！
8、Redis.conf詳解
啟動的時候就通過配置檔案來啟動的！
單位
- 配置檔案unit單位對大小寫不敏感
包含
就好比我們學習spring，import,include
網路
bind 127.0.0.1 -::1 # 繫結的ip
protected-mode yes # 保護模式
port 6379 #埠設定
通用GENERAL
daemonize yes # 預設是no 以守護程序的方式進行
pidfile /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 # 是否顯示logo
快照
持久化，在規定的時間內，執行了多少次操作，則會持久化到檔案.rdb .aof
redis是記憶體資料庫，如果沒有持久化，那麼資料斷電即失
# 如果900s內，如果至少有一個key進行修改，我們及時進行持久化操作
save 900 1
# 如果300s內，如果至少有10個key進行修改，我們及時進行持久化操作
save 300 10
# 如果60s內，如果至少有10000個key進行修改，我們及時進行持久化操作
save 60 10000
# 我們之後學習持久化，會自己定義這個測試
stop-writes-on-bgsave-error yes # 持久化如果出錯了，是否還需要繼續工作
rdbcompression yes # 是否壓縮rdb檔案，需要消耗一些cpu資源！
rdbchecksum yes # 儲存rdb檔案的時候，進行錯誤的校驗檢查
dir ./ # rdb 檔案儲存的目錄
REPLICATION複製 主從複製
SECURITY 安全
可以在這裡設定redis的密碼，預設沒有密碼！
127.0.0.1:6379> config get requirepass # 獲取redis的密碼
1) "requirepass"
2) ""
127.0.0.1:6379> config set requirepass "123456" # 設定redis密碼
OK
127.0.0.1:6379> config get requirepass
1) "requirepass"
2) "123456"
127.0.0.1:6379> ping
PONG
127.0.0.1:6379> auth 123456 # 登入驗證
OK
限制CLIENTS
maxclients 10000 #設定能連線上redis的最大客戶端的數量
maxmemory <bytes> # redis 配置最大記憶體容量
maxmemory-policy noeviction # 記憶體到達上限之後的處理策略
#移除一些過期的key
#報錯
# 。。。
1、volatile-lru: 只對設定了過期時間的key進行LRU(預設值)
2、allkeys-lru: 刪除lru演算法的key
3、volatile-random: 隨機刪除即將過期key
4、allkeys-random: 隨機刪除
5、valatile-ttl: 刪除即將過期的
6、noeviction: 永不過期 ，返回錯誤
APPEND ONLY MODE 模式 aof配置
appendonly no # 預設是不開啟aof模式的，預設是使用rdb方式持久化的，在大部分所有的情況下，rdb完全夠用
appendfilename "appendonly.aof" # 持久化的檔案的名字
# appendfsync always # 每次修改都會sync 消耗效能
appendfsync everysec # 每秒執行一次 sync 可能會丟失這一秒的資料
# appendfsync no # 不執行 sync 這個時候作業系統自己同步資料，速度最快！
9、Redis持久化
Redis是記憶體資料庫，如果不將記憶體中的資料庫狀態儲存到磁碟，那麼一旦伺服器程序退出，伺服器中的資料庫狀態也會消失。所以Redis提供了持久化功能！
RDB(Redis DataBase)
什麼是RDB
在主從複製中，rdb就是備用的！
在指定的時間間隔內將記憶體中的資料集快照寫入磁碟，也就是行話講的Snapshot快照，它恢復時是將快照檔案直接讀到記憶體裡Redis會單獨建立(fork)一個子程序來進行持久化，會先將資料寫入到一個臨時檔案中，待持久化過程都結束了，再用這個臨時檔案替換上次持久化好的檔案。整個過程中，主程序是不進行任何i○操作的。這就確保了極高的效能。如果需要進行大規模資料的恢復，且對於資料恢復的完整性不是非常敏感，那RDB方式要比AOF方式更加的高效。RDB的缺點是最後一次持久化後的資料可能丟失。我們預設的就是RDB,一般情況下不需要修改這個配置！
rdb儲存的檔案是dump.rdb都是我們的配置檔案中快照中進行配置的
觸發機制
- save的規則滿足的情況下，會自動觸發rdb規則
- 執行flushall命令，也會觸發我們的rdb規則
- 退出redis，也會產生rdb檔案！
備份就自動生成一個dump.rdb
如何恢復rdb檔案
- 只需要將rdb檔案放在我們的redis啟動目錄即可，redis啟動的時候就會自動檢測dump.db恢復其中的資料
- 檢視需要存在的位置
127.0.0.1:6379> config get dir
1) "dir"
2) "/usr/local/bin" # 如果在這個目錄下存在dump.rdb檔案 啟動就會自動恢復其中的資料
優點：
- 適合大規模的資料恢復！dump.rdb
- 對資料的完整性要求不高！
缺點：
- 需要一定的時間間隔程序操作！如果redis意外宕機了，這個最後一條修改資料就沒有了
- fork程序的時候，會佔用一定的內容空間！
AOF(Append Only File)
將我們的所有命令都記錄下來，history,恢復的時候就把這個檔案全部再執行一遍
以日誌的形式來記錄每個寫操作，將Redis執行過的所有指令記錄下來（讀操作不記錄），只許追加檔案但不可以改寫檔案，redis啟動之初會讀取該檔案重新構建資料，換言之，Redis重啟的話就根據日誌檔案的內容將寫指令從前到後執行一次以完成資料的恢復工作
Aof儲存的是appendonly.aof檔案
append
預設是不開啟的，我們需要手動修改！我們只需要將appendonly改為yes 就開啟了aof
重啟即可生效！
如果這個aof檔案有錯誤，這個時候redis是啟動不起來的，我們需要修復這個aof檔案
redis給我們提供了一個工具redis-check-aof --fix 檔名
如果檔案正常，重啟就可以恢復了！
重寫規則說明
aof預設就是檔案的無限追加，檔案會越來越大！
如果aof檔案大於64m,太大了！fork一個新的程序來講我們的檔案進行重寫！
優點和缺點
appendonly no # 預設是不開啟aof模式的，預設是使用rdb方式持久化的，在大部分所有的情況下，rdb完全夠用
appendfilename "appendonly.aof" # 持久化的檔案的名字
# appendfsync always # 每次修改都會sync 消耗效能
appendfsync everysec # 每秒執行一次 sync 可能會丟失這一秒的資料
# appendfsync no # 不執行 sync 這個時候作業系統自己同步資料，速度最快！
# rewrite
優點：
- 每一次修改都同步，檔案的完整性會更加好
- 每秒同步一次，可能會丟失一秒的資料
- 從不同步，效率是最高的！
缺點：
- 相對於資料檔案來說，aof遠遠大於rdb，修復的速度也比rdb慢
- aof執行效率也要比rdb慢，所以我們redis預設的配置就是rdb持久化
擴充套件：
1、RDB持久化方式能夠在指定的時間間隔內對你的資料進行快照儲存
2、AOF持久化方式記錄每次對伺服器寫的操作，當伺服器重啟的時候會重新執行這些命令來恢復原始的資料，AOF命令以Redis協議追加儲存每次寫的操作到檔案末尾，Redisi還能對AOF檔案進行後臺重寫，使得AOF檔案的體積不至於過大。
3、只做快取，如果你只希望你的資料在伺服器執行的時候存在，你也可以不使用任何持久化
4、同時開啟兩種持久化方式
- 在這種情況下，當redis重啟的時候會優先載入AOF檔案來恢復原始的資料，因為在通常情況下AOF檔案儲存的資料集要比RDB檔案儲存的資料集要完整。
- RDB的資料不實時，同時使用兩者時伺服器重啟也只會找AOF檔案，那要不要只使用AOF呢？作者建議不要，因為RDB更適合用於備份資料庫(AOF在不斷變化不好備份)，快速重啟，而且不會有AOF可能潛在的Bug,留著作為一個萬一的手段。
5、效能建議
- 因為RDB檔案只用作後備用途，建議只在Slave.上持久化RDB檔案，而且只要15分鐘備份一次就夠了，只保留save9001這條規則。
- 如果Enable AOF,好處是在最惡劣情況下也只會丟失不超過兩秒資料，啟動指令碼較簡單隻load自己的AOF檔案就可以了，代價一是帶來了持續的IO,二是AOF rewrite的最後將rewrite過程中產生的新資料寫到新檔案造成的阻塞幾乎是不可避免的。只要硬碟許可，應該儘量減少AOF rewrite的頻率，AOF重寫的基礎大小預設值64M太小了，可以設到SG以上，預設超過原大小100%大小重寫可以改到適當的數值。
- 如果不Enable AOF,僅靠Master-Slave Repllcation實現高可用性也可以，能省掉一大筆IO,也減少了rewrite時帶來的系統波動。代價是如果Master/Slave同時倒掉(斷電)，會丟失十幾分鐘的資料，啟動指令碼也要比較兩個Master/Slave中的RDB檔案，載入較新的那個，微博就是這種架構。
10、Redis釋出訂閱
Redis釋出訂閱(pub/sub)是一種訊息通訊模式：傳送者(pub)傳送訊息，訂閱者(sub接收訊息。Redis客戶端可以訂閱任意數量的頻道。
訂閱/釋出訊息圖：
第一個：訊息傳送者 第二個：頻道 第三個：訊息訂閱者！
下圖展示了頻道channel1,以及訂閱這個頻道的三個客戶端一一client2、client5和client1之間的關係：
當有新訊息通過PUBLISH命令傳送給頻道channel1時，這個訊息就會被髮送給訂閱它的三個客戶端：
命令
這些命令被廣泛用於構建即時通訊應用，比如網路聊天室(chatroom)和實時廣播、實時提醒等。
| 序號 | 命令及描述 |
|---|---|
| 1 | [PSUBSCRIBE pattern pattern ...] 訂閱一個或多個符合給定模式的頻道。 |
| 2 | [PUBSUB subcommand argument [argument ...]] 檢視訂閱與釋出系統狀態。 |
| 3 | PUBLISH channel message 將資訊傳送到指定的頻道。 |
| 4 | [PUNSUBSCRIBE pattern [pattern ...]] 退訂所有給定模式的頻道。 |
| 5 | [SUBSCRIBE channel channel ...] 訂閱給定的一個或多個頻道的資訊。 |
| 6 | [UNSUBSCRIBE channel [channel ...]] 指退訂給定的頻道。 |
測試
訂閱端：
127.0.0.1:6379> SUBSCRIBE pengfei # 訂閱一個頻道
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "pengfei"
3) (integer) 1
# 等待讀取推送的資訊
1) "message" # 訊息
2) "pengfei" # 哪個頻道的訊息
3) "nimeide" # 訊息的內容
1) "message"
2) "pengfei"
3) "NB"
傳送端：
127.0.0.1:6379> PUBLISH pengfei "nimeide" # 釋出者釋出訊息到頻道
(integer) 1
127.0.0.1:6379> PUBLISH pengfei "NB"
(integer) 1
127.0.0.1:6379>
原理
Redis是使用C實現的，通過分析Redis原始碼裡的pubsub.c檔案，瞭解釋出和訂閱機制的底層實現，籍此加深對Redis的理解。Redis通過PUBLISH、SUBSCRIBE和PSUBSCRIBE等命令實現釋出和訂閱功能
通過SUBSCRIBE命令訂閱某頻道後，redis-server裡維護了一個字典，字典的鍵就是一個個頻道,而字典的值則是一個連結串列，連結串列中儲存了所有訂閱這個channel的客戶端。SUBSCRIBE命令的關鍵，就是將客戶端新增到給定channel的訂閱連結串列中。
通過PUBLISH命令向訂閱者傳送訊息，redis-server會使用給定的頻道作為鍵，在它所維護的channel字典中查詢記錄了訂閱這個頻道的所有客戶端的連結串列，遍歷這個連結串列，將訊息釋出給所有訂閱者。
Pub/Sub從字面上理解就是釋出(Publish)與訂閱(Subscribe),在Redis中，你可以設定對某一個key值進行訊息釋出及訊息訂閱，當一個ky值上進行了訊息釋出後，所有訂閱它的客戶端都會收到相應的訊息。這一功能最明顯的用法就是用作實時訊息系統，比如普通的即時聊天，群聊等功能。
使用場景：
- 實時訊息系統！
- 實時聊天（頻道當作聊天室，將資訊回顯給所有人即可）
- 訂閱，關注系統都是可以的！
稍微複雜的場景我們就會使用訊息中介軟體 MQ
11、Redis主從複製
概念
主從複製，是指將一臺Redis伺服器的資料，複製到其他的Redis伺服器。前者稱為主節點(master/leader),後者稱為從節點(slave/follower);資料的複製是單向的，只能由主節點到從節點。Masterl以寫為主，Slave以讀為主。
預設情況下，每臺Rdis伺服器都是主節點；
且一個主節點可以有多個從節點（或沒有從節點），但一個從節點只能有一個主節點。
主從複製的作用主要包括：
1、資料冗餘：主從複製實現了資料的熱備份，是持久化，之外的一種資料冗餘方式。
2、故障恢復：當主節點出現問題時，可以由從節點提供服務，實現快速的故障恢復；實際上是一種服務的冗餘。
3、負載均衡：在主從複製的基礎上，配合讀寫分離，可以由主節點提供寫服務，由從節點提供讀服務（即寫Rdis資料時應用連線主節點，讀Rdis資料時應用連線從節點)，分擔伺服器負載；尤其是在寫少讀多的場景下，通過多個從節點分擔讀負載，可以大大提高Redis伺服器的併發量。
4、高可用基石（叢集）：除了上述作用以外，主從複製還是哨兵和叢集能夠實施的基礎，因此說主從複製是Rds高可用的基礎。
一般來說，要將Redis運用於工程專案中，只使用一臺Redis,是萬萬不能的（宕機），原因如下：
1、從結構上，單個Rdis伺服器會發生單點故障，並且一臺伺服器需要處理所有的請求負載，壓力較大
2、從容量上，單個Redis伺服器記憶體容量有限，就算一臺Redis伺服器記憶體容量為256G,也不能將所有記憶體用作Redis儲存記憶體，一般來說，單臺Redis最大使用記憶體不應該超過20G。
電商網站上的商品，一般都是一次上傳，無數次瀏覽的，說專業點也就是"多讀少寫”。
對於這種場景，我們可以使如下這種架構：
主從複製，讀寫分離！80%的情況下都是在進行讀操作！減緩伺服器的壓力！架構中經常使用！一主二從！
只要在公司中，主從複製就是必須要使用的，因為在真實的專案中不可能單機使用Redis!
環境配置
只配置從庫，不用配置主庫！
127.0.0.1:6379> info replication # 檢視當前庫的資訊
# Replication
role:master # 角色 master
connected_slaves:0 # 沒有從機
master_failover_state:no-failover
master_replid:c1fdd3edce47dbd04d5a05663e9b15da53561b9d
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
複製三個配置檔案，然後修改對應的資訊
- 埠
- pid名字
- log檔名字
- dump.rdb名字
修改完畢之後啟動我們的三個redis伺服器
一主二從
預設情況下，每臺Rdis伺服器都是主節點；我們一般情況下只用配置從機就好了！
認老大！一主（79）二從（80、81）
127.0.0.1:6380> SLAVEOF 127.0.0.1 6379 # slaveof host 6379
OK
127.0.0.1:6380> info replication
# Replication
role:slave # 當前角色是從機
master_host:127.0.0.1 #可以看到主機的資訊
master_port:6379
master_link_status:down
master_last_io_seconds_ago:-1
master_sync_in_progress:0
slave_read_repl_offset:0
slave_repl_offset:0
master_link_down_since_seconds:-1
slave_priority:100
slave_read_only:1
replica_announced:1
connected_slaves:0
master_failover_state:no-failover
master_replid:e0e0baef3b4a566051eb180651c4dd76a6eeda50
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
在主機中檢視
127.0.0.1:6379> info replication
# Replication
role:master
connected_slaves:1 #多了從機的配置
slave0:ip=127.0.0.1,port=6380,state=online,offset=28,lag=0
master_failover_state:no-failover
master_replid:9362865c1001b2c98c6e65259d359abb4dd1ddf0
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=294,lag=1
slave1:ip=127.0.0.1,port=6381,state=online,offset=294,lag=1
master_failover_state:no-failover
master_replid:9362865c1001b2c98c6e65259d359abb4dd1ddf0
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:294
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:1
repl_backlog_histlen:294
真實的主從配置應該在配置檔案中配置，這樣的話是永久的，我們這裡使用的是命令，暫時的！
細節
主機可以寫，從機不能寫只能讀！主機中的所有資訊和資料，都會自動被從機儲存！
主機寫：
從機只能讀不能寫：
測試：主機斷開連線，從機依舊連線到主機的，但是沒有寫操作，這個時候，主機如果回來了，從機依舊可以直接獲取到主機寫的資訊！
如果是使用命令列，來配置的主從，這個時候如果重啟了，就會變為主機！只要變回從機，立馬就可以從主機獲取值。
複製原理
Slave啟動成功連線到master後會傳送一個sync命令
Master接到命令，啟動後臺的存檔程序，同時收集所有接收到的用於修改資料集命令，在後臺程序執行完畢之後，master將傳送整個資料檔案到slave,並完成一次完全同步。
全量複製：而slve服務在接收到資料庫檔案資料後，將其存檔並載入到記憶體中。
增量複製：Master繼續將新的所有收集到的修改命令依次傳給slave,完成同步
但是隻要是重新連線master,一次完全同步（全量複製）將被自動執行!我們的資料一定可以在從機中看到！
層層鏈路
上一個M連結下一個S!
這時候也可以完成我們的主從複製！
如果沒有老大了，這個時候能不能選擇一個老大出來呢？手動！
如果主機斷開了連線，我們可以使用SLAVEOF no one讓自己變成主機！其他節點就可以手動連線到最新的主節點（手動）！如果這個時候老大修復了，就要重新連線！
哨兵模式（重點）
自動選取老大
概述
主從切換技術的方法是：當主伺服器宕機後，需要手動把一臺從伺服器切換為主伺服器，這就需要人工干預，費時費力，還會造成一段時間內服務不可用。這不是一種推薦的方式，更多時候，我們優先考慮哨兵模式。Redis從2.8開始正式提供了Sentinel(哨兵)架構來解決這個問題。
謀朝篡位的自動版，能夠後臺監控主機是否故障，如果故障了根據投票數自動將從庫轉換為主庫。
哨兵模式是一種特殊的模式，首先Rdis提供了哨兵的命令，哨兵是一個獨立的程序，作為程序，它會獨立執行。其原理是哨兵通過傳送命令，等待Redis伺服器響應，從而監控執行的多個Redis例項。
這裡的哨兵有兩個作用
- 通過傳送命令，讓Rdis伺服器返回監控其執行狀態，包括主伺服器和從伺服器。
- 當哨兵監測到master宕機，會自動將slavet切換成master,然後通過釋出訂閱模式通知其他的從伺服器，修改配置檔案，讓它們切換主機。
- 然而一個哨兵程序對Rdis伺服器進行監控，可能會出現問題，為此，我們可以使用多個哨兵進行監控。各個哨兵之間還會進行監控，這樣就形成了多哨兵模式。
假設主伺服器宕機，哨兵1先檢測到這個結果，系統並不會馬上進行failoveri過程，僅僅是哨兵1主觀的認為主伺服器不可用，這個現象成為主觀下線。當後面的哨兵也檢測到主伺服器不可用，並且數量達到一定值時，那麼哨兵之間就會進行一次投票，投票的結果由一個哨兵發起，進行failover[故障轉移]操作。切換成功後，就會通過釋出訂閱模式，讓各個哨兵把自己監控的從伺服器實現切換主機，這個過程稱為客觀下線。
測試
目前狀態一主二從
-
配置哨兵配置檔案sentinel.conf
# sentinel monitor 被監控的名稱 host port 1 sentinel monitor myredis 127.0.0.1 6379 1後面的這個數字1，代表主機掛了，slave投票看讓誰接替成為主機，票數最多的，就會成為主機
-
啟動哨兵！
[root@pengfei bin]# redis-sentinel pconfig/sentinel.conf 4568:X 26 May 2022 20:24:19.327 # oO0OoO0OoO0Oo Redis is starting oO0OoO0OoO0Oo 4568:X 26 May 2022 20:24:19.327 # Redis version=7.0.0, bits=64, commit=00000000, modified=0, pid=4568, just started 4568:X 26 May 2022 20:24:19.327 # Configuration loaded 4568:X 26 May 2022 20:24:19.327 * monotonic clock: POSIX clock_gettime _._ _.-``__ ''-._ _.-`` `. `_. ''-._ Redis 7.0.0 (00000000/0) 64 bit .-`` .-```. ```\/ _.,_ ''-._ ( ' , .-` | `, ) Running in sentinel mode |`-._`-...-` __...-.``-._|'` _.-'| Port: 26379 | `-._ `._ / _.-' | PID: 4568 `-._ `-._ `-./ _.-' _.-' |`-._`-._ `-.__.-' _.-'_.-'| | `-._`-._ _.-'_.-' | https://redis.io `-._ `-._`-.__.-'_.-' _.-' |`-._`-._ `-.__.-' _.-'_.-'| | `-._`-._ _.-'_.-' | `-._ `-._`-.__.-'_.-' _.-' `-._ `-.__.-' _.-' `-._ _.-' `-.__.-' 4568:X 26 May 2022 20:24:19.331 * Sentinel new configuration saved on disk 4568:X 26 May 2022 20:24:19.331 # Sentinel ID is 2931573b676ced8e3605333c1207a719d8bfb3db 4568:X 26 May 2022 20:24:19.331 # +monitor master myredis 127.0.0.1 6379 quorum 1 4568:X 26 May 2022 20:24:19.332 * +slave slave 127.0.0.1:6380 127.0.0.1 6380 @ myredis 127.0.0.1 6379 4568:X 26 May 2022 20:24:19.335 * Sentinel new configuration saved on disk 4568:X 26 May 2022 20:24:19.335 * +slave slave 127.0.0.1:6381 127.0.0.1 6381 @ myredis 127.0.0.1 6379 4568:X 26 May 2022 20:24:19.338 * Sentinel new configuration saved on disk -
如果Master節點斷開了，這個時候就會在從機中隨機選擇一個伺服器（這裡有一個投票的演算法）
哨兵模式
如果主機此時回來了，只能歸併到新的主機下，當作從機，這就是哨兵模式的規則！
優點：
- 哨兵叢集，基於主從複製模式，所有的主從配置優點，它全有
- 主從可以切換，故障可以轉移，系統的可用性就會更好
- 哨兵模式就是主從模式的升級，手動到自動，更加健壯
缺點：
- Redis不好線上擴容，叢集容量一旦到達上限，線上擴容就十分麻煩
- 實現哨兵模式的配置其實是很麻煩的，裡面有很多選擇！
哨兵模式的全部配置
# 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無法正常啟動成功。
#通知指令碼
# 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
12、快取穿透和雪崩
快取穿透
概念
當用戶想要查詢某個資料，Redis記憶體資料庫中沒有，也就是快取沒有命中，於是向持久層資料庫發起請求；當用戶很多的時候，Redis快取都沒有命中，都去請求持久層資料庫，會造成持久層資料庫很大的壓力，這就是快取穿透
解決方案
- 布隆過濾器
- 快取空物件
一次請求若在快取和資料庫中都沒找到，就在快取中放一個該請求的空物件用於處理後續請求
快取擊穿（量太大，資料過期）
快取擊穿是指快取中存在一個熱點key，大併發集中對這點進行訪問，當這個key在失效後瞬間，持續的大併發就直接請求持久層資料庫，導致資料庫壓力過大
解決方案
- 熱點資料永不過期
- 加互斥鎖(分散式鎖)
使用分散式鎖保證每個key同時只有一個執行緒去查詢後端服務，其他執行緒沒有獲得分散式鎖需要等待，這樣就將高併發的壓力轉移到了分散式鎖
快取雪崩
在某一個時間段，快取集中過期或者Redis宕機
解決方案
- Redis高可用：搭建Redis叢集
- 限流降級：在快取失效後，通過加鎖或者佇列來控制讀資料庫寫快取的執行緒數量。比如對某個key只允許一個執行緒查詢資料和寫快取，其他執行緒等待
- 資料預熱：把可能的資料先預先訪問一遍，這樣部分可能大量訪問的資料就會載入到快取中；在即將發生大併發訪問前手動觸發載入快取不同的key，設定不同的過期時間，讓快取失效的時間點儘量均勻*