深入理解Redis內存回收和內存淘汰機制
1 概念
Redis 所有的數據都是存儲在內存中的, 如果不進行任何的內存回收, 那么很容易出現內存爆滿的情況。因此，在某些情況下需要對占用的內存空間進行釋放。
Redis 中內存的釋放主要分為兩類
Redis 中內存的釋放主要分為兩類:
內存回收: 將過期的 key 清除，以減少內存占用
內存淘汰: 在內存使用達到上限(max_memory), 按照一定的策略刪除一些鍵，以釋放內存空間
兩者都是通過刪除 key (及其對應的 value) 來達到釋放空間的效果。
區(qū)別在于前者清除的是用戶明確不需要的 key, 而后者清除的則是用戶可能仍然需要的 key。
2 內存回收
2.1 過期策略
在內存中的大量 key 中, 如何清除其中已經過期的 key 呢?
常用的方式有 3 種
- 定時過期
- 惰性過期
- 定期過期
定時過期
為每個 key 都創(chuàng)建一個定時器, 時間到了, 就將這個 key 清除。
該策略可以立即清除過期的數據, 對內存很友好。但是會占用大量的 CPU 資源去處理過期的數據, 從而影響緩存的響應時間和吞吐量。
惰性過期
key 過期了, 不進行處理。當后續(xù)訪問到這個 key 時, 才會判斷該 key 是否已過期, 過期則清除。
該策略可以最大化地節(jié)省 CPU 資源, 卻對內存非常不友好。極端情況可能出現大量的過期 key 沒有再次被訪問, 從而不會被清除, 占用大量內存。
定期過期
將所有的 key 維護在一起, 每隔一段時間就從中掃描一定的數量的 key(采樣), 并清除其中已經過期的 key。
通過調整定時掃描的時間間隔和每次掃描的耗時, 可以在不同情況下使得 CPU 和內存資源達到最優(yōu)的平衡效果。
在 Reids 的實現中是通過 惰性過期 + 定期過期 2 種策略配合, 達到內存回收的效果。
2.2 惰性過期 在 Redis 中的實現
前提: Redis 中一個對象的過期時間存放在 dictEntry 的 v.s64 中, 至于 dictEntry 的設計可以看一下后面的附錄。
Redis 大部分讀寫對象的命令, 在執(zhí)行前都會調用 expireIfNeeded 函數做一個過期檢查
- 如果 key 已經過期了, 將其刪除
- 如果 key 未過期, 不做任何處理
expireIfNeeded 函數的定義如下
int expireIfNeeded(redisDb *db, robj *key) {
// key 未過期返回 0
if (!keyIsExpired(db,key)) return 0;
// 下面的邏輯都是 Key 過期的邏輯處理
// 當前的節(jié)點是從節(jié)點, 返回 1, 然后結束
// 為了保持主從數據的一致, 從節(jié)點不會主動清除數據, 都是主節(jié)點同步消息在刪除
if (server.masterhost != NULL) return 1;
// 已經刪除過期鍵個數 + 1
server.stat_expiredkeys++;
// 向從節(jié)點和 AOF 文件傳播 key 過期信息, 清除過期 key
propagateExpire(db,key,server.lazyfree_lazy_expire);
// 發(fā)送事件通知
notifyKeyspaceEvent(NOTIFY_EXPIRED,"expired",key,db->id);
// lazyfree-lazy-expire 配置參數 (版本 4.0 以上支持), 默認為 0
// 根據配置, 同步或異步刪除 key (異步刪除: 先將 key 邏輯刪除, 然后在通過后臺的線程池進行真正的空間釋放)
return server.lazyfree_lazy_expire ? dbAsyncDelete(db,key) : dbSyncDelete(db,key);
}
int keyIsExpired(redisDb *db, robj *key) {
// 從過期字典中獲取 key 對應的過期時間, 實際就是獲取 dictEntity 的 v 中的 s64 值 (dictEntity.v.s64)
mstime_t when = getExpire(db,key);
mstime_t now;
// 沒有過期時間
if (when < 0) return 0;
// redis 在加載數據中
if (server.loading) return 0;
// 獲取當前的事件
if (server.lua_caller) {
// 有 lua 腳本在執(zhí)行中, 當前時間等于腳本開始執(zhí)行前的時間
now = server.lua_time_start;
} else if (server.fixed_time_expire > 0) {
// 有緩存時間, 線使用緩存時間
// server.mstime 這個時間會在調用執(zhí)行命令函數的 call() 前進行更新
// 這樣可以避免一些批量操作的命令, 比如 RPOPLPUSH 等命令, 這些命令會執(zhí)行過程中可能多次訪問這個 key
// 而在多次的訪問過程中, 可能出現上一次訪問未過期, 下次訪問已經過期了, 通過這個緩沖時間可以解決這個問題
now = server.mstime;
} else {
// 其他情況, 直接獲取當前時間
now = mstime();
}
// 當前時間是否大于 key 的過期時間
return now > when;
}
expireIfNeeded 的調用時機, 基本都是在各個命令內部。 以 String 的 get 命令為例, 大體的流程如下
/**
* get 命令對應的執(zhí)行函數
* 需要的參數都封裝在 client 對象中
*/
void getCommand(client *c) {
// getGenericCommand -> lookupKeyReadOrReply -> lookupKeyRead -> lookupKeyReadWithFlags
// getGenericCommand 經過幾個函數最終調用到 lookupKeyReadWithFlags
getGenericCommand(c);
}
robj *lookupKeyReadWithFlags(redisDb *db, robj *key, int flags) {
robj *val;
// expireIfNeeded 返回 > 0, 過期了
if (expireIfNeeded(db,key) == 1) {
// 省略過期處理
// 過期的處理, 然后 return null
}
// 非過期處理, 查找然后返回
val = lookupKey(db,key,flags);
if (val == NULL)
server.stat_keyspace_misses++;
else
server.stat_keyspace_hits++;
return val;
}
上面就是 get 指令的中的惰性過期的過程, 其他命令的邏輯差不多, 核心就是一個 expireIfNeeded 函數。
2.3 定期過期在 Redis 中的實現
Redis 默認是 16 個數據庫, 每個數據庫會將設置了過期時間的 key 放到各自的一個獨立的字典中, 稱為過期字典 (redisDb 對象的 dict *expires 屬性)。
然后 Redis 默認會按照每秒 10 次的頻率（可以通過 redis.conf 中的 hz 配置）進行過期掃描。
掃描的過程不會遍歷整個過期字典，而是按照以下策略進行
- 從過期字典中隨機選擇 20 個 key
- 刪除其中已經過期的鍵
- 如果超過 25% 的鍵被刪除, 則重復步驟 1, 2, 3, 沒有超過, 就結束這次掃描
- 同時為防止重復循環(huán), 導致線程卡死, 增加了每 16 次抽樣, 就做一次掃描時間的上限的檢查 (默認是慢模式下, 上限是 25 毫秒, 如果是快模式，掃描上限是 1 毫秒), 超過就結束循環(huán)
定期過期刪除的實現主要在 /activeExpireCycle 函數, 大體的邏輯如下
/**
* 過期循環(huán)清除
* 為了便于理解, 這里對函數的邏輯做了一點小調整和刪除一些非必要的邏輯, 但是整體的邏輯不變
* @type 模式, 取值有 2 個 ACTIVE_EXPIRE_CYCLE_SLOW (0, 慢模式), ACTIVE_EXPIRE_CYCLE_FAST (1, 快模式)
*/
void activeExpireCycle(int type) {
// 靜態(tài)變量, 當前處理的數據庫索引
// 靜態(tài)的效果, 這個變量執(zhí)行后的值不會被清空, 每次調用這個方法, 是上一次執(zhí)行的值
// 這樣就可以保證 16 個數據庫, 每次方法執(zhí)行完, 下次進來可以執(zhí)行到下一個數據庫, 循環(huán)起來，而不是每次進來都從第 0 個開始
static unsigned int current_db = 0;
// 上一次清理是否是因為時間超時結束循環(huán)的, 同樣是靜態(tài)變量
static int timelimit_exit = 0;
// 上一次快速循環(huán)循環(huán)的時間, 同樣是靜態(tài)變量
static long long last_fast_cycle = 0;
// 當前時間
long long start = ustime(),
// 本次循環(huán)清除是快速循環(huán), 上一次是時間超時獲取 2 次快速循環(huán)的時間差在 2 毫秒內, 不執(zhí)行
if (type == ACTIVE_EXPIRE_CYCLE_FAST) {
// 上一次循環(huán)是因為時間超時結束的, 本次快速循環(huán)不進行
if (!timelimit_exit) return;
// 上次快速循環(huán)距離當前時間在 1000 * 2 = 2 毫秒內, 也不進行快速循環(huán)
if (start < last_fast_cycle + ACTIVE_EXPIRE_CYCLE_FAST_DURATION*2) return;
last_fast_cycle = start;
}
// 計算循環(huán)的上限毫秒限制
// server.hz 默認等于 10, ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC 等于 25
// 1000000 * 25 / 10 / 100 = 25000 單位: 微秒, 即 25 毫秒
long long timelimit = 1000000*ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC/server.hz/100;
// ACTIVE_EXPIRE_CYCLE_FAST_DURATION = 1000
// 如果是快模式, 修改為 1000 微秒, 即 1 毫秒超時
if (type == ACTIVE_EXPIRE_CYCLE_FAST)
timelimit = ACTIVE_EXPIRE_CYCLE_FAST_DURATION;
// CRON_DBS_PER_CALL = 16, 每次循環(huán)處理的數據庫數量
int dbs_per_call = CRON_DBS_PER_CALL;
// 遍歷當前數據庫的次數
int iteration = 0;
// 遍歷循環(huán) 16 個數據庫
for (int j = 0; j < dbs_per_call && timelimit_exit == 0; j++) {
// 清理過期的 key 個數
int expired;
// 計算本次處理的數據庫
redisDb *db = server.db+(current_db % server.dbnum);
current_db++;
do {
// 開始循環(huán)清除當前數據庫中過期的 key
// 遍歷次數 + 1
iteration++;
// dictSize 獲取整個過期字典的已經使用大小
unsigned long num = dictSize(db->expires);
// num == 0 表示整個字典沒有數據, 跳出循環(huán)，處理下一個數據庫
if (num == 0) {
break;
}
// 計算整個過期字典的總大小
unsigned long slots = dictSlots(db->expires);
// DICT_HT_INITIAL_SIZE = 4, 每個字典初始化時的默認值
// num > 0, 字典中有數據了, slots 大于 4, 表示當前的字典擴容過了
// num && slots > DICT_HT_INITIAL_SIZE, 當前的字典擴容過同時里面有數據
// num * 100 / slots < 1 計算當前使用的數據占整個字典的百分比是否小于 1%
// Redis 認為, 如果一個字典中的使用率小于 1%, 花時間去進行清理是一個昂貴的操作
// 應該停下來，等待更好的時間再進行調整
// 所以簡單理解: 當這個字典中使用的空間小于 1%, 這里跳過了這個數據的處理
if (num && slots > DICT_HT_INITIAL_SIZE && (num * 100 / slots < 1))
break;
expired = 0;
// ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP = 20
// 本次從過期字典中獲取多少個 key, 如果字典中的已經使用的 key 大于 20, 則只取 20 個, 否則有多少取多少
if (num > ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP)
num = ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP;
// 循環(huán) num 次從字典中獲取 key
while (num--) {
dictEntry *de;
// 從過期字典中隨機獲取一個 key, 獲取不到, 就停止本次循環(huán)
if ((de = dictGetRandomKey(db->expires)) == NULL) break;
// 嘗試釋放這個 key, 如果 key 釋放成功, 過期次數 + 1
if (activeExpireCycleTryExpire(db,de,now)) expired++;
}
// 0xf = 15, iteration 表示遍歷了 15 次
if ((iteration & 0xf) == 0) {
// 計算消耗時間
int elapsed = ustime()-start;
// 消耗時間超過了限制時間, 結束本次循環(huán)
if (elapsed > timelimit) {
// 超過時間限制標識設置為 true, 本次循環(huán)清除超時了, 結束本次循環(huán)清除
timelimit_exit = 1;
break;
}
}
// 本次清理的過期 key 超過了 25%, 繼續(xù), 否則結束
// ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP = 20
// 每次抽取的個數最大為 20 個, 控制 25%, 20 * 25% = 5 個
// 也就是過期的個數大于 5 就是大于 25%, (ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP/4 = 5)
} while (expired > ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP/4);
}
// 省略各種分析數據的記錄
}
調用 activeExpireCycle 的入口有 2 個
- Redis 定時事件觸發(fā)
/**
* Reids 啟動時, 向事件輪詢中注冊的唯一一個定時事件(默認 100 毫秒執(zhí)行一次), 執(zhí)行的函數
*/
int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
...
// 數據庫掃描
databasesCron();
...
}
void databasesCron(void) {
// 過期功能開啟中, 默認為開啟
if (server.active_expire_enabled) {
// 主節(jié)點
if (server.masterhost == NULL) {
// 慢模式循環(huán)清除
activeExpireCycle(ACTIVE_EXPIRE_CYCLE_SLOW);
} else {
// 從節(jié)點處理
expireSlaveKeys();
}
}
...
}
- 事件輪詢中, 進入阻塞前的調用函數
void beforeSleep(struct aeEventLoop *eventLoop) {
...
// 過期功能開啟中同時為主節(jié)點
if (server.active_expire_enabled && server.masterhost == NULL)
// 快模式循環(huán)清除
activeExpireCycle(ACTIVE_EXPIRE_CYCLE_FAST);
...
}
3 內存淘汰
3.1 淘汰算法
為了能夠騰出內存空間, 需要在一大群對象中選擇某一些進行淘汰, 哪么應該基于什么標準進行選擇呢?
比較常見的算法有 2 個: LRU 和 LFU。
LRU (Least Recently Used): 最近最少使用算法, 根據數據的歷史訪問記錄進行淘汰數據，優(yōu)先移除最近最少使用的數據。
簡單理解就是根據對象的訪問時間, 優(yōu)先淘汰訪問時間最早的對象。
LFU (Least Frequently Used): 最少頻率使用算法, 根據數據的訪問頻率頻率進行淘汰數據, 優(yōu)先移除最近使用頻率最少的數據。
簡單理解就是根據對象的訪問次數, 優(yōu)先淘汰訪問次數最少的對象。
3.2 Redis 內存淘汰策略
在 LFU 和 LRU 的基礎上, Redis 提供了 8 種淘汰策略
| 策略 | 說明 |
|---|---|
| noeviction | 默認策略, 不會刪除任何數據, 但是拒絕所有寫入操作并返回客戶端錯誤信息 (error)OOM command not allow when used memory。此時 Redis 只響應讀操作。 |
| volatile-lru | Least Recently Used, 最近最少使用。在所有設置了 expire 的 key 中刪除最近最少使用的鍵值對, 即距離上次訪問時間最久的。 |
| allkeys-lru | Least Recently Used, 最近最少使用。在所有的 key 中刪除最近最少使用的鍵值對, 即距離上次訪問時間最久的。 |
| volatile-lfu | Least Frequently Used, 最不經常使用。在所有設置了 expire 的 key 中刪除最不經常使用的鍵值對, 即訪問次數最少的。 |
| allkeys-lfu | Least Frequently Used, 最不經常使用。在所有的 key 中刪除最不經常使用的鍵值對, 即訪問次數最少的。 |
| volatile-random | 在所有設置了 expire 的 key 中隨機選擇刪除 |
| allkeys-random | 在所有的 key 中隨機選擇刪除。 |
| volatile-ttl | Time To Live, 存活時間。 在所有設置了 expire 的 key 中刪除 ttl 值最多的。 |
volatile-lru, volatile-random, volatile-ttl, 在沒有符合條件的 key 的情況下, 會按照 noeviction 的策略進行處理。
3.3 Redis 對象淘汰判斷標準設計
在上面介紹的幾種策略可以知道, 要判斷一個對象是否可以被淘汰, 需要對象自身存放使用策略對應的數據, 以便于判斷
比如:
2 個 lru 策略, 需要對象自身保存好上次訪問的時間
2 個 lfu 策略, 需要對象自身保存好訪問次數
ttl 策略, 需要對象自身保存好過期時間
2 個 random 策略, 不需要保存額外的數據, 通過隨機一個數, 根據這個數從字典中獲取數據即可
3.3.1 Redis 對象的設計
正常情況下, 當我們向 Redis 中存入一對鍵值對, 實際可以拆分為 2 個對象, 一個 key, 一個 value。
其中 key 可以明確為是一個字符串, 所以存入到 Redis 的鍵值對的 key 會被封裝為 sds 對象。
但是 value 可以類型可以很多, 為了行為的統(tǒng)一等, 需要對 value 做一個封裝, 落實到源碼中就是一個 redisObject 對象, 其定義如下
typedef struct redisObject {
/**
* 標識這個對象的數據類型, 常說的 String, Hash, List 等
*/
unsigned type:4;
/**
* 可以理解為數據類型的具體實現類型
* 比如數據類型為 List, 在具體的實現中可以是 ArrayList LinkedList 等
*/
unsigned encoding:4;
/**
* LRU_BITS = 24,
* 一個 24 位的變量, 表示對象最后一次被程序訪問的時間或者訪問的次數, 與內存回收有關
* 暫時知道有這個對象即可, 后面有分析
*/
unsigned lru:LRU_BITS;
/**
* 被引用的次數, 當 refcount 為 0 的時候, 表示該對象已經不被任何對象引用, 則可以進行垃圾回收了
*/
int refcount;
/**
* 一個指針, 指向具體的數據
*/
void *ptr;
} robj;
一個對象的 lru 和 lfu 計算后的值, 都是存放在這個對象的 lru 字段中的, 但是 lru 和 lfu 的計算方式是不一樣的。
3.3.2 lru 策略, 對象的訪問時間設計
3.3.2.1 全局時間 lruclock
在 Redis 的中維護了一個全局的變量 lruclock, 表示當前時間的一個相對值。
/**
* redisServer 可以看做整個 Redis 運行時的上下文, 保存的數據, 配置等都在這個結構體中
*/
struct redisServer {
unsigned int lruclock = getLRUClock();
}
unsigned int getLRUClock(void) {
// LRU_CLOCK_RESOLUTION = 1000
// mstime() 當前時間毫秒, 當前時間的毫秒/LRU_CLOCK_RESOLUTION = 當前時間的毫秒/1000 = 變?yōu)閱挝幻?
// LRU_CLOCK_MAX = ((1<<LRU_BITS)-1) = 1<<24-1 = redisObject lru 字段的最大值
// (當前的時間 / 1000) & (1<<24-1) 確保時間的精度是秒, 同時不會超過 24 位的整數的最多值
// 整個全局時間的進度為秒, 2 個對象的訪問時間差如果在秒內, 得到的是他們的訪問時間是一樣的
// 得到一個當前時間的相對值
return (mstime()/LRU_CLOCK_RESOLUTION) & LRU_CLOCK_MAX;
}
同時這個時間會在 Redis 的定時任務 serverCron 中定時的更新為最新的值
int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
// serverCron 默認是 100 毫秒執(zhí)行一次
unsigned int lruclock = getLRUClock();
atomicSet(server.lruclock,lruclock);
}
3.3.2.2 對象的訪問時間設計
Redis 每次通過 key 在數據庫中查詢對應的 value 時, 在找到時, 就會進行 lru 字段的更新
robj *lookupKey(redisDb *db, robj *key, int flags) {
// 從字典中獲取 key 對應的 dictEntry (字典的設計可以看一下后面的附錄)
dictEntry *de = dictFind(db->dict,key->ptr);
if (de) {
// 獲取 key 對應的 dictEntry 的存在
// 獲取 dictEntry 的 value 也就是 redisObject 對象
robj *val = dictGetVal(de);
if (server.rdb_child_pid == -1 && server.aof_child_pid == -1 && !(flags & LOOKUP_NOTOUCH)) {
// 沒有在進行 RDB 或 AOF 操作, 并且 flags 沒有設置 LOOKUP_NOTOUCH
// 淘汰策略設置的的 LFU 策略
if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
updateLFU(val);
} else {
// 其他策略, 更新 lru 為全局的 lruclock
val->lru = LRU_CLOCK();
}
}
} else {
// key 不存在, 返回 null
return NULL;
}
}
unsigned int LRU_CLOCK(void) {
unsigned int lruclock;
// LRU_CLOCK_RESOLUTION = 1000
// 1000/server.hz 就是上面定時任務 serverCron 的執(zhí)行時間
// <= 1000 說明 serverCron 的執(zhí)行時間小于 1 秒, 直接獲取 server.lruclock 的值
// 如果大于 1000, 就調用 getLRUClock() 實時獲取當前的時間, 因為頻率太低了, 會造成更多的對象的訪問時間一樣
if (1000/server.hz <= LRU_CLOCK_RESOLUTION) {
atomicGet(server.lruclock,lruclock);
} else {
lruclock = getLRUClock();
}
return lruclock;
}
3.3.3 lfu 策略, 對象的訪問頻率設計
對象的 lfu 同樣是存放在 redisObject 的 lru:LRU_BITS 字段。 這個 24 bits 字段, 被分為兩部分
高 16 位用來記錄訪問時間 (單位為分鐘，ldt, last decrement time)
低 8 位用來記錄相對的訪問次數, 簡稱 counter (logc, logistic counter)
Redis 中對 LFU 的實現比較特殊, 通過時間衰減的方式近似達到了 LFU 的效果。
大體的思路如下:
對象創(chuàng)建時, 初始訪問次數為 5 (避免剛創(chuàng)建出來, 對象就被回收), 同時記錄下當前時間, 單位分鐘
對象被訪問時, 獲取當前時間, 單位分鐘, 當前時間 - 對象本身記錄的時間, 得到相差多少分鐘, 訪問次數就減少多少
然后對象的訪問次數 + 1, 再次記錄下當前時間
這樣對象在單位分鐘內, 訪問越頻繁, 訪問次數越大, 同時隨著時間的推移, 沒有進行訪問, 訪問次數會逐漸減少, 從而達到了 LFU 的效果。
ldt 記錄的是最近一次訪問的時間, 16 位, 所以最大值為 65535, 單位是分鐘, 差不多 45 天左右。
也就是一個對象如果一直被訪問, 到了第 45 天后, 這個值又會重新回到 0 開始計算。
ldt 的計算
unsigned long LFUGetTimeInMinutes(void) {
// & 65535 保證時間的范圍在 0 ~ 65535 之間, 不會超過 16 數值的大小
return (server.unixtime/60) & 65535;
}
同 lru 一樣, lruclock 的計算, 后面的時間比前面的時間小,
說明后面的時間到了下一輪的重新開始了, 這時只需要后面的時間 + 65535 - 前面的時間, 就能得到 2 個時間的差值了。
logc 記錄的是一個相對的訪問次數。
本身只有 8 位, 也就是最大值為 255, 也就是一個對象只能保存 255 次訪問次數, 這個基本不同滿足日常的使用。
所以 Redis 內部設計了一個隨機公式, 控制訪問次數的增長, 即每次訪問, 訪問次數加不加一, 通過隨機判斷。
uint8_t LFULogIncr(uint8_t counter) {
// 當前的訪問次數已經達到了最大值了
if (counter == 255)
return 255;
// 產生一個隨機數
double r = (double)rand()/RAND_MAX;
// 獲取一個基礎值, 當前的次數 - 對象初始化的默認次數 (LFU_INIT_VAL = 5)
double baseval = counter - LFU_INIT_VAL;
if (baseval < 0) baseval = 0;
// 1.0 / 基礎值 * server.lfu_log_factor (默認值, 10, 可配置) + 1, 得到一個數
double p = 1.0/(baseval*server.lfu_log_factor+1);
// 得到的數大于隨機出來的數, 訪問次數 + 1
if (r < p) counter++;
return counter;
}
官方的測試數據 (可以簡單看成, counter = 5, 在 100 - 1000w 次的調用, lfu_log_factor 不同取值下, 最終的 counter 的值)
| lfu_log_factor 取值 | 100 次 | 1000 次 | 10w 次 | 100w 次 | 1000w 次 |
|---|---|---|---|---|---|
| 0 | 104 | 255 | 255 | 255 | 255 |
| 1 | 18 | 49 | 255 | 255 | 255 |
| 10 | 10 | 18 | 142 | 255 | 255 |
| 100 | 8 | 11 | 49 | 143 | 255 |
lfu_log_factor 設置為 10 的情況下, 在 100w 次的訪問中, 訪問次數才達到為 255, 也就是最大值。
基本可以滿足 10w 次的使用
3.3.3.1 counter 衰減機制
每個對象被返回時, counter 都會先進行一個衰減操作, 然后再通過上面的隨機公式進行判斷次數是否需要增加。
衰減的過程如下
unsigned long LFUDecrAndReturn(robj *o) {
// 右移 8 為, 也就是得的了高位的 16 位, 即 ldt, 得到上次記錄的時間
unsigned long ldt = o->lru >> 8;
// 得到當前保存的次數
unsigned long counter = o->lru & 255;
// lfu_decay_time 衰減時間, 默認 1, 單位分鐘
// 如果沒有配置 lfu_decay_time, 則默認不進行衰減, counter 當前是多少就是多少
// 獲取 2 次訪問的時間差 / lfu_decay_time, 得到經過了多少個時間段
unsigned long num_periods = server.lfu_decay_time ? LFUTimeElapsed(ldt) / server.lfu_decay_time : 0;
if (num_periods)
// 最新的次數 = 當前的次數 - 經過了多少個時間段, 小于 0 時, 設置為 0
counter = (num_periods > counter) ? 0 : counter - num_periods;
return counter;
}
// 距離上次訪問相差多少分鐘
unsigned long LFUTimeElapsed(unsigned long ldt) {
unsigned long now = LFUGetTimeInMinutes();
if (now >= ldt) return now-ldt;
return 65535-ldt+now;
}
3.3.3.2 對象的訪問頻率設計
Redis 每次通過 key 在數據庫中查詢對應的 value 時, 在找到時, 就會進行 lru 字段的更新
robj *lookupKey(redisDb *db, robj *key, int flags) {
dictEntry *de = dictFind(db->dict,key->ptr);
if (de) {
robj *val = dictGetVal(de);
if (server.rdb_child_pid == -1 && server.aof_child_pid == -1 && !(flags & LOOKUP_NOTOUCH)) {
// 淘汰策略設置的的 LFU 策略
if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
updateLFU(val);
} else {
val->lru = LRU_CLOCK();
}
}
} else {
return NULL;
}
}
void updateLFU(robj *val) {
// 通過衰減機制, 得到最新的 counter
unsigned long counter = LFUDecrAndReturn(val);
// 通過隨機公式, 得到最新的 counter
counter = LFULogIncr(counter);
// 將最新的 counter 和 當前時間保存到 lru 字段中
val->lru = (LFUGetTimeInMinutes()<<8) | counter;
}
3.4 Redis 內存淘汰策略的實現
Redis 的內存的實現方式都是通過隨機采樣 + 比較 lru 值決定是否淘汰的方式實現的。
大體過程如下:
- Redis 啟動時, 會初始一個默認容量為 16 的待淘汰數據池 evictionPoolEntry (本質就是一個數組)
- 每個存入到 Redis 的對象 (redisObject) 都會在初始其 24 位的 lru 字段 (lru: 一個相對的訪問時間, lfu: 一個相對的訪問次數)
- 后面每次訪問 Redis 的對象時, 更新其 lru 字段的值
- 同時每次執(zhí)行一個 Redis 命令時, 就會判斷一下當前的內存是否足夠, 如果不夠, 就計算出需要釋放多少內存, 然后進行內存淘汰
內存淘汰的過程如下:
4.1 首次淘汰從數據字典或過期字典 (由配置的淘汰策略決定) 中隨機抽樣選出最多 N 個數據放入到一個樣例池。
數據量 N: 由 redis.conf 配置的 maxmemory-samples 決定, 默認值是 5。 配置為 10 將非常接近真實 LRU 效果。
采樣參數 maxmemory-samples 配置的數值越大, 就越能精確的查找到待淘汰的緩存數據, 但是也消耗更多的 CPU 計算, 執(zhí)行效率降低。
同時為了避免長時間找不到足夠的數據填充樣例池, 強制寫死了單次尋找數據的最大次數是 maxsteps = N*10。
4.2 再次淘汰遍歷整個樣例池, 遍歷的對象通過 lru 計算處理的值, 只要比待淘汰數據池中的任意一條數據的小, 就將該數據填充至待淘汰數據池。
第一次淘汰時, 待淘汰數據池為空, 所以第一次淘汰時, 會將所有的樣例數據填充到待淘汰數據池中, 這個池子后面就都會有數據, 一直存在著。
后續(xù)的淘汰時, 樣例池 中的數據就有可能進入到待淘汰數據池中, 也有可能不進入。
4.3 執(zhí)行淘汰從待淘汰數據池的尾部向前找到第一個可以刪除的 key (此時找到的 key 就是值最小/大的, 既空閑時間最大/訪問次數最小/存活時間最小), 對其進行淘汰
4.4 繼續(xù)淘汰計算刪除了一個 key 后內存釋放了多少, 如果沒達到要求的釋放量, 就回到步驟 4.1 繼續(xù)淘汰
3.4.1 Redis 內存淘汰策略的代碼實現
入口: 每個命令的執(zhí)行處
int processCommand(client *c) {
...
// 有設置最大內存 同時當前沒有 lua 腳本超時的情況
if (server.maxmemory && !server.lua_timedout) {
// 有必要時, 嘗試釋放內存
int out_of_memory = freeMemoryIfNeededAndSafe() == C_ERR;
// 內存不夠 同時執(zhí)行的命令是變更命令 或者 當前的客戶端開啟了事務, 同時執(zhí)行的命令不是 exec
if (out_of_memory && (c->cmd->flags & CMD_DENYOOM || (c->flags & CLIENT_MULTI && c->cmd->proc != execCommand))) {
flagTransaction(c);
// 響應 -OOM command not allowed when used memory > 'maxmemory'
addReply(c, shared.oomerr);
return C_OK;
}
}
...
}
int freeMemoryIfNeededAndSafe(void) {
// 當前有 lua 腳本執(zhí)行超時或者真正加載數據, 返回成功
if (server.lua_timedout || server.loading) return C_OK;
// 是否內存如果有必要的話
return freeMemoryIfNeeded();
}
釋放內存的核心函數
int freeMemoryIfNeeded(void) {
// 如果是從節(jié)點同時配置了從節(jié)點忽略內存配置, 直接返回
if (server.masterhost && server.repl_slave_ignore_maxmemory) return C_OK;
// mem_reported 保存了整個 Redis 已經使用的內存
// mem_tofree 經過計算本次應該釋放的內存, 等于當前已經使用的內存 - 用于主從復制的復制緩沖區(qū)大小 - 配置的 maxmemory
// mem_freed 已經釋放了多少內存
size_t mem_reported, mem_tofree, mem_freed;
long long delta;
// 從節(jié)點個數
int slaves = listLength(server.slaves);
// 判斷當前的內存狀態(tài), 如果足夠, 直接返回
if (getMaxmemoryState(&mem_reported,NULL,&mem_tofree,NULL) == C_OK)
return C_OK;
// 如果配置的策略為 noeviction
if (server.maxmemory_policy == MAXMEMORY_NO_EVICTION)
goto cant_free;
mem_freed = 0;
// 沒有達到需要的內存大小, 繼續(xù)循環(huán)
while (mem_freed < mem_tofree) {
static unsigned int next_db = 0;
sds bestkey = NULL;
int bestdbid;
redisDb *db;
dict *dict;
dictEntry *de;
if (server.maxmemory_policy & (MAXMEMORY_FLAG_LRU|MAXMEMORY_FLAG_LFU) || server.maxmemory_policy == MAXMEMORY_VOLATILE_TTL) {
// LRU + LFU + TTL 策略
// 淘汰池
struct evictionPoolEntry *pool = EvictionPoolLRU;
while(bestkey == NULL) {
// 遍歷 16 個數據庫
for (i = 0; i < server.dbnum; i++) {
db = server.db+i;
// 根據 volatile 或 all 選擇對應的數據字典
dict = (server.maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS) ? db->dict : db->expires;
// 獲取字典的數據大小, keys 為當前數據庫的 key 的數量
if ((keys = dictSize(dict)) != 0) {
evictionPoolPopulate(i, dict, db->dict, pool);
total_keys += keys;
}
}
// 沒有可以處理的 keys
if (!total_keys) break;
// EVPOOL_SIZE = 16
for (k = EVPOOL_SIZE-1; k >= 0; k--) {
if (pool[k].key == NULL) continue;
bestdbid = pool[k].dbid;
// 從數據庫中獲取對應的節(jié)點
if (server.maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS) {
de = dictFind(server.db[pool[k].dbid].dict, pool[k].key);
} else {
de = dictFind(server.db[pool[k].dbid].expires, pool[k].key);
}
// 釋放緩存
if (pool[k].key != pool[k].cached)
sdsfree(pool[k].key);
pool[k].key = NULL;
pool[k].idle = 0;
// 找到的釋放對象存在, 先跳出這次循環(huán)
if (de) {
bestkey = dictGetKey(de);
break;
} else {
// 不存在, 進行循環(huán)查找
}
}
}
} else if (server.maxmemory_policy == MAXMEMORY_ALLKEYS_RANDOM || server.maxmemory_policy == MAXMEMORY_VOLATILE_RANDOM) {
// random 策略
}
// 刪除找到的 key
if (bestkey) {
db = server.db+bestdbid;
// 將 key 封裝為 redisObject 對象
robj *keyobj = createStringObject(bestkey,sdslen(bestkey));
// 傳播 key 過期信息到主從復制和 AOF 文件
propagateExpire(db,keyobj,server.lazyfree_lazy_eviction);
// 獲取當前的內存大小
delta = (long long) zmalloc_used_memory();
// 同步刪除或異步刪除 key
if (server.lazyfree_lazy_eviction) {
dbAsyncDelete(db,keyobj);
else
dbSyncDelete(db,keyobj);
}
// 計算本次釋放的內存
delta -= (long long) zmalloc_used_memory();
mem_freed += delta;
// 釋放創(chuàng)建的 key redisObject 對象
decrRefCount(keyobj);
keys_freed++;
// 如果有從節(jié)點, 推送緩沖區(qū)的數據
if (slaves) flushSlavesOutputBuffers();
// 支持異步清除 同時 清除了 16 個 key
if (server.lazyfree_lazy_eviction && !(keys_freed % 16)) {
// 再次判斷內存情況, 如果內存足夠了
if (getMaxmemoryState(NULL,NULL,NULL,NULL) == C_OK) {
// 更新已經釋放的緩存大小 = 需要釋放的緩存大小
mem_freed = mem_tofree;
}
}
}
// 本次釋放沒有處理成功任何一個 key
if (!keys_freed) {
goto cant_free;
}
}
return C_OK;
cant_free:
// 沒有內存可以分配了, 做唯一可以做的一件事: 檢查是否有 lazyfree 線程在執(zhí)行釋放內存任務, 有進行等待
// 知道沒有任務或者已有的內存達到了需要釋放的內存
while(bioPendingJobsOfType(BIO_LAZY_FREE)) {
// 當前的內存達到了現在需要的釋放的內存, 結束檢查
if (((mem_reported - zmalloc_used_memory()) + mem_freed) >= mem_tofree)
break;
usleep(1000);
}
return C_ERR;
淘汰池的填充
void evictionPoolPopulate(int dbid, dict *sampledict, dict *keydict, struct evictionPoolEntry *pool) {
int j, k, count;
// 采樣結果數組, 最大容量為 mamemory_samples 的大小
dictEntry *samples[server.maxmemory_samples];
// 從 sampledict 字典中采樣 server.maxmemory_samples 個 key 存放到 samples, 同時返回總共采樣的多少個
count = dictGetSomeKeys(sampledict,samples,server.maxmemory_samples);
for (j = 0; j < count; j++) {
unsigned long long idle;
sds key;
robj *o;
dictEntry *de;
de = samples[j];
key = dictGetKey(de);
if (server.maxmemory_policy != MAXMEMORY_VOLATILE_TTL) {
if (sampledict != keydict) de = dictFind(keydict, key);
o = dictGetVal(de);
}
if (server.maxmemory_policy & MAXMEMORY_FLAG_LRU) {
// LRU 算法
idle = estimateObjectIdleTime(o);
} else if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
// LRU 算法
idle = 255 - LFUDecrAndReturn(o);
} else if (server.maxmemory_policy == MAXMEMORY_VOLATILE_TTL) {
// TTL 算法
idle = ULLONG_MAX - (long)dictGetVal(de);
} else {
serverPanic("Unknown eviction policy in evictionPoolPopulate()");
}
k = 0;
// 從 evictionPoolEntry 淘汰池中找到第一個閑置時間比當前淘汰 key 大的
while (k < EVPOOL_SIZE && pool[k].key && pool[k].idle < idle)
k++;
if (k == 0 && pool[EVPOOL_SIZE-1].key != NULL) {
// 如果找到的 key 比淘汰池中閑置時間最小的 key 還小, 同時淘汰池沒有空間了, 則跳過這個 key
continue;
} else if (k < EVPOOL_SIZE && pool[k].key == NULL) {
// 插入的位置為空, 直接進入到下面的賦值節(jié)點
} else {
// 核心就是將找到的位置 k 空出來
// 最后的位置為空
if (pool[EVPOOL_SIZE-1].key == NULL) {
// 將原本 k 位置和后面的數據向后移動 1 位
sds cached = pool[EVPOOL_SIZE-1].cached;
memmove(pool+k+1, pool+k, sizeof(pool[0])*(EVPOOL_SIZE-k-1));
pool[k].cached = cached;
} else {
// 插入的位置不為空
// 將原本 k 位置前面的數據往前移動 1 位, 原本的第一位丟棄
k--;
sds cached = pool[0].cached;
if (pool[0].key != pool[0].cached) sdsfree(pool[0].key);
memmove(pool,pool+1,sizeof(pool[0])*k);
pool[k].cached = cached;
}
}
// 把找到的 key 放到 k 的位置
int klen = sdslen(key);
// EVPOOL_CACHED_SDS_SIZE = 255
if (klen > EVPOOL_CACHED_SDS_SIZE) {
// 創(chuàng)建一個新的 key 賦值給 pool[k].key
pool[k].key = sdsdup(key);
} else {
// 從 key 中拷貝 klen + 1 的長度到 pool[k].cached
memcpy(pool[k].cached,key,klen+1);
sdssetlen(pool[k].cached,klen);
pool[k].key = pool[k].cached;
}
pool[k].idle = idle;
pool[k].dbid = dbid;
}
}
unsigned int dictGetSomeKeys(dict *d, dictEntry **des, unsigned int count) {
unsigned long j;
unsigned long tables;
unsigned long stored = 0, maxsizemask;
unsigned long maxsteps;
// 字典中的數據量小于需要的個數, 取的個數變?yōu)樽值涞臄祿笮?
if (dictSize(d) < count) count = dictSize(d);
// 最大次數 = 次數 * 10
maxsteps = count*10;
/* 如果字典在 rehash 中, 嘗試 count 一樣次數的 rehash */
for (j = 0; j < count; j++) {
if (dictIsRehashing(d))
_dictRehashStep(d);
else
break;
}
// 獲取總的 HashTable 個數, 如果在 rehash 中就是 2 個, 否則 1 個
tables = dictIsRehashing(d) ? 2 : 1;
// 獲取數組大小的掩碼, 用于計算索引值
maxsizemask = d->ht[0].sizemask;
if (tables > 1 && maxsizemask < d->ht[1].sizemask)
maxsizemask = d->ht[1].sizemask;
// 隨機獲取一個位置
unsigned long i = random() & maxsizemask;
unsigned long emptylen = 0;
// 獲取到的個數沒達到需要的個數 或者嘗試的次數還沒達到 0
while(stored < count && maxsteps--) {
for (j = 0; j < tables; j++) {
// 如果字典在 rehash 中, 同時當前處理的是第一個字典, 處理的位置小于 rehash 下次處理的位置,
// 則跳過這個位置, 直接到 rehash 下次處理的位置
// 因為第一個字典 rehash 下次處理的位置前的數據都遷移到第二個字典中了
if (tables == 2 && j == 0 && i < (unsigned long) d->rehashidx) {
// 防止獲取數據的位置 i 超過第二個字典的大小
if (i >= d->ht[1].size)
i = d->rehashidx;
else
continue;
}
// 超過了數組的長度
if (i >= d->ht[j].size) continue;
// 獲取對應位置的數據
dictEntry *he = d->ht[j].table[i];
// 對應的位置為 null
if (he == NULL) {
emptylen++;
// 獲取 null 數據的次數大于 5 次 同時 大于需要的過期 key 的個數
if (emptylen >= 5 && emptylen > count) {
// 重新計算獲取的位置 i, 重新獲取
i = random() & maxsizemask;
emptylen = 0;
}
} else {
emptylen = 0;
while (he) {
// he 本身是鏈表, 計算從鏈表中獲取到的個數, 夠了結束, 不夠就 i+1, 從字典的下一個位置繼續(xù)獲取
*des = he;
des++;
he = he->next;
stored++;
if (stored == count) return stored;
}
}
}
i = (i+1) & maxsizemask;
}
return stored;
}
dictGetSomeKeys 函數簡單理解就是, 通過 random() 得到一個隨機數, 這個隨機數 & 數組大小的掩碼, 得到一個位置, 從這個位置向后獲取 count 個過期 key。
這個處理的過程中
有可能字典在 rehash 中, 數據分布在 2 個字典中, 所以有時第一個字典獲取不到需要到第二個字典獲取
需要的過期 key 的個數小于等于 5 個, 通過計算得到的位置獲取到的數據連續(xù)都為 null, 則重新通過 random() 計算一個新的位置
為了防止長時間的需要, 在外面還計算了最大的循環(huán)次數
從上面的代碼實現可以看出, Redis 內部對 LRU + LFU 的實現都是不是很正式的實現, 帶有一定的誤差和隨機性。
其本身考慮主用是從性能上做的折中。比如傳統(tǒng)的 LRU 算法, 需要將所有的數據維護一個雙向鏈表
訪問節(jié)點, 如果節(jié)點存在, 則將該節(jié)點移動到鏈表的頭節(jié)點, 并返回節(jié)點值, 不存在就返回 null
新增節(jié)點, 節(jié)點不存在, 就在鏈表的頭部新增節(jié)點, 如果節(jié)點存在, 則更新節(jié)點數據, 然后將節(jié)點移動到鏈表的頭節(jié)點
需要消耗的內存在維護鏈表的 + 節(jié)點的挑戰(zhàn), 對于一個大規(guī)模的數據, 這個消耗是非常大的。
所以 Redis 采用了其思想, 通過另外的方式達到類似的效果。
4 附錄: Redis 幾個對象的介紹
4.1 Redis 中的字典
4.2.1 HashTable
存儲在 Redis 中的基本都是鍵值對, 而這種鍵值對存儲, 同時可以通過 key 快速查詢到對應的 value, 最合適的實現就是 HashTable 了。
而實現 HashTable 的底層結構，基本就是一個數組或者鏈表, 同時為了解決 hash 沖突, 數組或鏈表的每個節(jié)點定義為一個鏈表。
Redis 中對 HashTable 的實現也是如此, 大體如下
Redis 中實現的 HastTable 叫做 dictht (Dictionary Hash Table)
對應的定義如下:
typedef struct dictht {
// 存放節(jié)點的數組
dictEntry **table;
// HashTable 的大小, 2 的冪次方
unsigned long size;
// HashTable 的大小掩碼, 用于計算索引值
unsigned long sizemask;
// HashTable 中已經使用的節(jié)點個數
unsigned long used;
} dictht;
真實存儲數據的鏈表節(jié)點的定義如下:
typedef struct dictEntry {
// 存儲的鍵值對的 key
void *key;
// 存儲的鍵值對的 value
union {
void *val;
uint64_t u64;
int64_t s64;
double d;
} v;
// 指向下一個節(jié)點
struct dictEntry *next;
} dictEntry;
key + v(value) + next 一個簡單的鏈表定義。
有點特殊的就是對應著 value 屬性的 v 的定義是一個聯(lián)合體, 會在不同場景下使用不同的字段,
比如一個鍵值對的過期時間就存放在 s64 中, 這個 value 存放的值就放在 val 中。
一個 dictEntry 的字段存放內容大體如下:
4.2.2 字典
在使用 HashTable 時, 都需要提前聲明好容量, 而隨著程序的運行, 存放到 HashTable 的數據會越來越多, 最終達到上限, 這時就需要進行擴容了。
在 Java 的 HashMap 的擴容過程
創(chuàng)建一個更大容量的數組
將 HashMap 中舊數組一次性遷移到新的數組中
清除掉舊數組
這個擴容沒多大問題, 但是放到 Redis 中合適嗎?
Redis 是一個存內存的數據庫, 所有的數據都存放在內存中, 基本是 GB 級別的數據量, 每次擴容遷移的數據量很多
Redis 是一個單線程的數據庫, 一次只能處理一個事情, 如果全力在做擴容, 那么其他的請求將無法處理
所以 Redis 采用了一種 漸進式 rehash 的方法解決擴容縮容的問題, 過程如下
維護 2 個 dictht, 一個是真實存儲數據的 HashTable A, 一個是擴容后存儲數據的 TableTable B + 一個 rehash 位置的索引, 初始值為 0
在 rehash >=0 期間, 每次對 HashTable 進行操作, 除了正常的操作外, 還會將 A rehash 位置的數據都遷移到 B, 然后 rehash + 1
隨著對 HashTable 的不斷操作, 最終 A 中的數據都會遷移到 B, 這時將 rehash 設置為 -1
基于上面的漸進式 rehash 分析, 實際是需要 2 個 dictht, 所以 Redis 在此至上多封裝了一層
typedef struct dict {
dictType *type;
void *privdata;
dictht ht[2]; // 2 個 HashTable
long rehashidx; // rehash 的索引
unsigned long iterators;
} dict;
這個就是 Redis 中的字典, 用于存儲鍵值對的結構。
在將這個結構放到一個 redisDb 就是我們常見的 Redis 數據庫了
typedef struct redisDb {
dict *dict;
dict *expires;
....
} redisDb;
redisDb 就是我們常說的 Redis 16 個數據庫的定義了。 每個數據庫中都有 2 個字典
dict 正常的字典, 存儲沒有設置過期時間的鍵值對
expires 過期字典, 存儲設置了過期時間的鍵值對
4.2 Redis 的內存待淘汰池
struct evictionPoolEntry {
unsigned long long idle; // 對象空閑時間 (使用的算法是 LFU 則是逆頻率)
sds key; // 待淘汰的鍵值對的 key
sds cached; // 緩存的 key 名稱 SDS 對象
int dbid; // 待淘汰鍵值對的 key 所在的數據庫 ID
};
5 參考
到此這篇關于深入理解Redis內存回收和內存淘汰機制的文章就介紹到這了,更多相關Redis內存回收和內存淘汰機制內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
這篇文章主要給大家介紹了關于redis實現加鎖的幾種方法，加鎖命令分別是INCR、SETNX和SET，文中給出了詳細的示例代碼，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧。2017-09-09
經常會有人問這個問題，redis中為什么要使用跳表?這個問題，redis作者已經給出過明確答案,今天通過本文再給大家講解下這個問題，對redis?skiplist跳表知識感興趣的朋友一起看看吧2022-10-10
這篇文章主要為大家詳細的介紹了Redis高效的原因以及分析了Redis高效的數據結構，有需要的朋友可以借鑒參考下，希望能夠有所幫助2021-09-09
這篇文章主要介紹了如何用redis?setNX命令來加鎖，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2023-01-01
布隆過濾器（Bloom?Filter）是Redis?4.0版本提供的新功能，它被作為插件加載到Redis服務器中，給Redis提供強大的去重功能，本文將給大家詳細介紹一下Redis布隆過濾器，文中有相關的代碼示例,需要的朋友可以參考下2023-07-07
讓Redis在你的系統(tǒng)中發(fā)揮更大作用的幾點建議
Redis在很多方面與其他數據庫解決方案不同：它使用內存提供主存儲支持，而僅使用硬盤做持久性的存儲；它的數據模型非常獨特，用的是單線程。另一個大區(qū)別在于，你可以在開發(fā)環(huán)境中使用Redis的功能，但卻不需要轉到Redis2014-06-06
本文主要介紹了Redis分布式鎖Redlock的實現，文中通過示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2021-08-08
這篇文章主要為大家詳細介紹了Linux下Redis安裝使用教程，具有一定的參考價值，感興趣的小伙伴們可以參考一下2018-09-09
這篇文章主要為大家介紹了Redis過期數據會被立馬刪除么的問題解析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2022-07-07
本文主要介紹了Redisson 加鎖解鎖的實現，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2022-08-08