消息队列有两种场景

• 生产者消费者：一个消息只能有一个消费者
• 发布者订阅者：一个消息可以被多个消费者收到

redis从2.0版本开始支持pub/sub。
而Producer/Consumer是借助于redis的list结构来实现的。

生产者/消费者

Producer调用redis的lpush往特定key里放消息，Consumer调用brpop去不断监听key。

代码（未实际测试）

Producer

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

consumer

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

实际使用中,redisDao.bropop建议不要设置为BLOCK_TIMEOUT，可以设置为30（单位秒）

发布/订阅

类似设计模式中的观察者模式。
redis可以作为pub/sub的服务端。订阅得通过subscribe和psubscribe命令向redis server订阅消息类型，redis将消息类型称为channel。
当发布者通过publish命令向redis server发送特定消息时，访问该消息类型全部client都会收到此消息。
一个client可以订阅多个channel，也可以向多个channel发送消息。

测试

client0 运行：

redis-cli -h localhost -a 密码
publish tv1 program1
publish tv2 programe2

client1

subscribe tv1

client2

subscribe tv2

client3

subscribe tv1 tv2

psubscribe tv*

批量订阅以tv开头的channel内容。

取消订阅 unsubscribe tv1

批量取消订阅 punsubscribe tv*

使用java

直接使用Tcp socket订阅

import java.net.*;
import java.io.*;
public class PubSubTest {public static void main(String[] args) {String cmd = "subscribe goods.add_msg\r\n";try {Socket socket = new Socket("127.0.0.1",6379);InputStream in = socket.getInputStream();OutputStream out = socket.getOutputStream();out.write(cmd.getBytes()); //发送订阅命令byte[] buffer = new byte[1024];while (true) {int readCount = in.read(buffer);System.out.write(buffer, 0, readCount);}} catch (Exception e) {}}
}

jedis发布

/*** 发布一个消息* @param channel* @param message*/public static void publishMsg(String channel,String message){Jedis jedis = null;try {jedis = getResource();jedis.publish(channel, message);logger.debug("publishMsg {} = {}", channel, message);} catch (Exception e) {logger.warn("publishMsg {} = {}", channel, message, e);} finally {returnResource(jedis);}}/*** 发布一个消息* @param channel* @param message*/
public static void publishMsg(byte[] channel,byte[] message){Jedis jedis = null;try {jedis = getResource();jedis.publish(channel, message);logger.debug("publishMsg {} = {}", channel, message);} catch (Exception e) {logger.warn("publishMsg {} = {}", channel, message, e);} finally {returnResource(jedis);}
}

jedis订阅

需要继承JedisPubSub，并实现其中的onMessage、onSubscribe等接口。

jedis.subscribe(listener, "channel1");

JedisPubSub函数功能：

package demo;import redis.clients.jedis.JedisPubSub;public class MyListener extends JedisPubSub {// 取得订阅的消息后的处理public void onMessage(String channel, String message) {System.out.println(channel + "=" + message);}// 初始化订阅时候的处理public void onSubscribe(String channel, int subscribedChannels) {// System.out.println(channel + "=" + subscribedChannels);}// 取消订阅时候的处理public void onUnsubscribe(String channel, int subscribedChannels) {// System.out.println(channel + "=" + subscribedChannels);}// 初始化按表达式的方式订阅时候的处理public void onPSubscribe(String pattern, int subscribedChannels) {// System.out.println(pattern + "=" + subscribedChannels);}// 取消按表达式的方式订阅时候的处理public void onPUnsubscribe(String pattern, int subscribedChannels) {// System.out.println(pattern + "=" + subscribedChannels);}// 取得按表达式的方式订阅的消息后的处理public void onPMessage(String pattern, String channel, String message) {System.out.println(pattern + "=" + channel + "=" + message);}
}

订阅示例：

final Jedis jedis = ru.getConnection();
final MyListener listener = new MyListener();
//可以订阅多个频道
//订阅得到信息在lister的onMessage(...)方法中进行处理
//jedis.subscribe(listener, "channel1", "channel2");//也用数组的方式设置多个频道
//jedis.subscribe(listener, new String[]{"channel3","channel4"});//这里启动了订阅监听，线程将在这里被阻塞
//订阅得到信息在lister的onPMessage(...)方法中进行处理
jedis.psubscribe(listener, new String[]{"channel*"});//使用模式匹配的方式设置频道

本文参考：
http://www.360doc.com/content/15/0512/15/1073512_469935043.shtml