分布式緩存

緩存的基本作用是在高并發(fā)場景下對應(yīng)服務(wù)的保護(hù)緩沖

– 基于Redis集群解決單機Redis存在的問題

單機的Redis存在四大問題：

• redis由于高強度性能采用內(nèi)存 但是意味著丟失的風(fēng)險
• 單結(jié)點redis并發(fā)能力有限
• 分布式服務(wù)中數(shù)據(jù)過多 依賴內(nèi)存的redis 明顯單機不能滿足
• 如果發(fā)生故障 需要能繼續(xù)服務(wù)

很明顯上述效果需要集群才能實現(xiàn)

解決方案

1.Redis持久化

Redis有兩種持久化方案：

• RDB持久化
• AOF持久化

1.1.RDB持久化

RDB全稱Redis Database Backup file（Redis數(shù)據(jù)備份文件），也被叫做Redis數(shù)據(jù)快照。簡單來說就是把內(nèi)存中的所有數(shù)據(jù)都記錄到磁盤中。當(dāng)Redis實例故障重啟后，從磁盤讀取快照文件，恢復(fù)數(shù)據(jù)。快照文件稱為RDB文件，默認(rèn)是保存在當(dāng)前運行目錄。

1.1.1.執(zhí)行時機

RDB持久化在四種情況下會執(zhí)行：

• 執(zhí)行save命令
• 執(zhí)行bgsave命令
• Redis停機時
• 觸發(fā)RDB條件時

1）save命令

執(zhí)行save的命令，可以立即執(zhí)行一次RDB：
redis是單線程,所以這里會耗時較久

save命令會導(dǎo)致主進(jìn)程執(zhí)行RDB，這個過程中其它所有命令都會被阻塞。只有在數(shù)據(jù)遷移時可能用到。

2）bgsave命令

下面的命令可以異步執(zhí)行RDB(background save)：

這個命令執(zhí)行后會開啟獨立進(jìn)程完成RDB，主進(jìn)程可以持續(xù)處理用戶請求，不受影響。(異步執(zhí)行持久化)

3）停機時

Redis停機時會執(zhí)行一次save命令，實現(xiàn)RDB持久化。
比如先啟動redis

然后停機

發(fā)現(xiàn)redis 輸出log 數(shù)據(jù)保存到磁盤, 以及保存到磁盤上(一般是運行目錄無論windos還是linux)

再次啟動
之前保存的數(shù)據(jù)被讀取 實現(xiàn)redis的持久化

日志輸出數(shù)據(jù)從磁盤讀取

4）觸發(fā)RDB條件

Redis內(nèi)部有觸發(fā)RDB的機制，可以在redis.conf文件中找到，格式如下：

# 900秒內(nèi)，如果至少有1個key被修改，則執(zhí)行bgsave ， 如果是save "" 則表示禁用RDB
save 900 1  
save 300 10  
save 60 10000 

RDB的其它配置也可以在redis.conf文件(windos系統(tǒng)是redis.windos.conf)中設(shè)置：

redis
默認(rèn)持久化策略

# 是否壓縮 ,建議不開啟，壓縮也會消耗cpu，磁盤的話不值錢
rdbcompression yes

# RDB文件名稱
dbfilename dump.rdb  

# 文件保存的路徑目錄
dir ./ 

這里為了測試把觸發(fā)持久化事件改短一點

5秒內(nèi)發(fā)生修改就持久化
重啟后 設(shè)置一個數(shù)據(jù)
日志輸出

后臺備份成功
設(shè)置備份時間的間隔應(yīng)該考慮業(yè)務(wù)中數(shù)據(jù)量的大小和耗時

1.1.2.RDB原理

bgsave開始時會fork主進(jìn)程得到子進(jìn)程，子進(jìn)程共享主進(jìn)程的內(nèi)存數(shù)據(jù)。完成fork后讀取內(nèi)存數(shù)據(jù)并寫入 RDB 文件。

fork采用的是copy-on-write技術(shù)：

• 當(dāng)主進(jìn)程執(zhí)行讀操作時，訪問共享內(nèi)存；
• 當(dāng)主進(jìn)程執(zhí)行寫操作時，則會拷貝一份數(shù)據(jù)，執(zhí)行寫操作。

所以如果在分布式業(yè)務(wù)中,有一臺服務(wù)器是專門用于做緩存,那么的為了redis的備份機制正常運行,應(yīng)該預(yù)留一些內(nèi)存給子進(jìn)程fork

1.1.3.小結(jié)

RDB方式bgsave的基本流程

• fork主進(jìn)程得到一個子進(jìn)程，共享內(nèi)存空間
• 子進(jìn)程讀取內(nèi)存數(shù)據(jù)并寫入新的RDB文件
• 用新RDB文件替換舊的RDB文件

RDB會在什么時候執(zhí)行

• 默認(rèn)是服務(wù)停止時
• 可以通通過配置環(huán)境修改頻率

RDB的缺點

• RDB執(zhí)行間隔時間長(如果修改配置備份時間變短,會大致大量資源消耗,耗時壓力更大)，兩次RDB之間寫入數(shù)據(jù)有丟失的風(fēng)險
• fork子進(jìn)程、壓縮、寫出RDB文件都比較耗時

1.2.AOF持久化

1.2.1.AOF原理

AOF全稱為Append Only File（追加文件）。Redis處理的每一個寫命令都會記錄在AOF文件，因為記錄的不是完整的數(shù)據(jù),而是數(shù)據(jù)的操作命令 可以看做是命令日志文件。

當(dāng)redis 宕機 ,redis重啟后 讀取aof文件,執(zhí)行指令來恢復(fù)數(shù)據(jù)

1.2.2.AOF配置

AOF默認(rèn)是關(guān)閉的，需要修改redis.conf配置文件來開啟AOF：

# 是否開啟AOF功能，默認(rèn)是no
appendonly yes
# AOF文件的名稱
appendfilename "appendonly.aof"

選擇執(zhí)行AOF前先將rdb 策略進(jìn)行注釋

開啟aof策略

重啟redis

log rdb 沒有讀取 說明確實禁止了
然后寫幾個存儲

此時aof 文件出現(xiàn)當(dāng)前目錄

打開查看 確實記錄的是命令

關(guān)閉服務(wù)且從重啟

關(guān)閉時候同步到本地日志

讀取數(shù)據(jù) 成功 并沒丟失

AOF的命令記錄的頻率也可以通過redis.conf文件來配：

# 表示每執(zhí)行一次寫命令，立即記錄到AOF文件 redis是單線程業(yè)務(wù) 這樣的同步需求 對性能消耗較大
appendfsync always 
# 寫命令執(zhí)行完先放入AOF緩沖區(qū)，然后表示每隔1秒將緩沖區(qū)數(shù)據(jù)寫到AOF文件，是默認(rèn)方案
appendfsync everysec 
# 寫命令執(zhí)行完先放入AOF緩沖區(qū)，由操作系統(tǒng)決定何時將緩沖區(qū)內(nèi)容寫回磁盤
appendfsync no

三種策略對比：

1.2.3.AOF文件重寫

因為是記錄命令，AOF文件會比RDB文件大的多。而且AOF會記錄對同一個key的多次寫操作，但只有最后一次寫操作才有意義。通過執(zhí)行bgrewriteaof命令，可以讓AOF文件執(zhí)行重寫功能，用最少的命令達(dá)到相同效果。

如圖，AOF原本有三個命令，但是set num 123 和 set num 666都是對num的操作，第二次會覆蓋第一次的值，因此第一個命令記錄下來沒有意義。

所以重寫命令后，AOF文件內(nèi)容就是：mset name jack num 666

在最后一次寫入操作后執(zhí)行

Redis也會在觸發(fā)閾值時自動去重寫AOF文件。閾值也可以在redis.conf中配置：

# AOF文件比上次文件 增長超過多少百分比則觸發(fā)重寫
auto-aof-rewrite-percentage 100
# AOF文件體積最小多大以上才觸發(fā)重寫 
auto-aof-rewrite-min-size 64mb 

1.3.RDB與AOF對比

RDB和AOF各有自己的優(yōu)缺點，如果對數(shù)據(jù)安全性要求較高，在實際開發(fā)中往往會結(jié)合兩者來使用。

2.Redis主從

2.1.搭建主從架構(gòu)

單節(jié)點Redis的并發(fā)能力是有上限的，要進(jìn)一步提高Redis的并發(fā)能力，就需要搭建主從集群(之前的常用集群模式是負(fù)載均衡)，實現(xiàn)讀寫分離。

Redis集群和主從復(fù)制是兩種不同的Redis部署模式，它們各自解決不同的問題和提供不同的功能。以下是它們的主要區(qū)別和用途：

1. Redis主從復(fù)制（Replication）：

2. 用途： 主從復(fù)制的主要目的是提高數(shù)據(jù)的可用性和可靠性。在主從復(fù)制中，一個Redis主節(jié)點（master）會將數(shù)據(jù)復(fù)制給一個或多個從節(jié)點（slaves）。

3. 數(shù)據(jù)復(fù)制： 主節(jié)點負(fù)責(zé)寫操作，從節(jié)點復(fù)制主節(jié)點的數(shù)據(jù)。這有助于實現(xiàn)數(shù)據(jù)備份和故障恢復(fù)。如果主節(jié)點出現(xiàn)故障，可以快速切換到從節(jié)點來提供服務(wù)，從而降低了系統(tǒng)的單點故障風(fēng)險。

4. 性能： 主從復(fù)制通常不旨在提高性能，因為主節(jié)點仍然處理所有的寫操作，而從節(jié)點主要用于讀取和故障切換。

Redis集群（Cluster）：

6. 用途： Redis集群的主要目的是橫向擴展Redis以處理大規(guī)模的數(shù)據(jù)和請求。它是一種分布式系統(tǒng)，用于提高性能、負(fù)載均衡和高可用性。
   7.數(shù)據(jù)分片： Redis集群將數(shù)據(jù)分為多個分片，每個分片可以在集群中的不同節(jié)點上。這有助于平衡負(fù)載和提高性能，因為請求可以同時分布到多個節(jié)點上。
7. 高可用性： Redis集群也提供了高可用性，因為它可以自動管理節(jié)點故障，并在需要時重新分配分片。

為什么不采用負(fù)載均衡來搭建Redis集群呢？

負(fù)載均衡和Redis集群的目標(biāo)不同： 負(fù)載均衡通常用于分發(fā)流量到多個相同的后端服務(wù)器，以提高性能和可用性。但Redis集群的目標(biāo)更廣泛，包括數(shù)據(jù)分片、高可用性、負(fù)載均衡等，它們是分布式系統(tǒng)的一部分。

數(shù)據(jù)一致性和分片： 在負(fù)載均衡中，通常不會考慮數(shù)據(jù)一致性，因為它主要關(guān)注請求的分發(fā)。然而，Redis集群需要確保數(shù)據(jù)的一致性和可用性，因此它使用數(shù)據(jù)分片和復(fù)制來實現(xiàn)這一目標(biāo)。
性能和可用性需求： 如果您需要橫向擴展Redis以處理大量請求并提供高可用性，那么Redis集群是更合適的選擇。負(fù)載均衡通常在單一后端服務(wù)或多個相同的后端服務(wù)之間分發(fā)請求，但它不提供數(shù)據(jù)的分片和自動故障切換。

綜上所述，Redis主從復(fù)制和Redis集群是針對不同需求的兩種不同部署方式。您應(yīng)根據(jù)您的具體需求和目標(biāo)來選擇適當(dāng)?shù)牟渴鸱绞?。如果需要橫向擴展和高可用性，Redis集群可能更適合您。如果只需備份和故障切換，主從復(fù)制可能足夠。負(fù)載均衡通常在應(yīng)用層用于分發(fā)請求到多個Redis節(jié)點，但它通常不提供數(shù)據(jù)分片和高可用性功能。

這里采用docker實現(xiàn)采用

Docker 部署主從redis集群

創(chuàng)建一個目錄作為演示集群的文件目錄,其中創(chuàng)建三個7001,7002,7003表示redis的三臺實列目錄,端口和目錄名字對應(yīng)

進(jìn)行各個子目錄創(chuàng)建對應(yīng)數(shù)據(jù)卷掛載目錄

創(chuàng)建一個data目錄 和re創(chuàng)建一個redis.config

配置文件的內(nèi)容就按照自己從默認(rèn)reidis中按照需求寫更改
然后運行docker 進(jìn)行掛載
掛載數(shù)據(jù)券和配置文件

docker run -d -p 7001:6379   -v /home/hadoop/Redis-cluster/7001/redis-data-7001/:/data   
-v /home/hadoop/Rediscluster/7001/redis.config:/data/redis.conf   --name redis7001 redis redis-server /data/redis.conf

進(jìn)入容器

docker exec -it redis7001  redis-cli

根據(jù)配置文件的密碼登錄后創(chuàng)建數(shù)據(jù),在進(jìn)行重啟

關(guān)閉服務(wù)

docker stop redis7001

日志和之前一樣

重啟

docker start redis7001

日志輸出

讀取數(shù)據(jù) 發(fā)現(xiàn)可以做到數(shù)據(jù)持久化

按照同樣步驟在啟動倆臺redis容器

成功啟動三臺實列
為了實現(xiàn)主從同步 ,將aof模式關(guān)閉,開始RDB配置

# 開啟RDB
# save ""
save 3600 1
save 300 100
save 60 10000


# 關(guān)閉AOF
appendonly no

集群配置

在使用Docker搭建Redis集群時，Redis官方提供了一個叫做Redis集群模式的特性。使用集群模式，Redis實例可以在不修改配置文件的情況下組成集群。Redis集群模式是通過Redis的內(nèi)建集群支持實現(xiàn)的，不需要手動修改配置文件。
在Redis集群中，每個節(jié)點都可以是主節(jié)點也可以是從節(jié)點。節(jié)點之間通過Gossip協(xié)議進(jìn)行通信。在Docker中，你可以簡單地通過運行多個Redis容器，并在它們之間配置集群。
以下是在Docker中使用集群模式配置Redis集群的步驟：

1. 創(chuàng)建Redis集群網(wǎng)絡(luò)：
   首先，創(chuàng)建一個Docker網(wǎng)絡(luò)，以便在Redis容器之間進(jìn)行通信。你可以使用以下命令創(chuàng)建一個名為redis-net的網(wǎng)絡(luò)：
   docker network create redis-net

2. 啟動Redis容器：
   運行多個Redis容器，并將它們加入到redis-net網(wǎng)絡(luò)中。在這里，你只需要指定容器的名稱、網(wǎng)絡(luò)、端口映射等信息。Docker會為每個容器自動分配一個唯一的Container ID和IP地址。

docker run -d --name redis-node1 --net redis-net -p 7001:6379 redis
docker run -d --name redis-node2 --net redis-net -p 7002:6379 redis
docker run -d --name redis-node3 --net redis-net -p 7003:6379 redis
# ...添加更多的Redis節(jié)點

3. 配置集群：
   連接到任意一個Redis容器中，并使用redis-cli命令配置集群。在下面的命令中，你需要指定所有的Redis節(jié)點地址及端口，以及–cluster-replicas 1參數(shù)表示每個主節(jié)點有一個從節(jié)點：

docker exec -it redis-node1 redis-cli --cluster create \
  172.18.0.2:6379 172.18.0.3:6379 172.18.0.4:6379 \
  172.18.0.5:6379 172.18.0.6:6379 172.18.0.7:6379 \
  --cluster-replicas 1

上述命令中，172.18.0.2至172.18.0.7是各個Redis容器的IP地址。運行這個命令后，Redis集群就會自動配置好，每個節(jié)點都知道其他節(jié)點的信息。
4. 驗證集群配置：
你可以使用以下命令驗證Redis集群的配置是否成功：

docker exec -it redis-node1 redis-cli cluster nodes

這將顯示Redis集群的節(jié)點信息。如果所有節(jié)點都顯示正常，那么你的Redis集群就已經(jīng)配置成功了,這里主要演示主從結(jié)果,所以不做演示

請注意，這種配置方式是基于Docker內(nèi)建的集群支持實現(xiàn)的，不需要手動修改配置文件。但在生產(chǎn)環(huán)境中，你可能需要進(jìn)一步配置安全性、持久性和高可用性等方面的參數(shù)，以確保集群的穩(wěn)定性和安全性。

2.2.主從數(shù)據(jù)同步原理

2.2.1.全量同步

主從第一次建立連接時，會執(zhí)行全量同步，將master節(jié)點的所有數(shù)據(jù)都拷貝給slave節(jié)點，流程：

這里有一個問題，master如何得知salve是第一次來連接呢？？

有幾個概念，可以作為判斷依據(jù)：

• Replication Id：簡稱replid，是數(shù)據(jù)集的標(biāo)記，id一致則說明是同一數(shù)據(jù)集。每一個master都有唯一的replid，slave則會繼承master節(jié)點的replid,從節(jié)點第一次申請同步id不同,所以會接收到 主節(jié)點的數(shù)據(jù),后續(xù)操作進(jìn)行通信時,id相等,說明以及是同步
• offset：偏移量，隨著記錄在repl_baklog中的數(shù)據(jù)增多而逐漸增大。slave完成同步時也會記錄當(dāng)前同步的offset。如果slave的offset小于master的offset，說明slave數(shù)據(jù)落后于master，需要更新。

因此slave做數(shù)據(jù)同步，必須向master聲明自己的replication id 和offset，master才可以判斷到底需要同步哪些數(shù)據(jù)。

因為slave原本也是一個master，有自己的replid和offset，當(dāng)?shù)谝淮巫兂蓅lave，與master建立連接時，發(fā)送的replid和offset是自己的replid和offset。

master判斷發(fā)現(xiàn)slave發(fā)送來的replid與自己的不一致，說明這是一個全新的slave，就知道要做全量同步了。

master會將自己的replid和offset都發(fā)送給這個slave，slave保存這些信息。以后slave的replid就與master一致了。

因此，master判斷一個節(jié)點是否是第一次同步的依據(jù)，就是看replid是否一致。

如圖：

完整流程描述：

• slave節(jié)點請求增量同步
• master節(jié)點判斷replid，發(fā)現(xiàn)不一致，拒絕增量同步
• master將完整內(nèi)存數(shù)據(jù)生成RDB，發(fā)送RDB到slave
• slave清空本地數(shù)據(jù)，加載master的RDB
• master將RDB期間的命令記錄在repl_baklog，并持續(xù)將log中的命令發(fā)送給slave
• slave執(zhí)行接收到的命令，保持與master之間的同步

2.2.2.增量同步

全量同步需要先做RDB，然后將RDB文件通過網(wǎng)絡(luò)傳輸個slave，成本太高了,資源消耗過多。因此除了第一次做全量同步，其它大多數(shù)時候slave與master都是做增量同步。

什么是增量同步？就是只更新slave與master存在差異的部分?jǐn)?shù)據(jù)(主節(jié)點數(shù)據(jù)更新,從結(jié)點因為宕機或網(wǎng)絡(luò)問題沒有一直,只做這部分差異更新)。如圖：

2.2.3.repl_backlog原理

master怎么知道slave與自己的數(shù)據(jù)差異在哪里呢?

這就要說到全量同步時的repl_baklog文件了。

這個文件是一個固定大小的數(shù)組，只不過數(shù)組是環(huán)形，也就是說角標(biāo)到達(dá)數(shù)組末尾后，會再次從0開始讀寫，這樣數(shù)組頭部的數(shù)據(jù)就會被覆蓋。

repl_baklog中會記錄Redis處理過的命令日志及offset，包括master當(dāng)前的offset，和slave已經(jīng)拷貝到的offset：

slave與master的offset之間的差異，就是salve需要增量拷貝的數(shù)據(jù)了。

隨著不斷有數(shù)據(jù)寫入，master的offset逐漸變大，slave也不斷的拷貝，追趕master的offset：

直到數(shù)組被填滿：

此時，如果有新的數(shù)據(jù)寫入，就會覆蓋數(shù)組中的舊數(shù)據(jù)。不過，舊的數(shù)據(jù)只要是綠色的，說明是已經(jīng)被同步到slave的數(shù)據(jù)，即便被覆蓋了也沒什么影響。因為未同步的僅僅是紅色部分。

但是，如果slave出現(xiàn)網(wǎng)絡(luò)阻塞，導(dǎo)致master的offset遠(yuǎn)遠(yuǎn)超過了slave的offset：

如果master繼續(xù)寫入新數(shù)據(jù)，其offset就會覆蓋舊的數(shù)據(jù)，直到將slave現(xiàn)在的offset也覆蓋：

棕色框中的紅色部分，就是尚未同步，但是卻已經(jīng)被覆蓋的數(shù)據(jù)。此時如果slave恢復(fù)，需要同步，卻發(fā)現(xiàn)自己的offset都沒有了，無法完成增量同步了。只能做全量同步。

2.3.主從同步優(yōu)化

主從同步可以保證主從數(shù)據(jù)的一致性，非常重要。

可以從以下幾個方面來優(yōu)化Redis主從就集群：

• 在master中配置repl-diskless-sync yes啟用無磁盤復(fù)制，避免全量同步時的磁盤IO。(要求服務(wù)器網(wǎng)絡(luò)高,而讀取磁盤慢,不然會造擦網(wǎng)絡(luò)服務(wù)堵塞)
• Redis單節(jié)點上的內(nèi)存占用不要太大，減少RDB導(dǎo)致的過多磁盤IO
• 適當(dāng)提高repl_baklog的大小，發(fā)現(xiàn)slave宕機時盡快實現(xiàn)故障恢復(fù)，盡可能避免全量同步
• 限制一個master上的slave節(jié)點數(shù)量，如果實在是太多slave，則可以采用主-從-從鏈?zhǔn)浇Y(jié)構(gòu)，減少master壓力

主從從架構(gòu)圖：

主從關(guān)系實現(xiàn)

現(xiàn)在三個實例還沒有任何關(guān)系，要配置主從可以使用replicaof 或者slaveof（5.0以前）命令。

有臨時和永久兩種模式：

• 修改從結(jié)點配置文件（永久生效）
  官方配置文件默認(rèn)注釋
  

  • 在redis.conf中添加一行配置：slaveof <masterip> <masterport>
  • 有密碼的需要配置密碼

slaveof 192.168.249.132 7001
masterauth 222222

• 使用redis-cli客戶端連接到redis服務(wù)，執(zhí)行slaveof命令（重啟后失效）：

  slaveof <masterip> <masterport>
  

這里有密碼 所以配置主從模式的信息 7002,7003配置

刪除舊容器,按照配置文件在啟動一遍

關(guān)閉任一容器查看日志 修改配置成功變回了redis默認(rèn)的RDB

進(jìn)入主節(jié)點容器7001 使用

# 查看狀態(tài)
info replication

發(fā)現(xiàn)只有master一個結(jié)點,這里需要配置

docker 容器進(jìn)入同一網(wǎng)絡(luò)

因為docker 容器是互相獨立的是,很多時候網(wǎng)絡(luò)不在一起無法鏈接刪除三個容器,先配置在同一docker網(wǎng)絡(luò)中

1. 創(chuàng)建網(wǎng)絡(luò)

docker network create my_redis_network

2.運行時候指定docker網(wǎng)絡(luò) 三個容器使用同一個網(wǎng)絡(luò)

network=xxxx
改進(jìn)后

docker run -d -p 7001:6379   -v /home/hadoop/Redis-cluster/7001/redis-data-7001/:/data   -v /home/hadoop/Redis-cluster/7001/redis.config:/data/redis.conf     --network my_redis_network     --name redis7001 redis redis-server /data/redis.conf


 docker run -d -p 7002:7002   -v /home/hadoop/Redis-cluster/7002/redis-data-7002/:/data   -v /home/hadoop/Redis-cluster/7002/redis.config:/data/redis.conf     --network my_redis_network     --name redis7002 redis redis-server /data/redis.conf


docker run -d -p 7003:6379   -v /home/hadoop/Redis-cluster/7003/redis-data-7003/:/data   -v /home/hadoop/Redis-cluster/7003/redis.config:/data/redis.conf     --network my_redis_network     --name redis7003 redis redis-server /data/redis.conf

3. 三個容器運行時候都拉入同一網(wǎng)絡(luò),查看從結(jié)點日志
   
   這個時候不在拒絕而是successful ,進(jìn)入主節(jié)點容器后查看結(jié)點狀態(tài)
4. docker exec -it redis7001 redis-cli進(jìn)入容器
5. 登錄redis后查看狀態(tài)
   輸入INFO
   此時就可以看到完整的主從結(jié)構(gòu)
   
   主結(jié)點寫一個數(shù)據(jù)
   
   從結(jié)點可以讀取出來
   
   并且無法在進(jìn)行寫數(shù)據(jù)了
   
   所以默認(rèn)主從結(jié)點完成讀寫分離,從結(jié)點失去寫入功能,從節(jié)點作為數(shù)據(jù)讀取的副本

這里采用指令鏈接,但是后續(xù)搭配哨兵機制以后修改為配置文件鏈接,因為測試哨兵機制需要多次重啟,并且由主從結(jié)構(gòu)的日志截圖可以看出,同一網(wǎng)絡(luò)下運行的docker容器交互采用的是容器內(nèi)的端口,為了防止后期哨兵集群部署后,結(jié)點數(shù)據(jù)量大,容器內(nèi)端口沖突等,建議容器內(nèi)端口修改為和映射宿主機端口一致

這樣避免一些為網(wǎng)絡(luò)問題

docker運行redis主從模式總結(jié)

1.首先配置創(chuàng)建對應(yīng)的redis容器的數(shù)據(jù)卷和配置文件,方便掛載
2. 數(shù)據(jù)備份采用默認(rèn)的RDB
3. 創(chuàng)建一個docker 網(wǎng)絡(luò) 運行時候保證在同一網(wǎng)絡(luò)
4.子實列配置文件寫清楚主節(jié)點信息

slaveof 主結(jié)點ip 端口
masterauth 密碼

5. docker 運行時指定network

2.4.小結(jié)

簡述全量同步和增量同步區(qū)別？

• 全量同步：master將完整內(nèi)存數(shù)據(jù)生成RDB，發(fā)送RDB到slave。后續(xù)命令則記錄在repl_baklog，逐個發(fā)送給slave。
• 增量同步：slave提交自己的offset到master，master獲取repl_baklog中從offset之后的命令給slave

什么時候執(zhí)行全量同步？

• slave節(jié)點第一次連接master節(jié)點時
• slave節(jié)點斷開時間太久，repl_baklog中的offset已經(jīng)被覆蓋時

什么時候執(zhí)行增量同步？

• slave節(jié)點斷開又恢復(fù)，并且在repl_baklog中能找到offset時

3.Redis哨兵

微服務(wù)中有專門的安全保護(hù)框架seatenl來對整個分布式進(jìn)行保護(hù),redis集群躲起來以后,同樣的也內(nèi)置了redis緩存集群監(jiān)控保護(hù)機制
Redis提供了哨兵（Sentinel）機制來實現(xiàn)主從集群的自動故障恢復(fù)。

3.1.哨兵原理

3.1.1.集群結(jié)構(gòu)和作用

哨兵的結(jié)構(gòu)如圖：

哨兵的作用如下：

• 監(jiān)控：Sentinel 會不斷檢查您的master和slave是否按預(yù)期工作
• 自動故障恢復(fù)：如果master故障，Sentinel會將一個slave提升為master。當(dāng)故障實例恢復(fù)后也以新的master為主
• 通知：Sentinel充當(dāng)Redis客戶端的服務(wù)發(fā)現(xiàn)來源，當(dāng)集群發(fā)生故障轉(zhuǎn)移時，會將最新信息推送給Redis的客戶端

3.1.2.集群監(jiān)控原理

Sentinel基于心跳機制監(jiān)測服務(wù)狀態(tài)，每隔1秒向集群的每個實例發(fā)送ping命令：

?主觀下線：如果某sentinel節(jié)點發(fā)現(xiàn)某實例未在規(guī)定時間響應(yīng)，則認(rèn)為該實例主觀下線。

?客觀下線：若超過指定數(shù)量（quorum）的sentinel都認(rèn)為該實例主觀下線，則該實例客觀下線。quorum值最好超過Sentinel實例數(shù)量的一半。

3.1.3.集群故障恢復(fù)原理

一旦發(fā)現(xiàn)master故障，sentinel需要在salve中選擇一個作為新的master，選擇依據(jù)是這樣的：

• 首先會判斷slave節(jié)點與master節(jié)點斷開時間長短，如果超過指定值（down-after-milliseconds * 10）則會排除該slave節(jié)點
• 然后判斷slave節(jié)點的slave-priority值，越小優(yōu)先級越高，如果是0則永不參與選舉
• 如果slave-prority一樣，則判斷slave節(jié)點的offset值，越大(偏移量越大,主從數(shù)據(jù)越接近)說明數(shù)據(jù)越新，優(yōu)先級越高
• 最后是判斷slave節(jié)點的運行id大小，越小優(yōu)先級越高。

當(dāng)選出一個新的master后，該如何實現(xiàn)切換呢？

流程如下：
這里假設(shè)master7001出現(xiàn)故障

• sentinel給備選的slave1節(jié)點發(fā)送slaveof no one命令，讓該節(jié)點成為master
• sentinel給所有其它slave發(fā)送slaveof 192.168.150.101 7002 命令，讓這些slave成為新master的從節(jié)點，開始從新的master上同步數(shù)據(jù)。
• 最后，sentinel將故障節(jié)點標(biāo)記為slave，當(dāng)故障節(jié)點恢復(fù)后會自動成為新的master的slave節(jié)點

3.1.4.小結(jié)

Sentinel的三個作用是什么？

• 監(jiān)控
• 故障轉(zhuǎn)移
• 通知

Sentinel如何判斷一個redis實例是否健康？

• 每隔1秒發(fā)送一次ping命令，如果超過一定時間沒有相向則認(rèn)為是主觀下線(和nacos一樣的心跳機制)
• 如果大多數(shù)sentinel都認(rèn)為實例主觀下線，則判定服務(wù)下線

故障轉(zhuǎn)移步驟有哪些？

• 首先選定一個slave作為新的master，執(zhí)行slaveof no one
• 然后讓所有節(jié)點都執(zhí)行slaveof 新master
• 修改故障節(jié)點配置，添加slaveof 新master

3.2.搭建哨兵集群 (docker實現(xiàn))

依舊采用docker 實現(xiàn),相比于直接部署服務(wù)器,docker 環(huán)境隔離,輕量的優(yōu)點可太爽了

1. 創(chuàng)建Docker網(wǎng)絡(luò)：
   首先確保所有的Redis實例和哨兵實例都在同一個網(wǎng)絡(luò)中：

docker network create redis-net  #之前的redis 主從集群就已經(jīng)運行到一個網(wǎng)絡(luò)了

2. 啟動Redis容器：
   您可以啟動多個Redis容器實例。假設(shè)您啟動了三個：

docker run -d --name redis1 --net redis-net redis
docker run -d --name redis2 --net redis-net redis
docker run -d --name redis3 --net redis-net redis

這里之前就已經(jīng)創(chuàng)建并且啟動了三個容器

3. 創(chuàng)建哨兵配置文件：
   新建三個目錄存放哨兵文件,以及后續(xù)的docker中哨兵工作目錄的掛載
   s1，s2,s3
   

創(chuàng)建一個基礎(chǔ)的sentinel.conf文件：

port 27001 #端口
sentinel monitor mymaster 192.168.249.132 7001 2
# 設(shè)置主服務(wù)器結(jié)點密碼
sentinel auth-pass mymaster 222222
sentinel down-after-milliseconds mymaster 5000 #超時時間默認(rèn)
sentinel failover-timeout mymaster 60000

解讀：

• port 27001：是當(dāng)前sentinel實例的端口
• sentinel monitor mymaster 192.168.249.132 7001 2：指定主節(jié)點信息 (通過主節(jié)點就可以檢測到從結(jié)點,所以redis主從架構(gòu)中,主節(jié)點可以代表一個集群)
  • mymaster：主節(jié)點名稱，自定義，任意寫
  • 192.168.249.132 7001 ：主節(jié)點的ip和端口
  • 2：選舉master時的quorum(這里三個哨兵 超過2就檢查不到心跳ping就客觀下下線)值
  • 如果不i是docker的話 工作目錄實在這個配置文件中指定的目錄
    dir “/tmp/s1”
    s1新建一個如上配置
    

4. 把配置文件復(fù)制到其他哨兵工作目錄：

然后修改端口,端口需要修改的,雖然docker 可以映射到宿主機的端口,但是docker網(wǎng)絡(luò)內(nèi)通信,使用的是docker 容器端口

5. 啟動哨兵容器：
   啟動前先查看docker 網(wǎng)絡(luò),要確定哨兵和容器能夠在同一網(wǎng)絡(luò),確保容器通行

 docker network ls

就像之前的主從結(jié)構(gòu)一樣,docker容器之間通信需要保證同一網(wǎng)絡(luò),并且在通信端口是容器內(nèi)端口
可以申明時定義同一網(wǎng)絡(luò) 端口和ip
使用這兩個參數(shù)后，從節(jié)點發(fā)送給主節(jié)點的ip和端口信息就是這里設(shè)定好了。
實列在其他電腦,在Docker哨兵的配置中將sentinel announce-ip設(shè)置為遠(yuǎn)程計算機上Redis節(jié)點的IP地址。這樣，Docker哨兵將正確地宣告Redis節(jié)點的位置給其他哨兵和Redis客戶端，以便它們知道如何連接到Redis節(jié)點。

replica-announce-ip 5.5.5.5
replica-announce-port 1234

哨兵的配置

sentinel announce-ip <ip>
sentinel announce-port <port>

sentinel announce-ip ：此選項允許您指定Redis Sentinel應(yīng)向其他Sentinel實例宣告其監(jiān)視的Redis實例的IP地址。您應(yīng)該將其設(shè)置為Redis實例所在主機的IP地址。這在Redis實例的實際IP地址與其他Redis Sentinel實例和客戶端應(yīng)用于連接的IP地址不同時非常有用。例如，如果Redis運行在Docker容器中，您可能希望宣告主機機器的IP地址。

sentinel announce-port ：此選項指定Redis Sentinel應(yīng)向其他Sentinel實例宣告的Redis實例的端口。您應(yīng)該將其設(shè)置為Redis實例實際監(jiān)聽的端口。

如果出現(xiàn)同一dockker網(wǎng)絡(luò)下無法通信就要配置上面的試試

每個哨兵都需要有它自己的配置文件，并掛載到對應(yīng)的工作目錄：

 docker run -d --name sentinel1 --net my_redis_network -p 27001:27001 -v /home/hadoop/Redis-cluster/s1:/data -v /home/hadoop/Redis-cluster/s1/sentinel.conf:/etc/sentinel.conf redis redis-sentinel /etc/sentinel.conf

 docker run -d --name sentinel2 --net my_redis_network -p 27002:27002 -v /home/hadoop/Redis-cluster/s2:/data -v /home/hadoop/Redis-cluster/s2/sentinel.conf:/etc/sentinel.conf redis redis-sentinel /etc/sentinel.conf

docker run -d --name sentinel3 --net my_redis_network -p 27003:27003 -v /home/hadoop/Redis-cluster/s3:/data -v /home/hadoop/Redis-cluster/s3/sentinel.conf:/etc/sentinel.conf redis redis-sentinel /etc/sentinel.conf

docker ps檢查運行狀態(tài)

現(xiàn)在，哨兵容器會使用s1，s2和s3這三個目錄作為它們的工作目錄(突然發(fā)現(xiàn)好多容器的工作目錄都是/data 配置文件目錄是/etc)。當(dāng)哨兵進(jìn)行某些操作時，您可以查看這些目錄來觀察它們的狀態(tài)。
6. 測試您的哨兵集群：
停止主節(jié)點，并觀察哨兵是否將其他節(jié)點提升為新的主節(jié)點：

docker stop redis7001

但是發(fā)現(xiàn)并沒有切換主節(jié)點 并且發(fā)現(xiàn)日志

Could not rename tmp config file (Device or resource busy)
WARNING: Sentinel was not able to save the new configuration on disk!!!: Device or resource busy

細(xì)節(jié)

因為Sentinel會在啟動后向自己的配置文件中追加內(nèi)容，它采用的是先創(chuàng)建一個臨時配置文件，然后使用它替換掉原來的配置文件的方式。

如果是使用掛載卷直接掛載文件的方式，docker貌似不允許這樣操作，所以會出現(xiàn)這個錯誤，你可以將配置文件放到單獨的目錄中，然后將目錄掛載到容器。

所以三個哨兵目錄創(chuàng)建config進(jìn)行配置目錄掛載,而不是文件
每個哨兵文件 新建配置目錄config,并把之前的配置文件放進(jìn)入

進(jìn)入整個目錄的掛載

docker run -d --name sentinel1 --net my_redis_network -p 27001:27001 -v /home/hadoop/Redis-cluster/s1/data/:/data -v /home/hadoop/Redis-cluster/s1/config/:/etc/ redis redis-sentinel /etc/sentinel.conf

 docker run -d --name sentinel2 --net my_redis_network -p 27002:27002 -v /home/hadoop/Redis-cluster/s2/data/:/data -v /home/hadoop/Redis-cluster/s2/config/:/etc/ redis redis-sentinel /etc/sentinel.conf


docker run -d --name sentinel3 --net my_redis_network -p 27003:27003 -v /home/hadoop/Redis-cluster/s3/data:/data -v /home/hadoop/Redis-cluster/s3/config/:/etc/ redis redis-sentinel /etc/sentinel.conf

此時查看任一哨兵日志

不在出現(xiàn)警告無法寫入磁盤
并且進(jìn)入任一哨兵容器

docker exec -it 哨兵容器 redis-cli -p 哨兵端口

查看監(jiān)視該集群的哨兵由那些,就能知道是否能通信了

 SENTINEL sentinels <master-name>

發(fā)現(xiàn)可以監(jiān)控

日志哨兵日志

發(fā)現(xiàn)成功監(jiān)控并且知道從主從信息已經(jīng)同一監(jiān)控
現(xiàn)在測試停止監(jiān)控的redis集群 7001

輸出哨兵日志,哨兵監(jiān)控到結(jié)點服務(wù)異常

最重要的一條

發(fā)現(xiàn)選擇哨兵選擇了7003結(jié)點代替原來的主節(jié)點,進(jìn)入7003容器

docker exec -it redis7003 redis-cli -p 7003

輸入info

此時7003成為主要結(jié)點,說明哨兵的功能起到作用,防止業(yè)務(wù)癱瘓,監(jiān)視集群并且替換主節(jié)點

3.2.1 哨兵集群結(jié)構(gòu)

3.3.RedisTemplate

在Sentinel集群監(jiān)管下的Redis主從集群，其節(jié)點會因為自動故障轉(zhuǎn)移而發(fā)生變化，Redis的客戶端必須感知這種變化，及時更新連接信息。Spring的RedisTemplate底層利用lettuce實現(xiàn)了節(jié)點的感知和自動切換。

下面，我們通過一個測試來實現(xiàn)RedisTemplate集成哨兵機制。

假如一個一般springboot項目
主要就以下一個controller 通過restful接口 風(fēng)格操作redis

@RestController
public class HelloController {

    @Autowired
    private StringRedisTemplate redisTemplate;

    @GetMapping("/get/{key}")
    public String hi(@PathVariable String key) {
        return redisTemplate.opsForValue().get(key);
    }

//    接口操作數(shù)據(jù)庫
    @GetMapping("/set/{key}/{value}")
    public String hi(@PathVariable String key, @PathVariable String value) {
        redisTemplate.opsForValue().set(key, value);
        return "success";
    }
}

3.3.2.引入依賴

在項目的pom文件中引入依賴：

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

3.3.3.配置Redis地址

然后在配置文件application.yml中指定redis的sentinel相關(guān)信息：
在sentinel 模式下,主從模式的集群有可能會隨著業(yè)務(wù)變更,所以只需要配置監(jiān)控者,就可以得到被監(jiān)控的信息

spring:
  redis:
    sentinel:
      master: mymaster
      nodes:
        - 192.168.249.132:27001
        - 192.168.249.132:27002
        - 192.168.249.132:27003
    password: 222222

我redis密碼集群吧都是222222

3.3.4.配置讀寫分離

在項目的啟動類中，添加一個新的bean：

Lettuce 是實現(xiàn)redisTemplate客戶端的底層實現(xiàn)框架

@Bean
public LettuceClientConfigurationBuilderCustomizer clientConfigurationBuilderCustomizer(){
//接口不能new 而只有一個方法明顯是函數(shù)接口
    return clientConfigurationBuilder -> clientConfigurationBuilder.readFrom(ReadFrom.REPLICA_PREFERRED);
}

主節(jié)點設(shè)置數(shù)據(jù)

測試游覽器輸入對應(yīng)api 路徑

游覽器操作redis設(shè)置數(shù)據(jù)

因為controller 返回的

在redis 圖形化庫查看

說明resttemplate都能實現(xiàn),并且日志也很詳細(xì)
從集群中的7001端口結(jié)點讀取

從7003端口的結(jié)點寫入

這個時候手動掛機主節(jié)點看看日志‘

哨兵是已經(jīng)發(fā)先了,并且switch 7002作為新的主結(jié)點
而java客戶端的日志是不斷的發(fā)同步數(shù)據(jù)

最后確定新的主從結(jié)構(gòu)
檢查新的現(xiàn)象 。。。。詳細(xì)信息

這個bean中配置的就是讀寫策略，包括四種：

• MASTER：從主節(jié)點讀取
• MASTER_PREFERRED：優(yōu)先從master節(jié)點讀取，master不可用才讀取replica
• REPLICA：從slave（replica）節(jié)點讀取
• REPLICA _PREFERRED：優(yōu)先從slave（replica）節(jié)點讀取，所有的slave都不可用才讀取master

4.Redis分片集群

4.1.搭建分片集群

主從和哨兵可以解決高可用、高并發(fā)讀的問題。但是依然有兩個問題沒有解決：

• 海量數(shù)據(jù)存儲問題

• 高并發(fā)寫的問題

基本只要可以存儲數(shù)據(jù)的分布式中間件都是采用這種方式,比如hadfs使用分片集群可以解決上述問題，如圖:

分布式場景建議單結(jié)點的redis不應(yīng)該給予太多內(nèi)存資源,數(shù)據(jù)輸入量大無論是RDB,AOF，數(shù)據(jù)備份占用的內(nèi)存都是很高的,大量的io會給架構(gòu)帶來壓力

分片集群特征：

• 集群中有多個master，每個master保存不同數(shù)據(jù),保證存儲大量數(shù)據(jù)

• 每個master都可以有多個slave節(jié)點,讀寫分離,擴大存儲量的同時還不會降低讀的能力

• master之間通過ping監(jiān)測彼此健康狀態(tài),這樣就代表不在需要哨兵機制,集群間相互通信,自動檢查集群內(nèi)存狀態(tài)

• 客戶端請求可以訪問集群任意節(jié)點，最終都會被轉(zhuǎn)發(fā)到正確節(jié)點

4.1.0 docker實現(xiàn)redis分片集群

因為我的虛擬機內(nèi)存大小優(yōu)先

1. 刪除之前的7001、7002、7003這幾個目錄，重新創(chuàng)建出7001、7002、7003、8001、8002、8003目錄：
   
   每個文件的結(jié)構(gòu)
   

2. 新建一個配置文件 通用的 redis.conf

#端口
port 7001
#2、修改bind或注釋掉
bind 0.0.0.0
#3、保護(hù)模式關(guān)閉，因為用docker否則無法訪問
protected-mode no
#4、關(guān)閉后臺運行
daemonize no
#5、設(shè)置密碼
#requirepass myredis
#6、配置與主節(jié)點驗證的密碼
#masterauth 設(shè)置密碼
#7、開啟aof日志
appendonly yes
#8、開啟集群模式
cluster-enabled yes
#9、根據(jù)你啟用的節(jié)點來命名，最好和端口保持一致，這個是用來保存其他節(jié)點的名稱，狀態(tài)等信息的
cluster-config-file nodes_7001.conf
#10、超時時間
cluster-node-timeout 5000
#11、集群節(jié)點 IP：服務(wù)器就填公網(wǎng)ip，或者內(nèi)部對應(yīng)容器的ip 
cluster-announce-ip 192.168.249.132

#12、集群節(jié)點映射端口
cluster-announce-port 7001
#13、總線監(jiān)控ip默認(rèn)端口+10000 如 7001 就是 17001
cluster-announce-bus-port 17001

每個Redis集群中的節(jié)點都需要打開兩個TCP連接。用于客戶端提供服務(wù)，比如6379，還有一個額外的端口（通過在這個端口號上加10000）作為數(shù)據(jù)端口，例如：redis的端口為6379，那么另外一個需要開通的端口是：6379 + 10000， 即需要開啟 16379。16379端口用于集群總線，這是一個用二進(jìn)制協(xié)議的點對點通信信道。這個集群總線（Cluster bus）用于節(jié)點的失敗偵測 ,如果不配置該項,很可能出現(xiàn)創(chuàng)建集群中一直waiting 線程堵塞

復(fù)制到每個文件夾,按照自己的需求進(jìn)行更改

運行

 docker run -d -p 7001:7001 -p 17001:17001  -v /home/hadoop/Redis-cluster/7001/data/:/data/   -v /home/hadoop/Redis-cluster/7001/conf/redis.conf:/data/redis.conf  --network my_redis_network   --name redis7001   redis redis-server /data/redis.conf 
   docker run -d -p 7002:7002 -p 17002:17002  -v /home/hadoop/Redis-cluster/7002/data/:/data/   -v /home/hadoop/Redis-cluster/7002/conf/redis.conf:/data/redis.conf  --network my_redis_network   --name redis7002   redis redis-server /data/redis.conf 
   docker run -d -p 7003:7003  -p 17003:17003 -v /home/hadoop/Redis-cluster/7003/data/:/data/   -v /home/hadoop/Redis-cluster/7003/conf/redis.conf:/data/redis.conf  --network my_redis_network   --name redis7003   redis redis-server /data/redis.conf 
   docker run -d -p 8001:8001 -p 18001:18001  -v /home/hadoop/Redis-cluster/8001/data/:/data/   -v /home/hadoop/Redis-cluster/8001/conf/redis.conf:/data/redis.conf  --network my_redis_network   --name redis8001  redis redis-server /data/redis.conf 
   docker run -d -p 8002:8002  -p 18002:18002  -v /home/hadoop/Redis-cluster/8002/data/:/data/   -v /home/hadoop/Redis-cluster/8002/conf/redis.conf:/data/redis.conf  --network my_redis_network   --name redis8002  redis redis-server /data/redis.conf 
   docker run -d -p 8003:8003  -p 18003:18003 -v /home/hadoop/Redis-cluster/8003/data/:/data/   -v /home/hadoop/Redis-cluster/8003/conf/redis.conf:/data/redis.conf  --network my_redis_network   --name redis8003  redis redis-server /data/redis.conf 

容器運行成功

）構(gòu)建集群

我們使用的是Redis6.2.4版本，集群管理以及集成到了redis-cli中，格式如下：

redis-cli --cluster create --cluster-replicas 1 192.168.249.132:7001 192.168.249.132:7002 192.168.249.132:7003 192.168.249.132:8001 192.168.249.132:8002 192.168.249.132:8003

• redis-cli --cluster或者./redis-trib.rb：代表集群操作命令
• create：代表是創(chuàng)建集群
• --replicas 1或者--cluster-replicas 1 ：指定集群中每個master的副本個數(shù)為1，此時節(jié)點總數(shù) ÷ (replicas + 1) 得到的就是master的數(shù)量。因此節(jié)點列表中的前n個就是master，其它節(jié)點都是slave節(jié)點，隨機分配到不同master
  因為docker的部署的redis,redis-cli 在容器內(nèi) 所以運行

  docker exec -it redis7001(任一結(jié)點)  redis-cli --cluster create --cluster-replicas 1 192.168.249.132:7001 192.168.249.132:7002 192.168.249.132:7003 192.168.249.132:8001 192.168.249.132:8002 192.168.249.132:8003

結(jié)構(gòu)輸出

redis集群正在分配插槽,然后詢問你是否滿意這樣的分配
回復(fù)yes后開始創(chuàng)建集群

如果出現(xiàn)上述截圖就是線程堵塞了,需要開啟線程總線配置,并且除了redis端口還要自己暴露總線結(jié)點

docker部署redis集群

修改后命令

 docker run -d -p 7001:7001 -p 17001:17001  -v /home/hadoop/Redis-cluster/7001/data/:/data/   -v /home/hadoop/Redis-cluster/7001/conf/redis.conf:/data/redis.conf  --network my_redis_network   --name redis7001   redis redis-server /data/redis.conf 
   docker run -d -p 7002:7002 -p 17002:17002  -v /home/hadoop/Redis-cluster/7002/data/:/data/   -v /home/hadoop/Redis-cluster/7002/conf/redis.conf:/data/redis.conf  --network my_redis_network   --name redis7002   redis redis-server /data/redis.conf 
   docker run -d -p 7003:7003  -p 17003:17003 -v /home/hadoop/Redis-cluster/7003/data/:/data/   -v /home/hadoop/Redis-cluster/7003/conf/redis.conf:/data/redis.conf  --network my_redis_network   --name redis7003   redis redis-server /data/redis.conf 
   docker run -d -p 8001:8001 -p 18001:18001  -v /home/hadoop/Redis-cluster/8001/data/:/data/   -v /home/hadoop/Redis-cluster/8001/conf/redis.conf:/data/redis.conf  --network my_redis_network   --name redis8001  redis redis-server /data/redis.conf 
   docker run -d -p 8002:8002  -p 18002:18002  -v /home/hadoop/Redis-cluster/8002/data/:/data/   -v /home/hadoop/Redis-cluster/8002/conf/redis.conf:/data/redis.conf  --network my_redis_network   --name redis8002  redis redis-server /data/redis.conf 
   docker run -d -p 8003:8003  -p 18003:18003 -v /home/hadoop/Redis-cluster/8003/data/:/data/   -v /home/hadoop/Redis-cluster/8003/conf/redis.conf:/data/redis.conf  --network my_redis_network   --name redis8003  redis redis-server /data/redis.conf 

此時運行成功

測試鏈接任一主節(jié)點
集群數(shù)據(jù)操作時，需要給redis-cli加上-c參數(shù)才可以：

redis-cli -c -p 7001

發(fā)現(xiàn)保存數(shù)據(jù)到該節(jié)點,最后卻轉(zhuǎn)發(fā)到另一個結(jié)點

集群中數(shù)據(jù)插入是根據(jù)集群算法來的,會根據(jù)轉(zhuǎn)發(fā)到這個插槽的結(jié)點

redis集群中保存數(shù)據(jù),會先對key進(jìn)行計算,然后保存到計算放置的插槽節(jié)點中,并且轉(zhuǎn)發(fā)到該結(jié)點,這樣取值的時候也會根據(jù)插槽
進(jìn)行轉(zhuǎn)發(fā)

查看結(jié)點情況

 docker exec -it redis7001 redis-cli -p 7001 cluster nodes

4.2.散列插槽

4.2.1.插槽原理

Redis會把每一個master節(jié)點映射到0~16383共16384個插槽（hash slot）上，查看集群信息時就能看到：

數(shù)據(jù)key不是與節(jié)點綁定，而是與插槽綁定。redis會根據(jù)key的有效部分計算插槽值，分兩種情況：

• key中包含"{}"，且“{}”中至少包含1個字符，“{}”中的部分是有效部分
• key中不包含“{}”，整個key都是有效部分

例如：key是num，那么就根據(jù)num計算，如果是{itcast}num，則根據(jù)itcast計算。計算方式是利用CRC16算法得到一個hash值，然后對16384取余，得到的結(jié)果就是slot值。

如圖，在7001這個節(jié)點執(zhí)行set a 1時，對a做hash運算，對16384取余，得到的結(jié)果是15495，因此要存儲到103節(jié)點。

到了7003后，執(zhí)行get num時，對num做hash運算，對16384取余，得到的結(jié)果是2765，因此需要切換到7001節(jié)點

4.2.1.小結(jié)

Redis如何判斷某個key應(yīng)該在哪個實例？

• 將16384個插槽分配到不同的實例
• 根據(jù)key的有效部分計算哈希值，對16384取余
• 余數(shù)作為插槽，尋找插槽所在實例即可

如何將同一類數(shù)據(jù)固定的保存在同一個Redis實例？(數(shù)據(jù)分類)

• 這一類數(shù)據(jù)使用相同的有效部分，例如key都以{typeId}為前綴

比如我這里對name 作為key 進(jìn)行分組,每個組的前綴不同作為分組

4.3.集群伸縮

redis-cli --cluster提供了很多操作集群的命令，可以通過下面方式查看：

比如，添加節(jié)點的命令：

4.3.1.需求分析

需求：向集群中添加一個新的master節(jié)點，并向其中存儲 num = 10

• 啟動一個新的redis實例，端口為7004
• 添加7004到之前的集群，并作為一個master節(jié)點
• 給7004節(jié)點分配插槽，使得num這個key可以存儲到7004實例

這里需要兩個新的功能：

• 添加一個節(jié)點到集群中
• 將部分插槽分配到新插槽

4.3.2.創(chuàng)建新的redis實例

創(chuàng)建一個文件夾：

mkdir 7004

拷貝之前配置文件：

cp redis.conf /7004

修改配置文件：

sed /s/6379/7004/g 7004/redis.conf

啟動

4.3.3.添加新節(jié)點到redis

添加節(jié)點的語法如下：

執(zhí)行命令：

docker exec -it redis7001 redis-cli --cluster add-node 192.168.249.132:7004  192.168.249.132:7001 

顯示ok 已經(jīng)成功
通過命令查看集群狀態(tài)：

redis-cli -p 7001 cluster nodes

如圖，7004加入了集群，并且默認(rèn)是一個master節(jié)點：

但是，可以看到7004節(jié)點的插槽數(shù)量為0，因此沒有任何數(shù)據(jù)可以存儲到7004上

4.3.4.轉(zhuǎn)移插槽

我們要將num存儲到7004節(jié)點，因此需要先看看num的插槽是多少：

如上圖所示，num的插槽為2765.

我們可以將0~3000的插槽從7001轉(zhuǎn)移到7004，命令格式如下：

具體命令如下：

建立連接：

得到下面的反饋：

詢問要移動多少個插槽，我們計劃是3000個：

復(fù)制這個id(nodes 命令中7004的id)，然后拷貝到剛才的控制臺后：

這里詢問，你的插槽是從哪里移動過來的？

• all：代表全部，也就是三個節(jié)點各轉(zhuǎn)移一部分
• 具體的id：目標(biāo)節(jié)點的id
• done：沒有了

這里我們要從7001獲取，因此填寫7001的id：可以從多結(jié)點分配插槽

填完后，點擊done，這樣插槽轉(zhuǎn)移就準(zhǔn)備好了：

確認(rèn)要轉(zhuǎn)移嗎？輸入yes：

日志不斷輸出轉(zhuǎn)移結(jié)果
然后，通過命令查看結(jié)果：

docker exec -it redis7001 redis-cli -p 7001 cluster nodes

可以看到：

目的達(dá)成。

4.4.故障轉(zhuǎn)移

現(xiàn)在集群狀態(tài)是這樣的：

其中7001、7002、7003,7004都是master，我們計劃讓7002宕機。

4.4.1.自動故障轉(zhuǎn)移

當(dāng)集群中有一個master宕機會發(fā)生什么呢？

直接停止一個redis實例，例如7002：

 docker stop redis7002
 

1）首先是該實例與其它實例失去連接

2）然后是疑似宕機：

3）最后是確定下線，自動提升一個slave為新的master：

4）當(dāng)7002再次啟動，就會變?yōu)橐粋€slave節(jié)點了：

這樣就通過各個主機之間的心跳檢查實現(xiàn)了哨兵的故障轉(zhuǎn)移機制

4.4.2.手動故障轉(zhuǎn)移

利用cluster failover命令可以手動讓集群中的某個master宕機，切換到執(zhí)行cluster failover命令的這個slave節(jié)點，實現(xiàn)無感知(一般是用于服務(wù)器硬件升級)的數(shù)據(jù)遷移。其流程如下：

這種failover命令可以指定三種模式：

• 缺省：默認(rèn)的流程，如圖1~6歩
• force：省略了對offset的一致性校驗
• takeover：直接執(zhí)行第5歩，忽略數(shù)據(jù)一致性、忽略master狀態(tài)和其它master的意見

比如我這個演示的集群手動把7002結(jié)點重新變?yōu)閙aster

 docker exec -it redis7002 redis-cli -p 7002

查找結(jié)點情況

docker exec -it  redis7001 redis-cli -p 7001 cluster nodes

7002再次成為master

4.5.RedisTemplate訪問分片集群

RedisTemplate底層同樣基于lettuce實現(xiàn)了分片集群的支持，所以使用的步驟與哨兵模式基本一致：

1）引入redis的starter依賴

2）配置分片集群地址

3）配置讀寫分離

與哨兵模式相比，其中只有分片集群的配置方式略有差異，如下：

哨兵

spring:
  redis:
    password: 222222
    sentinel:
      master: mymaster
      nodes:
        - 192.168.249.132:27001
        - 192.168.249.132:27002
        - 192.168.249.132:27003

集群

spring:
  redis:
    cluster:
      nodes:
        - 192.168.150.101:7001
        - 192.168.150.101:7002
        - 192.168.150.101:7003
        - 192.168.150.101:8001
        - 192.168.150.101:8002
        - 192.168.150.101:8003

測試
使用之前演示的restful操作

讀取成功

查看日志

發(fā)現(xiàn)主節(jié)點7002寫內(nèi)存
而從節(jié)點8002讀
說明實現(xiàn)了讀寫分離


并且保存結(jié)點是根據(jù)key計算來確定插槽位置的最后附上redis-cli
java客戶端的工具類文章來源地址http://www.zghlxwxcb.cn/news/detail-720499.html

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.annotation.CachingConfigurerSupport;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.DataType;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.StringRedisConnection;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.TimeUnit;

@Component
public class CacheService extends CachingConfigurerSupport {
    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    public StringRedisTemplate getstringRedisTemplate() {
        return this.stringRedisTemplate;
    }

    /** -------------------key相關(guān)操作--------------------- */

    /**
     * 刪除key
     *
     * @param key
     */
    public void delete(String key) {
        stringRedisTemplate.delete(key);
    }

    /**
     * 批量刪除key
     *
     * @param keys
     */
    public void delete(Collection<String> keys) {
        stringRedisTemplate.delete(keys);
    }

    /**
     * 序列化key
     *
     * @param key
     * @return
     */
    public byte[] dump(String key) {
        return stringRedisTemplate.dump(key);
    }

    /**
     * 是否存在key
     *
     * @param key
     * @return
     */
    public Boolean exists(String key) {
        return stringRedisTemplate.hasKey(key);
    }

    /**
     * 設(shè)置過期時間
     *
     * @param key
     * @param timeout
     * @param unit
     * @return
     */
    public Boolean expire(String key, long timeout, TimeUnit unit) {
        return stringRedisTemplate.expire(key, timeout, unit);
    }

    /**
     * 設(shè)置過期時間
     *
     * @param key
     * @param date
     * @return
     */
    public Boolean expireAt(String key, Date date) {
        return stringRedisTemplate.expireAt(key, date);
    }

    /**
     * 查找匹配的key
     *
     * @param pattern
     * @return
     */
    public Set<String> keys(String pattern) {
        return stringRedisTemplate.keys(pattern);
    }

    /**
     * 將當(dāng)前數(shù)據(jù)庫的 key 移動到給定的數(shù)據(jù)庫 db 當(dāng)中
     *
     * @param key
     * @param dbIndex
     * @return
     */
    public Boolean move(String key, int dbIndex) {
        return stringRedisTemplate.move(key, dbIndex);
    }

    /**
     * 移除 key 的過期時間，key 將持久保持
     *
     * @param key
     * @return
     */
    public Boolean persist(String key) {
        return stringRedisTemplate.persist(key);
    }

    /**
     * 返回 key 的剩余的過期時間
     *
     * @param key
     * @param unit
     * @return
     */
    public Long getExpire(String key, TimeUnit unit) {
        return stringRedisTemplate.getExpire(key, unit);
    }

    /**
     * 返回 key 的剩余的過期時間
     *
     * @param key
     * @return
     */
    public Long getExpire(String key) {
        return stringRedisTemplate.getExpire(key);
    }

    /**
     * 從當(dāng)前數(shù)據(jù)庫中隨機返回一個 key
     *
     * @return
     */
    public String randomKey() {
        return stringRedisTemplate.randomKey();
    }

    /**
     * 修改 key 的名稱
     *
     * @param oldKey
     * @param newKey
     */
    public void rename(String oldKey, String newKey) {
        stringRedisTemplate.rename(oldKey, newKey);
    }

    /**
     * 僅當(dāng) newkey 不存在時，將 oldKey 改名為 newkey
     *
     * @param oldKey
     * @param newKey
     * @return
     */
    public Boolean renameIfAbsent(String oldKey, String newKey) {
        return stringRedisTemplate.renameIfAbsent(oldKey, newKey);
    }

    /**
     * 返回 key 所儲存的值的類型
     *
     * @param key
     * @return
     */
    public DataType type(String key) {
        return stringRedisTemplate.type(key);
    }

    /** -------------------string相關(guān)操作--------------------- */

    /**
     * 設(shè)置指定 key 的值
     * @param key
     * @param value
     */
    public void set(String key, String value) {
        stringRedisTemplate.opsForValue().set(key, value);
    }

    /**
     * 獲取指定 key 的值
     * @param key
     * @return
     */
    public String get(String key) {
        return stringRedisTemplate.opsForValue().get(key);
    }

    /**
     * 返回 key 中字符串值的子字符
     * @param key
     * @param start
     * @param end
     * @return
     */
    public String getRange(String key, long start, long end) {
        return stringRedisTemplate.opsForValue().get(key, start, end);
    }

    /**
     * 將給定 key 的值設(shè)為 value ，并返回 key 的舊值(old value)
     *
     * @param key
     * @param value
     * @return
     */
    public String getAndSet(String key, String value) {
        return stringRedisTemplate.opsForValue().getAndSet(key, value);
    }

    /**
     * 對 key 所儲存的字符串值，獲取指定偏移量上的位(bit)
     *
     * @param key
     * @param offset
     * @return
     */
    public Boolean getBit(String key, long offset) {
        return stringRedisTemplate.opsForValue().getBit(key, offset);
    }

    /**
     * 批量獲取
     *
     * @param keys
     * @return
     */
    public List<String> multiGet(Collection<String> keys) {
        return stringRedisTemplate.opsForValue().multiGet(keys);
    }

    /**
     * 設(shè)置ASCII碼, 字符串'a'的ASCII碼是97, 轉(zhuǎn)為二進(jìn)制是'01100001', 此方法是將二進(jìn)制第offset位值變?yōu)関alue
     *
     * @param key
     * @param
     * @param value
     *            值,true為1, false為0
     * @return
     */
    public boolean setBit(String key, long offset, boolean value) {
        return stringRedisTemplate.opsForValue().setBit(key, offset, value);
    }

    /**
     * 將值 value 關(guān)聯(lián)到 key ，并將 key 的過期時間設(shè)為 timeout
     *
     * @param key
     * @param value
     * @param timeout
     *            過期時間
     * @param unit
     *            時間單位, 天:TimeUnit.DAYS 小時:TimeUnit.HOURS 分鐘:TimeUnit.MINUTES
     *            秒:TimeUnit.SECONDS 毫秒:TimeUnit.MILLISECONDS
     */
    public void setEx(String key, String value, long timeout, TimeUnit unit) {
        stringRedisTemplate.opsForValue().set(key, value, timeout, unit);
    }

    /**
     * 只有在 key 不存在時設(shè)置 key 的值
     *
     * @param key
     * @param value
     * @return 之前已經(jīng)存在返回false,不存在返回true
     */
    public boolean setIfAbsent(String key, String value) {
        return stringRedisTemplate.opsForValue().setIfAbsent(key, value);
    }

    /**
     * 用 value 參數(shù)覆寫給定 key 所儲存的字符串值，從偏移量 offset 開始
     *
     * @param key
     * @param value
     * @param offset
     *            從指定位置開始覆寫
     */
    public void setRange(String key, String value, long offset) {
        stringRedisTemplate.opsForValue().set(key, value, offset);
    }

    /**
     * 獲取字符串的長度
     *
     * @param key
     * @return
     */
    public Long size(String key) {
        return stringRedisTemplate.opsForValue().size(key);
    }

    /**
     * 批量添加
     *
     * @param maps
     */
    public void multiSet(Map<String, String> maps) {
        stringRedisTemplate.opsForValue().multiSet(maps);
    }

    /**
     * 同時設(shè)置一個或多個 key-value 對，當(dāng)且僅當(dāng)所有給定 key 都不存在
     *
     * @param maps
     * @return 之前已經(jīng)存在返回false,不存在返回true
     */
    public boolean multiSetIfAbsent(Map<String, String> maps) {
        return stringRedisTemplate.opsForValue().multiSetIfAbsent(maps);
    }

    /**
     * 增加(自增長), 負(fù)數(shù)則為自減
     *
     * @param key
     * @param
     * @return
     */
    public Long incrBy(String key, long increment) {
        return stringRedisTemplate.opsForValue().increment(key, increment);
    }

    /**
     *
     * @param key
     * @param
     * @return
     */
    public Double incrByFloat(String key, double increment) {
        return stringRedisTemplate.opsForValue().increment(key, increment);
    }

    /**
     * 追加到末尾
     *
     * @param key
     * @param value
     * @return
     */
    public Integer append(String key, String value) {
        return stringRedisTemplate.opsForValue().append(key, value);
    }

    /** -------------------hash相關(guān)操作------------------------- */

    /**
     * 獲取存儲在哈希表中指定字段的值
     *
     * @param key
     * @param field
     * @return
     */
    public Object hGet(String key, String field) {
        return stringRedisTemplate.opsForHash().get(key, field);
    }

    /**
     * 獲取所有給定字段的值
     *
     * @param key
     * @return
     */
    public Map<Object, Object> hGetAll(String key) {
        return stringRedisTemplate.opsForHash().entries(key);
    }

    /**
     * 獲取所有給定字段的值
     *
     * @param key
     * @param fields
     * @return
     */
    public List<Object> hMultiGet(String key, Collection<Object> fields) {
        return stringRedisTemplate.opsForHash().multiGet(key, fields);
    }

    public void hPut(String key, String hashKey, String value) {
        stringRedisTemplate.opsForHash().put(key, hashKey, value);
    }

    public void hPutAll(String key, Map<String, String> maps) {
        stringRedisTemplate.opsForHash().putAll(key, maps);
    }

    /**
     * 僅當(dāng)hashKey不存在時才設(shè)置
     *
     * @param key
     * @param hashKey
     * @param value
     * @return
     */
    public Boolean hPutIfAbsent(String key, String hashKey, String value) {
        return stringRedisTemplate.opsForHash().putIfAbsent(key, hashKey, value);
    }

    /**
     * 刪除一個或多個哈希表字段
     *
     * @param key
     * @param fields
     * @return
     */
    public Long hDelete(String key, Object... fields) {
        return stringRedisTemplate.opsForHash().delete(key, fields);
    }

    /**
     * 查看哈希表 key 中，指定的字段是否存在
     *
     * @param key
     * @param field
     * @return
     */
    public boolean hExists(String key, String field) {
        return stringRedisTemplate.opsForHash().hasKey(key, field);
    }

    /**
     * 為哈希表 key 中的指定字段的整數(shù)值加上增量 increment
     *
     * @param key
     * @param field
     * @param increment
     * @return
     */
    public Long hIncrBy(String key, Object field, long increment) {
        return stringRedisTemplate.opsForHash().increment(key, field, increment);
    }

    /**
     * 為哈希表 key 中的指定字段的整數(shù)值加上增量 increment
     *
     * @param key
     * @param field
     * @param delta
     * @return
     */
    public Double hIncrByFloat(String key, Object field, double delta) {
        return stringRedisTemplate.opsForHash().increment(key, field, delta);
    }

    /**
     * 獲取所有哈希表中的字段
     *
     * @param key
     * @return
     */
    public Set<Object> hKeys(String key) {
        return stringRedisTemplate.opsForHash().keys(key);
    }

    /**
     * 獲取哈希表中字段的數(shù)量
     *
     * @param key
     * @return
     */
    public Long hSize(String key) {
        return stringRedisTemplate.opsForHash().size(key);
    }

    /**
     * 獲取哈希表中所有值
     *
     * @param key
     * @return
     */
    public List<Object> hValues(String key) {
        return stringRedisTemplate.opsForHash().values(key);
    }

    /**
     * 迭代哈希表中的鍵值對
     *
     * @param key
     * @param options
     * @return
     */
    public Cursor<Map.Entry<Object, Object>> hScan(String key, ScanOptions options) {
        return stringRedisTemplate.opsForHash().scan(key, options);
    }

    /** ------------------------list相關(guān)操作---------------------------- */

    /**
     * 通過索引獲取列表中的元素
     *
     * @param key
     * @param index
     * @return
     */
    public String lIndex(String key, long index) {
        return stringRedisTemplate.opsForList().index(key, index);
    }

    /**
     * 獲取列表指定范圍內(nèi)的元素
     *
     * @param key
     * @param start
     *            開始位置, 0是開始位置
     * @param end
     *            結(jié)束位置, -1返回所有
     * @return
     */
    public List<String> lRange(String key, long start, long end) {
        return stringRedisTemplate.opsForList().range(key, start, end);
    }

    /**
     * 存儲在list頭部
     *
     * @param key
     * @param value
     * @return
     */
    public Long lLeftPush(String key, String value) {
        return stringRedisTemplate.opsForList().leftPush(key, value);
    }

    /**
     *
     * @param key
     * @param value
     * @return
     */
    public Long lLeftPushAll(String key, String... value) {
        return stringRedisTemplate.opsForList().leftPushAll(key, value);
    }

    /**
     *
     * @param key
     * @param value
     * @return
     */
    public Long lLeftPushAll(String key, Collection<String> value) {
        return stringRedisTemplate.opsForList().leftPushAll(key, value);
    }

    /**
     * 當(dāng)list存在的時候才加入
     *
     * @param key
     * @param value
     * @return
     */
    public Long lLeftPushIfPresent(String key, String value) {
        return stringRedisTemplate.opsForList().leftPushIfPresent(key, value);
    }

    /**
     * 如果pivot存在,再pivot前面添加
     *
     * @param key
     * @param pivot
     * @param value
     * @return
     */
    public Long lLeftPush(String key, String pivot, String value) {
        return stringRedisTemplate.opsForList().leftPush(key, pivot, value);
    }

    /**
     *
     * @param key
     * @param value
     * @return
     */
    public Long lRightPush(String key, String value) {
        return stringRedisTemplate.opsForList().rightPush(key, value);
    }

    /**
     *
     * @param key
     * @param value
     * @return
     */
    public Long lRightPushAll(String key, String... value) {
        return stringRedisTemplate.opsForList().rightPushAll(key, value);
    }

    /**
     *
     * @param key
     * @param value
     * @return
     */
    public Long lRightPushAll(String key, Collection<String> value) {
        return stringRedisTemplate.opsForList().rightPushAll(key, value);
    }

    /**
     * 為已存在的列表添加值
     *
     * @param key
     * @param value
     * @return
     */
    public Long lRightPushIfPresent(String key, String value) {
        return stringRedisTemplate.opsForList().rightPushIfPresent(key, value);
    }

    /**
     * 在pivot元素的右邊添加值
     *
     * @param key
     * @param pivot
     * @param value
     * @return
     */
    public Long lRightPush(String key, String pivot, String value) {
        return stringRedisTemplate.opsForList().rightPush(key, pivot, value);
    }

    /**
     * 通過索引設(shè)置列表元素的值
     *
     * @param key
     * @param index
     *            位置
     * @param value
     */
    public void lSet(String key, long index, String value) {
        stringRedisTemplate.opsForList().set(key, index, value);
    }

    /**
     * 移出并獲取列表的第一個元素
     *
     * @param key
     * @return 刪除的元素
     */
    public String lLeftPop(String key) {
        return stringRedisTemplate.opsForList().leftPop(key);
    }

    /**
     * 移出并獲取列表的第一個元素， 如果列表沒有元素會阻塞列表直到等待超時或發(fā)現(xiàn)可彈出元素為止
     *
     * @param key
     * @param timeout
     *            等待時間
     * @param unit
     *            時間單位
     * @return
     */
    public String lBLeftPop(String key, long timeout, TimeUnit unit) {
        return stringRedisTemplate.opsForList().leftPop(key, timeout, unit);
    }

    /**
     * 移除并獲取列表最后一個元素
     *
     * @param key
     * @return 刪除的元素
     */
    public String lRightPop(String key) {
        return stringRedisTemplate.opsForList().rightPop(key);
    }

    /**
     * 移出并獲取列表的最后一個元素， 如果列表沒有元素會阻塞列表直到等待超時或發(fā)現(xiàn)可彈出元素為止
     *
     * @param key
     * @param timeout
     *            等待時間
     * @param unit
     *            時間單位
     * @return
     */
    public String lBRightPop(String key, long timeout, TimeUnit unit) {
        return stringRedisTemplate.opsForList().rightPop(key, timeout, unit);
    }

    /**
     * 移除列表的最后一個元素，并將該元素添加到另一個列表并返回
     *
     * @param sourceKey
     * @param destinationKey
     * @return
     */
    public String lRightPopAndLeftPush(String sourceKey, String destinationKey) {
        return stringRedisTemplate.opsForList().rightPopAndLeftPush(sourceKey,
                destinationKey);
    }

    /**
     * 從列表中彈出一個值，將彈出的元素插入到另外一個列表中并返回它； 如果列表沒有元素會阻塞列表直到等待超時或發(fā)現(xiàn)可彈出元素為止
     *
     * @param sourceKey
     * @param destinationKey
     * @param timeout
     * @param unit
     * @return
     */
    public String lBRightPopAndLeftPush(String sourceKey, String destinationKey,
                                        long timeout, TimeUnit unit) {
        return stringRedisTemplate.opsForList().rightPopAndLeftPush(sourceKey,
                destinationKey, timeout, unit);
    }
    
    /**
     * 刪除集合中值等于value得元素
     *
     * @param key
     * @param index
     *            index=0, 刪除所有值等于value的元素; index>0, 從頭部開始刪除第一個值等于value的元素;
     *            index<0, 從尾部開始刪除第一個值等于value的元素;
     * @param value
     * @return
     */
    public Long lRemove(String key, long index, String value) {
        return stringRedisTemplate.opsForList().remove(key, index, value);
    }

    /**
     * 裁剪list
     *
     * @param key
     * @param start
     * @param end
     */
    public void lTrim(String key, long start, long end) {
        stringRedisTemplate.opsForList().trim(key, start, end);
    }

    /**
     * 獲取列表長度
     *
     * @param key
     * @return
     */
    public Long lLen(String key) {
        return stringRedisTemplate.opsForList().size(key);
    }


    /** --------------------set相關(guān)操作-------------------------- */

    /**
     * set添加元素
     *
     * @param key
     * @param values
     * @return
     */
    public Long sAdd(String key, String... values) {
        return stringRedisTemplate.opsForSet().add(key, values);
    }

    /**
     * set移除元素
     *
     * @param key
     * @param values
     * @return
     */
    public Long sRemove(String key, Object... values) {
        return stringRedisTemplate.opsForSet().remove(key, values);
    }

    /**
     * 移除并返回集合的一個隨機元素
     *
     * @param key
     * @return
     */
    public String sPop(String key) {
        return stringRedisTemplate.opsForSet().pop(key);
    }

    /**
     * 將元素value從一個集合移到另一個集合
     *
     * @param key
     * @param value
     * @param destKey
     * @return
     */
    public Boolean sMove(String key, String value, String destKey) {
        return stringRedisTemplate.opsForSet().move(key, value, destKey);
    }

    /**
     * 獲取集合的大小
     *
     * @param key
     * @return
     */
    public Long sSize(String key) {
        return stringRedisTemplate.opsForSet().size(key);
    }

    /**
     * 判斷集合是否包含value
     *
     * @param key
     * @param value
     * @return
     */
    public Boolean sIsMember(String key, Object value) {
        return stringRedisTemplate.opsForSet().isMember(key, value);
    }

    /**
     * 獲取兩個集合的交集
     *
     * @param key
     * @param otherKey
     * @return
     */
    public Set<String> sIntersect(String key, String otherKey) {
        return stringRedisTemplate.opsForSet().intersect(key, otherKey);
    }

    /**
     * 獲取key集合與多個集合的交集
     *
     * @param key
     * @param otherKeys
     * @return
     */
    public Set<String> sIntersect(String key, Collection<String> otherKeys) {
        return stringRedisTemplate.opsForSet().intersect(key, otherKeys);
    }

    /**
     * key集合與otherKey集合的交集存儲到destKey集合中
     *
     * @param key
     * @param otherKey
     * @param destKey
     * @return
     */
    public Long sIntersectAndStore(String key, String otherKey, String destKey) {
        return stringRedisTemplate.opsForSet().intersectAndStore(key, otherKey,
                destKey);
    }

    /**
     * key集合與多個集合的交集存儲到destKey集合中
     *
     * @param key
     * @param otherKeys
     * @param destKey
     * @return
     */
    public Long sIntersectAndStore(String key, Collection<String> otherKeys,
                                   String destKey) {
        return stringRedisTemplate.opsForSet().intersectAndStore(key, otherKeys,
                destKey);
    }

    /**
     * 獲取兩個集合的并集
     *
     * @param key
     * @param otherKeys
     * @return
     */
    public Set<String> sUnion(String key, String otherKeys) {
        return stringRedisTemplate.opsForSet().union(key, otherKeys);
    }

    /**
     * 獲取key集合與多個集合的并集
     *
     * @param key
     * @param otherKeys
     * @return
     */
    public Set<String> sUnion(String key, Collection<String> otherKeys) {
        return stringRedisTemplate.opsForSet().union(key, otherKeys);
    }

    /**
     * key集合與otherKey集合的并集存儲到destKey中
     *
     * @param key
     * @param otherKey
     * @param destKey
     * @return
     */
    public Long sUnionAndStore(String key, String otherKey, String destKey) {
        return stringRedisTemplate.opsForSet().unionAndStore(key, otherKey, destKey);
    }

    /**
     * key集合與多個集合的并集存儲到destKey中
     *
     * @param key
     * @param otherKeys
     * @param destKey
     * @return
     */
    public Long sUnionAndStore(String key, Collection<String> otherKeys,
                               String destKey) {
        return stringRedisTemplate.opsForSet().unionAndStore(key, otherKeys, destKey);
    }

    /**
     * 獲取兩個集合的差集
     *
     * @param key
     * @param otherKey
     * @return
     */
    public Set<String> sDifference(String key, String otherKey) {
        return stringRedisTemplate.opsForSet().difference(key, otherKey);
    }

    /**
     * 獲取key集合與多個集合的差集
     *
     * @param key
     * @param otherKeys
     * @return
     */
    public Set<String> sDifference(String key, Collection<String> otherKeys) {
        return stringRedisTemplate.opsForSet().difference(key, otherKeys);
    }

    /**
     * key集合與otherKey集合的差集存儲到destKey中
     *
     * @param key
     * @param otherKey
     * @param destKey
     * @return
     */
    public Long sDifference(String key, String otherKey, String destKey) {
        return stringRedisTemplate.opsForSet().differenceAndStore(key, otherKey,
                destKey);
    }

    /**
     * key集合與多個集合的差集存儲到destKey中
     *
     * @param key
     * @param otherKeys
     * @param destKey
     * @return
     */
    public Long sDifference(String key, Collection<String> otherKeys,
                            String destKey) {
        return stringRedisTemplate.opsForSet().differenceAndStore(key, otherKeys,
                destKey);
    }

    /**
     * 獲取集合所有元素
     *
     * @param key
     * @param
     * @param
     * @return
     */
    public Set<String> setMembers(String key) {
        return stringRedisTemplate.opsForSet().members(key);
    }

    /**
     * 隨機獲取集合中的一個元素
     *
     * @param key
     * @return
     */
    public String sRandomMember(String key) {
        return stringRedisTemplate.opsForSet().randomMember(key);
    }

    /**
     * 隨機獲取集合中count個元素
     *
     * @param key
     * @param count
     * @return
     */
    public List<String> sRandomMembers(String key, long count) {
        return stringRedisTemplate.opsForSet().randomMembers(key, count);
    }

    /**
     * 隨機獲取集合中count個元素并且去除重復(fù)的
     *
     * @param key
     * @param count
     * @return
     */
    public Set<String> sDistinctRandomMembers(String key, long count) {
        return stringRedisTemplate.opsForSet().distinctRandomMembers(key, count);
    }

    /**
     *
     * @param key
     * @param options
     * @return
     */
    public Cursor<String> sScan(String key, ScanOptions options) {
        return stringRedisTemplate.opsForSet().scan(key, options);
    }

    /**------------------zSet相關(guān)操作--------------------------------*/

    /**
     * 添加元素,有序集合是按照元素的score值由小到大排列
     *
     * @param key
     * @param value
     * @param score
     * @return
     */
    public Boolean zAdd(String key, String value, double score) {
        return stringRedisTemplate.opsForZSet().add(key, value, score);
    }

    /**
     *
     * @param key
     * @param values
     * @return
     */
    public Long zAdd(String key, Set<TypedTuple<String>> values) {
        return stringRedisTemplate.opsForZSet().add(key, values);
    }

    /**
     *
     * @param key
     * @param values
     * @return
     */
    public Long zRemove(String key, Object... values) {
        return stringRedisTemplate.opsForZSet().remove(key, values);
    }

    public Long zRemove(String key, Collection<String> values) {
        if(values!=null&&!values.isEmpty()){
            Object[] objs = values.toArray(new Object[values.size()]);
            return stringRedisTemplate.opsForZSet().remove(key, objs);
        }
       return 0L;
    }

    /**
     * 增加元素的score值，并返回增加后的值
     *
     * @param key
     * @param value
     * @param delta
     * @return
     */
    public Double zIncrementScore(String key, String value, double delta) {
        return stringRedisTemplate.opsForZSet().incrementScore(key, value, delta);
    }

    /**
     * 返回元素在集合的排名,有序集合是按照元素的score值由小到大排列
     *
     * @param key
     * @param value
     * @return 0表示第一位
     */
    public Long zRank(String key, Object value) {
        return stringRedisTemplate.opsForZSet().rank(key, value);
    }

    /**
     * 返回元素在集合的排名,按元素的score值由大到小排列
     *
     * @param key
     * @param value
     * @return
     */
    public Long zReverseRank(String key, Object value) {
        return stringRedisTemplate.opsForZSet().reverseRank(key, value);
    }

    /**
     * 獲取集合的元素, 從小到大排序
     *
     * @param key
     * @param start
     *            開始位置
     * @param end
     *            結(jié)束位置, -1查詢所有
     * @return
     */
    public Set<String> zRange(String key, long start, long end) {
        return stringRedisTemplate.opsForZSet().range(key, start, end);
    }
    
    /**
     * 獲取zset集合的所有元素, 從小到大排序
     *
     */
    public Set<String> zRangeAll(String key) {
        return zRange(key,0,-1);
    }

    /**
     * 獲取集合元素, 并且把score值也獲取
     *
     * @param key
     * @param start
     * @param end
     * @return
     */
    public Set<TypedTuple<String>> zRangeWithScores(String key, long start,
                                                    long end) {
        return stringRedisTemplate.opsForZSet().rangeWithScores(key, start, end);
    }

    /**
     * 根據(jù)Score值查詢集合元素
     *
     * @param key
     * @param min
     *            最小值
     * @param max
     *            最大值
     * @return
     */
    public Set<String> zRangeByScore(String key, double min, double max) {
        return stringRedisTemplate.opsForZSet().rangeByScore(key, min, max);
    }


    /**
     * 根據(jù)Score值查詢集合元素, 從小到大排序
     *
     * @param key
     * @param min
     *            最小值
     * @param max
     *            最大值
     * @return
     */
    public Set<TypedTuple<String>> zRangeByScoreWithScores(String key,
                                                           double min, double max) {
        return stringRedisTemplate.opsForZSet().rangeByScoreWithScores(key, min, max);
    }

    /**
     *
     * @param key
     * @param min
     * @param max
     * @param start
     * @param end
     * @return
     */
    public Set<TypedTuple<String>> zRangeByScoreWithScores(String key,
                                                           double min, double max, long start, long end) {
        return stringRedisTemplate.opsForZSet().rangeByScoreWithScores(key, min, max,
                start, end);
    }

    /**
     * 獲取集合的元素, 從大到小排序
     *
     * @param key
     * @param start
     * @param end
     * @return
     */
    public Set<String> zReverseRange(String key, long start, long end) {
        return stringRedisTemplate.opsForZSet().reverseRange(key, start, end);

    }

    public Set<String> zReverseRangeByScore(String key, long min, long max) {
        return stringRedisTemplate.opsForZSet().reverseRangeByScore(key, min, max);

    }

    /**
     * 獲取集合的元素, 從大到小排序, 并返回score值
     *
     * @param key
     * @param start
     * @param end
     * @return
     */
    public Set<TypedTuple<String>> zReverseRangeWithScores(String key,
                                                           long start, long end) {
        return stringRedisTemplate.opsForZSet().reverseRangeWithScores(key, start,
                end);
    }

    /**
     * 根據(jù)Score值查詢集合元素, 從大到小排序
     *
     * @param key
     * @param min
     * @param max
     * @return
     */
    public Set<String> zReverseRangeByScore(String key, double min,
                                            double max) {
        return stringRedisTemplate.opsForZSet().reverseRangeByScore(key, min, max);
    }

    /**
     * 根據(jù)Score值查詢集合元素, 從大到小排序
     *
     * @param key
     * @param min
     * @param max
     * @return
     */
    public Set<TypedTuple<String>> zReverseRangeByScoreWithScores(
            String key, double min, double max) {
        return stringRedisTemplate.opsForZSet().reverseRangeByScoreWithScores(key,
                min, max);
    }

    /**
     *
     * @param key
     * @param min
     * @param max
     * @param start
     * @param end
     * @return
     */
    public Set<String> zReverseRangeByScore(String key, double min,
                                            double max, long start, long end) {
        return stringRedisTemplate.opsForZSet().reverseRangeByScore(key, min, max,
                start, end);
    }

    /**
     * 根據(jù)score值獲取集合元素數(shù)量
     *
     * @param key
     * @param min
     * @param max
     * @return
     */
    public Long zCount(String key, double min, double max) {
        return stringRedisTemplate.opsForZSet().count(key, min, max);
    }

    /**
     * 獲取集合大小
     *
     * @param key
     * @return
     */
    public Long zSize(String key) {
        return stringRedisTemplate.opsForZSet().size(key);
    }

    /**
     * 獲取集合大小
     *
     * @param key
     * @return
     */
    public Long zZCard(String key) {
        return stringRedisTemplate.opsForZSet().zCard(key);
    }

    /**
     * 獲取集合中value元素的score值
     *
     * @param key
     * @param value
     * @return
     */
    public Double zScore(String key, Object value) {
        return stringRedisTemplate.opsForZSet().score(key, value);
    }

    /**
     * 移除指定索引位置的成員
     *
     * @param key
     * @param start
     * @param end
     * @return
     */
    public Long zRemoveRange(String key, long start, long end) {
        return stringRedisTemplate.opsForZSet().removeRange(key, start, end);
    }

    /**
     * 根據(jù)指定的score值的范圍來移除成員
     *
     * @param key
     * @param min
     * @param max
     * @return
     */
    public Long zRemoveRangeByScore(String key, double min, double max) {
        return stringRedisTemplate.opsForZSet().removeRangeByScore(key, min, max);
    }

    /**
     * 獲取key和otherKey的并集并存儲在destKey中
     *
     * @param key
     * @param otherKey
     * @param destKey
     * @return
     */
    public Long zUnionAndStore(String key, String otherKey, String destKey) {
        return stringRedisTemplate.opsForZSet().unionAndStore(key, otherKey, destKey);
    }

    /**
     *
     * @param key
     * @param otherKeys
     * @param destKey
     * @return
     */
    public Long zUnionAndStore(String key, Collection<String> otherKeys,
                               String destKey) {
        return stringRedisTemplate.opsForZSet()
                .unionAndStore(key, otherKeys, destKey);
    }

    /**
     * 交集
     *
     * @param key
     * @param otherKey
     * @param destKey
     * @return
     */
    public Long zIntersectAndStore(String key, String otherKey,
                                   String destKey) {
        return stringRedisTemplate.opsForZSet().intersectAndStore(key, otherKey,
                destKey);
    }

    /**
     * 交集
     *
     * @param key
     * @param otherKeys
     * @param destKey
     * @return
     */
    public Long zIntersectAndStore(String key, Collection<String> otherKeys,
                                   String destKey) {
        return stringRedisTemplate.opsForZSet().intersectAndStore(key, otherKeys,
                destKey);
    }

    /**
     *
     * @param key
     * @param options
     * @return
     */
    public Cursor<TypedTuple<String>> zScan(String key, ScanOptions options) {
        return stringRedisTemplate.opsForZSet().scan(key, options);
    }

    /**
     * 掃描主鍵，建議使用
     * @param patten
     * @return
     */
    public Set<String> scan(String patten){
        Set<String> keys = stringRedisTemplate.execute((RedisCallback<Set<String>>) connection -> {
            Set<String> result = new HashSet<>();
            try (Cursor<byte[]> cursor = connection.scan(new ScanOptions.ScanOptionsBuilder()
                    .match(patten).count(10000).build())) {
                while (cursor.hasNext()) {
                    result.add(new String(cursor.next()));
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            return result;
        });
        return  keys;
    }
    
    /**
     * 管道技術(shù)，提高性能
     * @param type
     * @param values
     * @return
     */
    public List<Object> lRightPushPipeline(String type,Collection<String> values){
        List<Object> results = stringRedisTemplate.executePipelined(new RedisCallback<Object>() {
                    public Object doInRedis(RedisConnection connection) throws DataAccessException {
                        StringRedisConnection stringRedisConn = (StringRedisConnection)connection;
                        //集合轉(zhuǎn)換數(shù)組
                        String[] strings = values.toArray(new String[values.size()]);
                        //直接批量發(fā)送
                        stringRedisConn.rPush(type, strings);
                        return null;
                    }
                });
        return results;
    }

    public List<Object> refreshWithPipeline(String future_key,String topic_key,Collection<String> values){

        List<Object> objects = stringRedisTemplate.executePipelined(new RedisCallback<Object>() {
            @Nullable
            @Override
            public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
                StringRedisConnection stringRedisConnection = (StringRedisConnection)redisConnection;
                String[] strings = values.toArray(new String[values.size()]);
                stringRedisConnection.rPush(topic_key,strings);
                stringRedisConnection.zRem(future_key,strings);
                return null;
            }
        });
        return objects;
    }

}

到了這里，關(guān)于Redis在分布式場景下的應(yīng)用的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！