Redis 做消息隊列

一般來說，消息隊列有兩種場景，一種是發布者訂閱者模式，一種是生產者消費者模式。利用redis這兩種場景的消息隊列都能夠實現。
定義：

• 生產者消費者模式：生產者生產消息放到隊列里，多個消費者同時監聽隊列，誰先搶到消息誰就會從隊列中取走消息；即對於每個消息只能被最多一個消費者擁有。
• 發布者訂閱者模式：發布者生產消息放到隊列里，多個監聽隊列的消費者都會收到同一份消息；即正常情況下每個消費者收到的消息應該都是一樣的。

 

那么如此多的MQ產品，為什么要使用redis作消息隊列呢？以下附上一份總結了別人的一些report或blog的表格，以及當初用來說服整個team的一句結論。

MQ	Env.	Weight	Disadvantage
RabbitMQ	Erlang	Heavy	Bad scalability;Low speed;
ZeroMQ	C	Light	difficult for development
ActiveMQ	Java	-	Low performance
Redis	C	-	Low performance while enqueuing big data (>= 10k)

Redis is easy to use and configure since we have experience in Redis, and most importantly, its performance satisfies our requirement.

Then, how to use redis as a MQ?

 

首先，redis的隊列實際在代碼邏輯中不需要由我們自己實現，因此一個所謂的 RedisMQ 對象實際是一個 redis key以及對其操作的一些封裝。

 

PubSub Mode:

redis 從 2.0.0 版本開始支持 pub/sub 指令。詳情見 http://redis.io/topics/pubsub

實現思想很簡單，Publisher調用redis的publish方法往特定的channel發送消息，Subscriber在初始化的時候要subscribe到該channel，一旦有消息就會立即接收。

比較簡單的demo可參見：http://shift-alt-ctrl.iteye.com/blog/1867454 ，此鏈接博客中寫得已較詳細，本文便不再贅述。

 

Producer/Consumer Mode:

該方法是借助redis的list結構實現的。

Producer調用redis的lpush往特定key里塞入消息，Consumer調用brpop去不斷監聽該key。

producer：

// producer code
String key = "demo:mq:test";
String msg = "hello world";
redisDao.lpush(key, msg);

consumer:

// consumer code
String key = "demo:mq:test";
while (true) {
    // block invoke
    List<String> msgs = redisDao.brpop(BLOCK_TIMEOUT, listKey);
    if (msgs == null) continue;
    String jobMsg = msgs.get(1);
    processMsg(jobMsg);
}

當有多個consumers的時候，它會按照brpop調用的順序分派消息，並非隨機。

BLOCK_TIMEOUT不建議設成infinity（有些redis驅動也直接不支持inifinity），我們目前設成30（單位是秒）情況良好。

 

P.S. 本文時間較久遠，適合redis 2的版本，不保證redis自己會不會有其他新特性 ；同時消息隊列產品有很多種，這里列的只是早年常用的，近兩三年的kafka和阿里的rocketmq也很火，至於怎么選擇，一部分是根據數據量，若數據量不大，容錯要求不是極高，redis是個高效開發易維護的好選擇；如果數據量很大或對消息准確性有一定要求，那應當考慮更成熟的消息隊列產品比如kafka等。所以mq的選型並不是本文的重點，本文只是介紹一下基於redis 2.6的mq的簡單封裝實現。