前言
Redis 當中提供了許多重要的高級特性，比如發布與訂閱，Lua 腳本等。Redis 當中也提供了自增的原子命令，但是假如我們需要同時執行好幾個命令的同時又想讓這些命令保持原子性，該怎么辦呢？這時候就可以使用本文介紹的 Lua 腳本來實現。
發布與訂閱
發布與訂閱功能理論上來說可以直接通過一個雙端鏈表就可以實現了，然而這種通過普通的雙端鏈表來實現的發布與訂閱功能有兩個局限性：
- 如果生產者生產消息的速度遠大於消費者消費消息的速度，那么鏈表中未消費的消息會大量堆積，導致占用大量的內存。
- 基於鏈表實現的消息隊列，不支持一對多的消息分發。
因為普通雙端鏈表來實現發布與訂閱功能有這兩個局限性，故而 Redis 當中並沒有直接通過雙端列表來實現。在 Redis 中的發布與訂閱可以分為兩種類型：基於頻道和基於模式。
基於頻道的實現
基於頻道的實現方式主要通過以下三個命令：
- subscribe channel-1 channel-2：訂閱一個或者多個頻道。
- unsubscribe channel-1：取消頻道的訂閱（命令操作界面上無法退訂）。
- publish channel-1 message：向頻道
channel-1發送消息message。
打開一個客戶端 一，輸入訂閱命令 subscribe music movie，表示當前客戶端訂閱了 music 和 movie 兩個頻道的消息：
然后再打開一個客戶端二，執行以下發布消息命令：
publish movie myCountry //發布消息 myCountry 到 movie 頻道
publish music love //發布消息 love 到 music 頻道
publish tv myHome //發布消息 myHome 到 tv 頻道
前面兩個頻道發布之后返回 1 就表示當前有 1 個客戶端訂閱了該頻道，消息已經發送到這個客戶端。
這時候我們再回到之前的客戶端一，就會發現客戶端一收到了消息 myCountry 和 love 兩條消息，而 myHome 這條消息是屬於頻道 tv，客戶端一並沒有訂閱，故而不會收到：
同時，還有以下 2 個命令可以查看當前客戶端訂閱的頻道信息：
- punsub channels [channel_name] ：查看當前服務器被訂閱的頻道。不帶參數則返回所有頻道，后面的參數可以使用通配符
?或者*。 - pubsub numsub channel_name [channel_name]：查看指定頻道的訂閱數（可同時查看多個）。
實現原理分析
客戶端與其訂閱的頻道信息被保存在 redisServer 對象中的 pubsub_channels 屬性中。
struct redisServer {
dict *pubsub_channels;//保存了客戶端及其訂閱的頻道信息
//... 省略其他信息
};
pubsub_channels 屬性是一個字典，其 key 值保存的就是頻道名，value 是一個鏈表，鏈表中保存的就是每個客戶端的 id，下圖就是基於頻道訂閱的存儲結構示意圖：
- 訂閱
訂閱的時候首先會檢查字典內是否存在這個頻道：如果不存在，則需要為當前頻道創建一個字典，同時創建一個鏈表作為value，並將當前客戶端id放入鏈表；如果存在，則直接將當前客戶端id放入鏈表末尾即可。 - 取消訂閱
取消訂閱的時候需要將客戶端id從對應的鏈表中移除，如果移除之后鏈表為空，則需要同時將該頻道從字典內刪除。 - 發送消息
發送消息時首先會去pubsub_channels字典內尋找鍵，如果發現有可以匹配上的鍵，則會找到對應的鏈表，進行遍歷發送消息。
基於模式的實現
基於模式的發布與訂閱實現方式主要通過以下三個命令：
- psubscribe pattern-1 pattern-2：訂閱一個或者多個模式，模式可以通過通配符
?和*來表示。 - punsubscribe pattern-1 pattern-1：取消模式的訂閱（基於命令操作，界面上無法退訂）
- publish channel-1 message ：向頻道
channel-1發送消息message。這里和上面基於頻道命令是一樣的。
打開一個客戶端 一，輸入訂閱命令 psubscribe m*，表示當前客戶端訂閱了所有以 m 開頭的頻道：
然后再打開一個客戶端二，執行一下發布消息命令：
publish movie myCountry //發布消息 myCountry 到 movie 頻道
publish music love //發布消息 love 到 music 頻道
publish tv myHome //發布消息 myHome 到 tv 頻道
前面兩個頻道發布之后返回 1 就表示當前有 1 個客戶端訂閱了該頻道（上面基於頻道訂閱的客戶端關閉之后會自動取消訂閱），消息已經發送到這個客戶端。
這時候我們再回到之前的客戶端一，就會發現客戶端一收到了 myCountry 和 love 兩條消息，因為這兩個頻道都是以 m 開頭的，而 myHome 這條消息是屬於頻道 tv，並不是以 m 開頭，客戶端一並沒有訂閱，故而不會收到：
同樣的，基於模式的訂閱也提供了一個查詢命令：
- pubsub numpat：查詢當前服務器被訂閱模式的數量。
實現原理分析
客戶端與其訂閱的模式信息被保存在 redisServer 對象中的 pubsub_patterns 屬性中。
struct redisServer {
list pubsub_patterns;//保存了客戶端及其訂閱的模式信息
//...省略其他信息
};
pubsub_patterns 屬性是一個列表，其列表內結構（源碼 serer.h 內）定義如下：
typedef struct pubsubPattern {
client *client;//訂閱模式的客戶端
robj *pattern;//被訂閱的模式
} pubsubPattern;
- 訂閱
新建一個pubsubPattern數據結構加入到鏈表pubsub_patterns的結尾。 - 取消訂閱
從鏈表中將當前取消訂閱的客戶端pubsubPattern從鏈表pubsub_patterns中移除 - 發送消息
此時需要遍歷整個鏈表來尋找能匹配的模式。之所以基於模式場景使用鏈表是因為模式支持通配符，所以沒有辦法直接用字典實現。
PS：當基於頻道和基於模式兩種訂閱同時都存在時，Redis 會先去尋找頻道字典，再去遍歷模式鏈表進行消息發送。
Lua 腳本
Redis 從 2.6 版本開始支持 Lua 腳本，為了支持 Lua 腳本，Redis 在服務器中嵌入了 Lua 環境。
使用 Lua 腳本最大的好處是 Redis 會將整個腳本作為一個整體執行，不會被其他請求打斷，可以保持原子性且減少了網絡開銷。
Lua 腳本的調用
Lua 腳本的執行語法如下：
eval lua-script numkeys key [key ...] arg [arg ...]
- eval：執行
Lua腳本的命令。 - lua-script：
Lua腳本內容。 - numkeys：表示的是
Lua腳本中需要用到多少個key，如果沒用到則寫0。 - key [key ...]：將
key作為參數按順序傳遞到Lua腳本，numkeys是0時則可省略。 - arg：
Lua腳本中用到的參數，如果沒有可省略。
接下來我們執行一個不帶任何參數的簡單 Lua 腳本命令：
eval "return 'Hello Redis'" 0
Lua 腳本中執行 Redis 命令
在 Lua 腳本中執行 Redis 命令時需要使用以下語法：
redis.call(command, key [key ...] argv [argv…])
- command：
Redis中的命令，如set、get等。 - key：操作
Redis中的key值，相當於我們調用方法時的形參。 - param：代表參數，相當於我們調用方法時的實參。
假如我們想執行一個命令 set name lonely_wolf，那么利用 Lua 腳本則應該這么執行：
eval "return redis.call('set',KEYS[1],ARGV[1])" 1 name lonely_wolf
需要注意的是：KEYS 和 ARGV 必須要大寫，參數的下標從 1 開始。上面命令中 1 表示當前需要傳遞 1 個 key
Lua 腳本摘要
有時候如果我們執行的一個 Lua 腳本很長的話，那么直接這么調用 Lua 腳本的話非常不方便，所以 Redis 當中提供了一個命令 script load 來手動給每一個 Lua 腳本生成摘要，這里之所以要說手動的原因是即使我們不使用這個命令，每次調用完 Lua 腳本的時候，Redis 也會為每個 Lua 腳本生成一個摘要。
其他相關命令：
script exists 摘要：判斷一個摘要是否存在。0表示不存在，1表示存在。script flush：清除所有Lua腳本緩存。
接下來我們來驗證一下，依次執行以下命令：
script load "return redis.call('set',KEYS[1],ARGV[1])" //給當前 Lua腳本生成摘要，這時候會返回一個摘要
evalsha "c686f316aaf1eb01d5a4de1b0b63cd233010e63d" 1 address china //相當於執行命令 set address china
get address //獲取 adress，確認上面的腳本是否執行成功
script exists "c686f316aaf1eb01d5a4de1b0b63cd233010e63d" //判斷當前摘要的 Lua腳本是否存在
script flush //清除所有 Lua腳本緩存
script exists "c686f316aaf1eb01d5a4de1b0b63cd233010e63d" //清除之后這里就不存在了
執行之后得到如下效果：
Lua 腳本文件
當我們的 Lua 腳本很長時，直接在命令窗口中寫腳本是不直觀的，也很難發現語法問題，所以 Redis 當中也支持我們直接把先把腳本寫入文件中，然后直接調用文件。
比如我們新建一個 test.lua 腳本：
redis.call('set',KEYS[1],ARGV[1])
return redis.call('get',KEYS[1])
將文件上傳到指定目錄之后，執行如下命令：
redis-cli --eval test.lua 1 age , 18 //注意 key 和 arg 參數之間要以逗號隔開，且逗號兩邊的空格不能省略
這時候就可以正常返回 18：
腳本異常
我們知道，Redis 的指令是單線程執行的，而現在使用了 Lua 腳本，我們就可以通過 Lua 腳本來實現一些業務邏輯，那么如果 Lua 腳本執行超時或者陷入了死循環，這個時候其他的指令就會被阻塞，導致 Redis 無法正常使用。這個時候應該如何處理呢？
腳本超時
為了解決 Lua 腳本超時的問題，Redis 提供了一個超時時間的參數 lua-time-limit 來控制 Lua 腳本執行的超時時間，單位是毫秒，默認是 5000 （即 5 秒），到達超時時間之后 Lua 會自動中斷腳本。
腳本陷入死循環
假如腳本陷入了死循環，這時候超時時間就不起作用了，我們來模擬一下：
首先打開客戶端一，執行一個死循環的 lua 腳本：
eval 'while(true) do end' 0
然后打開另一個客戶端二，任意執行一個命令：
get name
這時候會返回 busy，表示當前無法執行這個命令：
提示 busy 之后，同時 Redis 也給出了解決方案，我們只能只用 script kill 或者 shutdown nosave 命令，這兩個命令又是做什么用的呢？
- script kill：當腳本陷入死循環之后，執行這個命令可以強制
Lua腳本中斷執行。這個腳本的局限性就是當前陷入死循環的Lua腳本必須沒有成功執行過命令。 - shutdown nosave：強制退出
Lua腳本，可以解決script kill命令的局限性。
接下來讓我們在客戶端二執行命令 script kill，然后再去看看陷入死循環的客戶端一的效果：
可以看到，客戶端一的 Lua 腳本已經退出了，根據后面的提示可以知道就是因為執行了 script kill 命令而導致了 Lua 腳本的中斷。
現在我們重新用客戶端一執行下面這個 Lua 腳本，這個腳本和上面的腳本區別就是這里執行成功了一個 Redis 命令之后才開始死循環：
eval "redis.call('set','age','28') while true do end" 0
這時候再去客戶端二執行 script kill 命令，發現無法中止 Lua 腳本了：
這里不允許直接中斷 Lua 腳本是因為在死循環前已經有 Redis 命令被成功執行了，如果直接中斷，那么就會造成數據不一致問題。
在這種場景下，只能通過執行 shutdown nosave 命令來強行中斷 Lua 腳本，這里因為加了 nosave 之后不會觸發 Redis 的持久化，所以當重啟 Redis 服務之后，可以保證數據的一致性，下圖就是執行 shutdown nosave 命令之后客戶端一的效果圖：
為什么可以執行 script kill 命令
Redis 當中執行命令是單線程的，那么為什么 Lua 腳本陷入死循環之后其他客戶端還可以執行 script kill 命令呢？
這是因為 Lua 腳本引擎提供了鈎子（hook）函數，它允許在內部虛擬機執行指令時運行鈎子代碼，所以 Redis 正是利用了這一原理，在執行 Lua 腳本之前設置了一個鈎子，也就是說 script kill 命令是通過鈎子（hook）函數來執行的。
總結
本文主要介紹了 Redis 中的發布訂閱功能和 Lua 腳本的使用，使用 Lua 腳本可以讓多個命令原子執行，減少網絡開銷，但是同時也要注意 Lua 腳本引發的死循環問題。