redis必會基礎命令、資料結構、lua腳本和分布式鎖等

導讀

本文介紹日常作業中redis的使用，涉及到redis的資料結構、對應的命令、持久化配置和Lua腳本，以及基于redis的分布式鎖實作方案，使用redis時這些都是必會的基礎知識，建議保存以下命令

通用命令

# 查看當前庫中key的數量
dbsize

# 清空當前庫
flushdb

# 清空所有庫
flushall

# 查看當前庫下所有key
keys *

# 當前庫下是否有指定key
exists key1

# 查看key的值型別
type key1

# 洗掉key
del key1

# 設定指定key的過期時間，單位秒
expire key1 10

# 查看key的剩余過期時間，-1表示永不過期，-2表示已過期
ttl key1

# 監視key
watch key1

# 取消監視key
unwatch key1
復制代碼

資料結構和命令

string

是key/value的資料結構，一個key對應一個string型別的value，單個value最大512M

這個是最常用的資料結構了

命令

# 設定key/value
set key1 value1

# 設定key/value的同時設定過期時間
setex key1 10 value1

# 獲取key的值
get key1

# 設定新值，同時回傳舊值
getset key1 value1

# 一次設定多個key/value，后面的....表示還可以寫key3 value3等
mset key1 value1 key2 value2 ...

# 一次獲取多個key的值
mget key1 key2

# 追加內容到指定key的值后面
append key1 xxx

# 獲取值的長度
strlen key1

# 只有在key不存在時才成功
setnx key1 value1

# 只有在所有key不存在時才成功
msetnx key1 value1 key2 value2 ...

# 給指定key的值加1
incr key1

# 減1
decr key1

# 給指定key的值加指定數值，本例是加2
incrby key1 2

# 給指定key的值減指定數值，本例是減2
decrby key1 2

# 獲取指定key的值中指定范圍的字符，如值為abcdefg，取1至2回傳bc，即包含1和2兩個位置的字符
getrange key1 1 2

# 設定指定位置的值，指定開始位置，然后直接覆寫，如下例中值為abcdefg，從第1個位置開始覆寫為cb，則結果為acbdefg
setrange key1 1 cb
復制代碼

list

雙向鏈表，無序可重復的集合，一般用來做佇列

命令

# 從表頭添加元素，value2是新的表頭
lpush key1 value1 value2 ...

# 從表尾添加元素，value2是新的表尾
rpush key1 value1 value2 ...

# 從表頭彈出元素
lpop key1

# 從表尾彈出元素
rpop key1

# 從key1表尾彈出一個元素，再加到key2表頭
rpoplpush key1 key2

# 從表中查看指定索引的范圍的元素
lrange key1 0 2

# 查看整個鏈表
lrange key1 1 0 -1

# 獲取鏈表中從左向右指定索引的元素
lindex key1 1

# 獲取鏈表中最后一個元素
lindex key1 -1

# 獲取鏈表長度
llen key1

# 向鏈表中的value1前面插入value2
linsert key1 before value1 value2

# 向鏈表中value1后面插入value2
linsert key1 after value1 value2

# 從鏈表中洗掉一個值為value1的元素，從左向右
lrem key1 1 value1

# 從鏈表中洗掉一個值為value1的元素，從右向左
lrem key1 -1 value1

# 洗掉鏈表中所有值為value1的元素
lrem key1 0 value1
復制代碼

set

無序不可重復的集合，常用來排除重復資料和隨機抽獎功能

命令

# 向集合中添加元素，重復元素會自動跳過
sadd key1 value1 value2 ...

# 取出集合所有元素
smembers key1

# 判斷集合中是否存在某個元素
sismember key1 value1

# 獲取集合中的元素個數
scard key1

# 從集合中洗掉指定元素
srem key1 value1 value2 ...

# 隨機從集合中彈出一個元素并洗掉該元素
spop key1

# 隨機從集合中取出元素，但不會洗掉元素，后面的1表示取出元素的個數
srandmember key1 1

# 求兩個集合交集
sinter key1 key2

# 求兩個集合并集
sunion key1 key2

# 求兩個集合差集
sdiff key1 key2
復制代碼

zset

有序不可重復的集合，常用來做排行榜

命令

# 添加元素，相同value不同score會覆寫score
zadd key1 score1 value1 score2 value2

# 獲取元素數量
zcard key1

# 取出全部元素，從小到大
zrange key1 0 -1

# 取出部分元素，從小到大
zrange key1 0 4

# 取出全部元素，從大到小
zrevrange key1 0 -1

# 取出部分元素，從大到小
zrevrange key1 0 4

# 取出score在指定范圍內的元素，從小到大，其中min和max是score的范圍
zrangebyscore key1 min max withscores

# 取出score在指定范圍內的元素，從大到小
zrevrangebyscore key1 max min withscores

# 為指定value的元素的score遞增，其中1是每次遞增多少，可以為負數
zincrby key1 1 value1

# 洗掉指定元素
zrem key1 value1

# 統計集合中score在范圍內的元素個數
zcount key1 min max

# 回傳指定值在集合中的排名，從小到大，排名從0開始
zrank key1 value1

# 回傳指定值在集合中的排名，從大到小
zrevrank key1 value1
復制代碼

hash

類似于Java中的Map<String, String>

命令

# 添加一個鍵值對
hset key1 field1 value1

# 獲取鍵值
hget key1 field1

# 批量設定鍵值對
hmset key1 field1 value1 field2 value2 ...

# 檢查鍵是否存在
hexists key1 field1

# 獲取所有鍵
hkeys key1

# 獲取所有值
hvals key1

# 鍵值遞增，后面的1表示每次遞增多少，可以為負數，當是負數時表示遞減
hincrby key1 field1 1

# 鍵不存在時成功
hsetnx key1 field1 value1

# 獲取所有鍵值對，奇數為鍵，偶數為值
hgetall key1
復制代碼

bitmap

bitmap以bit為單位設定各個位的值（要么是0，要么是1），根據實際應用場景可以設計出節省空間的演算法，如布隆過濾器，本文以記錄用戶簽到為例，假設用戶ID為1，每年一個key，并且key=用戶ID_年份，如1_2021

ID=1的用戶在2021-01-01這一天簽到，這一天是2021年第1天（也就是第0天），可以執行以下命令，保存簽到記錄

# 設定1_2021這個key的第0個bit值為1，以此表示第0天簽到成功
setbit 1_2021 0 1
復制代碼

該用戶在2021-01-03這一天簽到，則執行以下命令

# 設定1_2021這個key的第2個bit值為1，以此表示第2天簽到成功
setbit 1_2021 2 1
復制代碼

現在想查詢該用戶在2021年的簽到情況，可通過get命令實作

get 1_2021
# 輸出\xa0
復制代碼

get命令輸出0xa0，這是十六進制，轉成二進制，就是10100000，二進制中為1的位就表示那一天簽到了，所以第0天和第2天簽到了

判斷該用戶2021-01-03是否簽到

getbit 1_2021 2
復制代碼

統計該用戶2021年有多少次簽到，實際上是統計有多個位是1

bitcount 1_2021
# 輸出2
復制代碼

統計該用戶指定日期范圍內的簽到次數，這個不太好實作，redis提供的命令中指定范圍的單位是byte，比如統計2021年1月的次數，就是第0個位元組到第3個位元組（第0 1 2 3共4個位元組），這樣多統計了1天，即把2021-02-01這一天也統計進來，如下：

# 統計第0到第3個位元組中為1的位個數，包含第3個位元組
bitcount 1_2021 0 3
復制代碼

這種情況要么按月來設定key值，要么單獨查詢2021-02-01這一天是否簽到，如果簽到則總次數就減1

通過上面的例子，可以看到以bit為單位存盤非常節省空間，用8個bit就可以表示8天內的簽到情況，也可以用bitmap來存盤所有用戶一天內的簽到情況，這種就以用戶ID作為bit的偏移量，如果用戶ID很大，超過了bitmap的最大范圍，可以通過用戶ID分片到不同的bitmap上

地理位置

在同一個key內添加多個位置（經緯度），計算位置各個位置之間的距離，也可指定圓心按半徑查找符合條件的位置，可實作附近的xxx功能

命令

# 向key1添加一個叫company的位置，經緯度為116.404844 39.915378
geoadd key1 116.404844 39.915378 company

# 向key1添加一個叫home的位置，經緯度為116.370924 39.930871
geoadd key1 116.370924 39.930871 home

# 查詢指定位置的經緯度
geopos key1 company

# 查詢多個位置的經緯度
geopos key1 company home

# 計算兩個位置的距離，單位是m
geodist key1 company home

# 計算兩個位置的距離，指定單位為km
geodist key1 company home km

# 以指定經緯度為圓心，查詢指定半徑內的所有位置，其中116.370924 39.930871是圓心點的經緯度，2000 m是半徑大小，單位m，withdist表示輸出符合條件的位置與圓心的距離，withcoord表示輸出符合條件的位置的經緯度，asc表示按距離從小到大排序
georadius key1 116.370924 39.930871 2000 m withdist withcoord asc

# 以指定位置為圓心，查詢指定半徑內的所有位置，回傳結果中包含圓心自身，其他可選引數與上一條georadius相同
georadiusbymember key1 home 4000 m
復制代碼

持久化

redis大部分時間用來做快取，通常資料丟失也可以恢復，但是有時候也會用來存盤熱門的資料，或者nginx直接連接redis做一些重要資料的存盤（丟失后很難恢復），所以redis中的資料需要持久化

redis提供了RDB和AOF兩種持久化方式，RDB是對當前資料的全量備份（理解成快照），AOF采用追加的方式記錄所有寫入的命令，所以一般AOF檔案更大，可能導致硬碟被占滿，這一點需要注意，需要及時的對AOF檔案進行瘦身

RDB

執行save和bgsave命令會生成一份當前記憶體資料的快照到.rdb檔案內，其中bgsave命令是另起一個執行緒去執行，因此不會阻塞主執行緒

redis默認開啟了RDB，redis會自動進行RDB存盤，RDB常用配置引數：

save 300 1000 # 每隔300秒，如果有1000個key發生了變化，則備份一次
save 30 10000 # 每隔30秒，如果有10000個key發生了變化，則備份一次
復制代碼

上面的引數不能亂改，要根據redis的寫入資料情況來設定，不能太頻繁的生成RDB快照，這會影響redis的性能

AOF

一般做持久化要同時開啟RDB和AOF，下面介紹作業中如何設定redis的AOF，編輯redis.conf組態檔，修改以下配置項

# 開啟AOF
appendonly yes

# AOF檔案名
appendfilename "xxx.aof"

# 將命令刷入磁盤的間隔，everysec是每秒一次
appendfsync everysec

# 在執行bgrewrite的時候不向磁盤重繪資料
no-appendfsync-on-rewrite no
復制代碼

關于appendfsync的意思：在計算機組成原理中，我們知道相對于記憶體而言對磁盤的讀寫是很慢的，所以CPU將資料寫入記憶體緩沖區，定時或存滿后再寫入磁盤，redis的AOF中appendfsync這個配置就是設定多久寫入磁盤一次，設為一秒是比較保險的，如果發生故障只會丟失最近1秒內的資料

RDB和AOF對比

1. RDB定期對記憶體中的資料進行快照，會影響redis的性能，所以不能太頻繁
2. RDB在快照之間如何發生錯誤會丟失此段時間內的資料
3. RDB在重啟redis時恢復速度更快，不像AOF那樣需要一條一條命令執行
4. AOF可設定每秒追加一次寫入命令到aof檔案中，所以發生故障時丟失資料最少
5. AOF檔案中保存的是redis的寫入命令，所以可以打開檔案進行修改，洗掉不需要的命令
6. AOF的缺點是要記錄redis做的每一步寫入命令，所以檔案很大，需要及時進行瘦身

lua腳本

介紹

redis中默認就支持lua腳本，我們通常會使用lua腳本來代替redis事務，可解決超賣和少賣的情況

以下是redis的lua腳本特性：

1. 原子操作，lua腳本會作為一個整體執行，不會被其他連接的命令中斷，因此可替代事務
2. lua腳本加載后可重復使用
3. 減少網路請求的開銷，一次性發出一個lua腳本到redis，redis執行完后回傳結果，不用多次請求

用法

直接執行

eval 腳本 引數數量 引數名1 引數名2 值1 值2
復制代碼

示例：

eval "return KEYS[1]..ARGV[1]" 1 key1 val1
# 輸出
key1val1
復制代碼

lua腳本中KEYS和ARGV兩個陣列名是固定的，且索引從1開始，上面例子中引數數量為1，引數名1為key1，值1為val1，而腳本是拼接key1和val1，所以結果就是key1val1了

注意：如果腳本有多個引數，則引數名寫在一起，引數值寫在一起，如key1 key2 val1 val2，而不是key1 val1 key2 val2

加載腳本

加載腳本的目的是重復使用，通過script load命令實作，回傳一個sha1的hash值，之后通過此值可呼叫已加載的腳本

script load "return KEYS[1]..ARGV[1]"
# 假設回傳3783a90bf1f43b15a1e06c4e7664da956ed959d9
復制代碼

呼叫腳本

# 3783a90bf1f43b15a1e06c4e7664da956ed959d9 是script load回傳的結果
# 1 是引數數量
# key1 是引數名1
# val1 是引數值1
evalsha 3783a90bf1f43b15a1e06c4e7664da956ed959d9 1 key1 val1
復制代碼

指定加載腳本回傳的sha1 hash值來呼叫腳本，并指定引數數量和引數名以及引數值

判斷腳本是否加載

script exists 3783a90bf1f43b15a1e06c4e7664da956ed959d9
復制代碼

Java中使用redis lua腳本

本例使用spring提供的RedisTemplate，首先引入依賴

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
復制代碼

定義一個RedisClient，對常用命令進行封裝

@Component
public class RedisClient {
    @Autowired
    @Qualifier("redisTemplate")
    private RedisTemplate redisTemplate;

    /**
     * 執行腳本
     * @param clazz 回傳結果型別
     * @param script lua腳本
     * @param keys lua腳本的引數名
     * @param args lua腳本的值
     * @param <T>
     * @return
     */
    public <T> T execute(Class<T> clazz, String script, List<Object> keys, Object... args) {
        DefaultRedisScript<T> redisScript = new DefaultRedisScript<>();
        redisScript.setResultType(clazz);
        redisScript.setScriptText(script);

        return (T) redisTemplate.execute(redisScript, keys, args);
    }
}
復制代碼

測驗使用，先在redis中set兩個key出來，分別是test_key1和test_key2，以下lua腳本是對這兩個key進行遞增操作，分別遞增1和2

String script = "redis.call(\"INCRBY\", KEYS[1], ARGV[1])\nredis.call(\"INCRBY\", KEYS[2], ARGV[2])";

ArrayList<Object> keys = Lists.newArrayList("test_key1", "test_key2");
Object execute = redisClient.execute(Object.class, script, keys, 1, 2);
System.out.println(execute);
復制代碼

分布式鎖

redis有個setnx命令，在key不存在時才能設定成功，因此也經常用來實作分布式鎖，但是只通過setnx命令來做分布式鎖是不安全的，假設執行緒1執行setnx成功并設定10秒后過期，執行緒2再執行setnx命令肯定是失敗的，如果執行緒1執行程序中發生故障沒有及時清除鎖，則其他執行緒只能等待10秒后才能獲取鎖；如果執行緒1在10秒內還沒有執行完，由于鎖已經過期，導致其他執行緒執行setnx成功，這就存在兩個執行緒同時拿到鎖，這樣的使用方式肯定是不行的，今天介紹第三方庫redisson，redisson幫我們搞定了以上兩種情況需要解決的問題

先引入依賴

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
</dependency>
復制代碼

定義一個工具類，方便使用

@Component
public class RedissonUtils {
    @Autowired
    private RedissonClient redissonClient;

    private static RedissonClient client;

    @PostConstruct
    private void init() {
        client = redissonClient;
    }

    /**
     * 通過指定key加鎖
     * @param key 指定key
     * @return
     */
    public static RLock lock(String key) {
        RLock lock = client.getLock(key);
        lock.lock();
        return lock;
    }

    /**
     * 通過指定key加鎖指定時間
     * @param key 指定key
     * @param expire 鎖定時間，默認單位：毫秒
     * @return
     */
    public static RLock lock(String key, long expire) {
        RLock lock = client.getLock(key);
        lock.lock(expire, TimeUnit.MILLISECONDS);
        return lock;
    }

    /**
     * 通過指定key加鎖指定時間
     * @param key 指定key
     * @param expire 鎖定時間
     * @param unit 鎖定時間單位
     * @return
     */
    public static RLock lock(String key, long expire, TimeUnit unit) {
        RLock lock = client.getLock(key);
        lock.lock(expire, unit);
        return lock;
    }

    /**
     * 通過指定key嘗試加鎖指定時間，如果加鎖失敗等待指定時間
     * @param key 指定key
     * @param wait 加鎖等待時間
     * @param expire 鎖定時間
     * @param unit 鎖定時間單位
     * @return
     */
    public static RLock tryLock(String key, long wait, long expire, TimeUnit unit) {
        RLock lock = client.getLock(key);
        try {
            if (lock.tryLock(wait, expire, unit)) {
                return lock;
            } else {
                return null;
            }
        } catch (InterruptedException e) {
            return null;
        }
    }

    /**
     * 通過指定key解鎖
     * @param key
     */
    public static void unlock(String key) {
        RLock lock = client.getLock(key);
        lock.unlock();
    }

    /**
     * 通過指定鎖解鎖
     * @param lock
     */
    public static void unlock(RLock lock) {
        if (lock != null) {
            lock.unlock();
        }
    }
}
復制代碼

獲取鎖

RLock lock = RedissonUtils.tryLock(LockKey.XXX, 5L, 10L, TimeUnit.SECONDS);
Assert.isNull(lock, "獲取鎖失敗");

// 取到鎖

try {
    // 拿到鎖后執行的操作
    ...
} finally {
    RedissonUtils.unlock(lock);
}
復制代碼

需要注意的是釋放鎖一定要放到finally中，防止發生例外而未釋放鎖

最后說明一下分布式鎖也不是非要使用redisson，使用redis實作分布式鎖有個致命的問題，當拿到鎖后主redis還未同步到從redis時，主redis掛掉，系統切換到從redis，此時其他執行緒仍然可以拿到鎖，這樣系統中就有兩個執行緒拿到了鎖