redis實現隊列queue

參考：《Redis入門指南》第4章進階

http://book.51cto.com/art/201305/395461.htm

4.4.2　使用Redis實現任務隊列

說到隊列很天然就能想到Redis的列表類型，3.4.2節介紹了使用LPUSH和RPOP命令實現隊列的概念。若是要實現任務隊列，只須要讓生產者將任務使用LPUSH命令加入到某個鍵中，另外一邊讓消費者不斷地使用RPOP命令從該鍵中取出任務便可。

在小白的例子中，完成發郵件的任務須要知道收件地址、郵件主題和郵件正文。因此生產者須要將這三個信息組成對象並序列化成字符串，而後將其加入到任務隊列中。而消費者則循環從隊列中拉取任務，就像以下僞代碼：

 

# 無限循環讀取任務隊列中的內容
loop
    $task = RPOR queue
    if $task
         # 若是任務隊列中有任務則執行它
        execute($task)
    else
         # 若是沒有則等待1秒以避免過於頻繁地請求數據
        wait 1 second

到此一個使用Redis實現的簡單的任務隊列就寫好了。不過還有一點不完美的地方：當任務隊列中沒有任務時消費者每秒都會調用一次RPOP命令查看是否有新任務。若是能夠實現一旦有新任務加入任務隊列就通知消費者就行了。其實藉助BRPOP命令就能夠實現這樣的需求。

BRPOP命令和RPOP命令類似，惟一的區別是當列表中沒有元素時BRPOP命令會一直阻塞住鏈接，直到有新元素加入。如上段代碼可改寫爲：

 

loop
    # 若是任務隊列中沒有新任務，BRPOP命令會一直阻塞，不會執行execute()。
    $task = BRPOP queue, 0
    # 返回值是一個數組（見下介紹），數組第二個元素是咱們須要的任務。
    execute($task[1])

BRPOP命令接收兩個參數，第一個是鍵名，第二個是超時時間，單位是秒。當超過了此時間仍然沒有得到新元素的話就會返回nil。上例中超時時間爲"0"，表示不限制等待的時間，即若是沒有新元素加入列表就會永遠阻塞下去。

當得到一個元素後BRPOP命令返回兩個值，分別是鍵名和元素值。爲了測試BRPOP命令，咱們能夠打開兩個redis-cli實例，在實例A中：

 

redis A> BRPOP queue 0

brpop:

返回值：

假如在指定時間內沒有任何元素被彈出，則返回一個  nil 和等待時長。

反之，返回一個含有兩個元素的列表，第一個元素是被彈出元素所屬的  key ，第二個元素是被彈出元素的值。

 

鍵入回車後實例1會處於阻塞狀態，這時在實例B中向queue中加入一個元素：

 

redis B> LPUSH queue task
(integer) 1

在LPUSH命令執行後實例A立刻就返回告終果：

 

1) "queue"
2) "task"

同時會發現queue中的元素已經被取走：

 

redis> LLEN queue
(integer) 0

除了BRPOP命令外，Redis還提供了BLPOP，和BRPOP的區別在與從隊列取元素時BLPOP會從隊列左邊取。具體能夠參照LPOP理解，這裏再也不贅述。

4.4.3　優先級隊列

前面說到了小白博客須要在發佈文章的時候向每一個訂閱者發送郵件，這一步驟一樣可使用任務隊列實現。因爲要執行的任務和發送確認郵件同樣，因此兩者能夠共用一個消費者。然而設想這樣的狀況：假設訂閱小白博客的用戶有1000人，那麼當發佈一篇新文章後博客就會向任務隊列中添加1000個發送通知郵件的任務。若是每發一封郵件須要10秒，所有完成這1000個任務就須要近3個小時。問題來了，假如這期間有新的用戶想要訂閱小白博客，當他提交完本身的郵箱並看到網頁提示他查收確認郵件時，他並不知道向本身發送確認郵件的任務被加入到了已經有1000個任務的隊列中。要收到確認郵件，他不得不等待近3個小時。多麼糟糕的用戶體驗！而另外一方面發佈新文章後通知訂閱用戶的任務並非很緊急，大多數用戶並不要求有新文章後立刻就能收到通知郵件，甚至延遲一天的時間在不少狀況下也是能夠接受的。

因此能夠得出結論當發送確認郵件和發送通知郵件兩種任務同時存在時，應該優先執行前者。爲了實現這一目的，咱們須要實現一個優先級隊列。

BRPOP命令能夠同時接收多個鍵，其完整的命令格式爲BRPOP key [key …] timeout，

按參數 key 的前後順序依次檢查各個列表，彈出第一個非空列表的尾部元素

如BRPOP queue:1 queue:2 0。意義是同時檢測多個鍵，若是全部鍵都沒有元素則阻塞，若是其中有一個鍵有元素則會從該鍵中彈出元素。例如，打開兩個redis-cli實例，在實例A中：

 

redis A> BRPOP queue:1 queue:2 queue:3 0

在實例B中：

 

redis B> LPUSH queue:2 task
(integer) 1

則實例A中會返回：

 

1) "queue:2"
2) "task"

若是多個鍵都有元素則按照從左到右的順序取第一個鍵中的一個元素。咱們先在queue:2和queue:3中各加入一個元素：

 

redis> LPUSH queue:2 task1
1) (integer) 1
redis> LPUSH queue:3 task2
2) (integer) 1

而後執行BRPOP命令：

 

redis> BRPOP queue:1 queue:2 queue:3 0
1) "queue:2"
2) "task1"

藉此特性能夠實現區分優先級的任務隊列。咱們分別使用queue:confirmation. email和queue:notification.email兩個鍵存儲發送確認郵件和發送通知郵件兩種任務，而後將消費者的代碼改成：

 

loop
    $task =
        BRPOP queue:confirmation.email,
               queue:notification.email,
              0
    execute($task[1])

這時一旦發送確認郵件的任務被加入到queue:confirmation.email隊列中，不管queue: notification.email還有多少任務，消費者都會優先完成發送確認郵件的任務。

參考：http://book.51cto.com/art/201305/395463.htm

官網說法：

RPOPLPUSH source destination

命令 RPOPLPUSH 在一個原子時間內，執行如下兩個動做：

• 將列表 source 中的最後一個元素(尾元素)彈出，並返回給客戶端。
• 將 source 彈出的元素插入到列表 destination ，做爲 destination 列表的的頭元素。

模式： 安全的隊列

Redis的列表常常被用做隊列(queue)，用於在不一樣程序之間有序地交換消息(message)。一個客戶端經過 LPUSH 命令將消息放入隊列中，而另外一個客戶端經過 RPOP 或者 BRPOP 命令取出隊列中等待時間最長的消息。

不幸的是，上面的隊列方法是『不安全』的，由於在這個過程當中，一個客戶端可能在取出一個消息以後崩潰，而未處理完的消息也就所以丟失。

使用 RPOPLPUSH 命令(或者它的阻塞版本 BRPOPLPUSH )能夠解決這個問題：由於它不只返回一個消息，同時還將這個消息添加到另外一個備份列表當中，若是一切正常的話，當一個客戶端完成某個消息的處理以後，能夠用 LREM 命令將這個消息從備份表刪除。

最後，還可以添加一個客戶端專門用於監視備份表，它自動地將超過必定處理時限的消息從新放入隊列中去(負責處理該消息的客戶端可能已經崩潰)，這樣就不會丟失任何消息了。

模式：循環列表  Circular list

經過使用相同的 key 做爲 RPOPLPUSH 命令的兩個參數，客戶端能夠用一個接一個地獲取列表元素的方式，取得列表的全部元素，而沒必要像 LRANGE 命令那樣一會兒將全部列表元素都從服務器傳送到客戶端中(兩種方式的總複雜度都是 O(N))。

以上的模式甚至在如下的兩個狀況下也能正常工做：

• 有多個客戶端同時對同一個列表進行旋轉(rotating)，它們獲取不一樣的元素，直到全部元素都被讀取完，以後又從頭開始。
• 有客戶端在向列表尾部(右邊)添加新元素。

這個模式使得咱們能夠很容易實現這樣一類系統：有 N 個客戶端，須要接二連三地對一些元素進行處理，並且處理的過程必須儘量地快。一個典型的例子就是服務器的監控程序：它們須要在儘量短的時間內，並行地檢查一組網站，確保它們的可訪問性。

注意，使用這個模式的客戶端是易於擴展(scala)且安全(reliable)的，由於就算接收到元素的客戶端失敗，元素仍是保存在列表裏面，不會丟失，等到下個迭代來臨的時候，別的客戶端又能夠繼續處理這些元素了。

 

 更多：

http://redis.io/commands/blpop

\http://redis.io/commands/rpoplpush