1. Redis的Pub/Sub命令
Redis的订阅和发布服务有如下图6个命令，下面分别对每个命令做简单说明。
publish: 向指定的channel(频道)发送message(消息)
subscribe:订阅指定channel，可以一次订阅多个
psubscribe:订阅指定pattern(模式，具有频道名的模式匹配)的频道
unsubscribe:取消订阅channel,可以一次取消多个订阅
punsubscribe:取消指定pattern的订阅
pubsub: 是一个查看订阅与发布系统状态的内省命令,它由数个不同格式的子命令组成(具体可参见：
http://redis.io/commands/pubsub)
在SDR（Spring Data Redis）中频道对应Topic类，Top类是一个接口有Channel和Pattern两个实现类，分别是指定名称的频道和模式匹配的频道。关于订阅信息由Subscription接口定义。
2.Redis消息监听容器声明和消息监听器注册
在SDR中可以用两种方式来实现消息监听容器的声明，一种是通过Redis的命名空间，一种是定义Bean。
这里主要涉及到RedisMessageListenerContainer，MessageListenerAdapter，MessageListener几个类。
2.1使用Redis命名空间的方式配置
|
1
2
3
4
5
6
7
8
|
<
bean
id
=
"mdpListener"
class
=
"secondriver.spring.redis.MyMessageListener"
/>
<
bean
id
=
"mdelegateListener"
class
=
"secondriver.spring.redis.DefaultMessageDelegate"
/>
<
redis:listener-container
connection-factory
=
"jedisConnectionFactory"
>
<
redis:listener
ref
=
"mdpListener"
topic
=
"spring*"
/>
<
redis:listener
ref
=
"mdelegateListener"
method
=
"handleMessage"
topic
=
"cctv5 cctv6 nbtv hello*"
/>
</
redis:listener-container
>
|
说明:
定义topic可以是具体的channel的名字也可以是Pattern，多个频道（主题）用空格隔开即可。
这里定义了两个Listner，MyMessageListener实现了MessageaListener接口,DefaultMessageDelegate实现了MessageDelegate接口。
MyMessageListener:
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
package
secondriver.spring.redis;
import
org.springframework.data.redis.connection.Message;
import
org.springframework.data.redis.connection.MessageListener;
public
class
MyMessageListener
implements
MessageListener {
@Override
public
void
onMessage(Message message,
byte
[] pattern) {
System.out.println(
"channel:"
+
new
String(message.getChannel())
+
",message:"
+
new
String(message.getBody()));
}
}
|
MessageDelegate接口：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
package
secondriver.spring.redis;
import
java.io.Serializable;
import
java.util.Map;
public
interface
MessageDelegate {
public
void
handleMessage(String message);
public
void
handleMessage(Map<?, ?> message);
public
void
handleMessage(
byte
[] message);
public
void
handleMessage(Serializable message);
// pass the channel/pattern as well
public
void
handleMessage(Serializable message, String channel);
}
|
DefaultMessageDelegate类型：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
|
package
secondriver.spring.redis;
import
java.io.Serializable;
import
java.util.Map;
public
class
DefaultMessageDelegate
implements
MessageDelegate {
@Override
public
void
handleMessage(String message) {
System.out.println(
"handleMessage(String message):"
+ message);
}
@Override
public
void
handleMessage(Map<?, ?> message) {
System.out.println(
"handleMessage(Map<?, ?> message):"
+ message);
}
@Override
public
void
handleMessage(
byte
[] message) {
System.out.println(
"handleMessage(byte[] message):"
+
new
String(message));
}
@Override
public
void
handleMessage(Serializable message) {
System.out.println(
"handleMessage(Serializable message):"
+ message.toString());
}
@Override
public
void
handleMessage(Serializable message, String channel) {
System.out
.println(
"handleMessage(Serializable message, String channel):"
+ message.toString() +
", channel:"
+ channel);
}
}
|
这种定义消息监听的方式不依赖于Redis，其被设计为一个message-driven POJOs (MDPs)，MessageListenerAdapter实现了MessageListener接口，它将会把Message委托给目标监听器（Target Listener）DefalutMessageDelegate对象的方法，并对其Message参数进行适当转换，然后通过反射来调用方法。
2.2定义Bean的方式配置
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
|
<!-- Bean Configuration -->
<bean id=
"messageListener"
class
=
"org.springframework.data.redis.listener.adapter.MessageListenerAdapter"
>
<constructor-arg>
<bean
class
=
"secondriver.spring.redis.MyMessageListener"
/>
</constructor-arg>
</bean>
<bean id=
"redisContainer"
class
=
"org.springframework.data.redis.listener.RedisMessageListenerContainer"
>
<property name=
"connectionFactory"
ref=
"jedisConnectionFactory"
/>
<property name=
"messageListeners"
>
<map>
<entry key-ref=
"messageListener"
>
<list>
<bean
class
=
"org.springframework.data.redis.listener.ChannelTopic"
>
<constructor-arg value=
"springtv"
/>
</bean>
<bean
class
=
"org.springframework.data.redis.listener.PatternTopic"
>
<constructor-arg value=
"hello*"
/>
</bean>
<bean
class
=
"org.springframework.data.redis.listener.PatternTopic"
>
<constructor-arg value=
"tv*"
/>
</bean>
</list>
</entry>
</map>
</property>
</bean>
|
3.模拟消息的发布和接收
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
// 简单测试RedisMessageListener
@Ignore
@Test
public
void
test10()
throws
InterruptedException {
RedisMessageListenerContainer rmlc;
// ctx.getBean(RedisMessageListenerContainer.class);
rmlc = (RedisMessageListenerContainer) ctx.getBean(
"redisContainer"
);
while
(
true
) {
if
(rmlc.isRunning()) {
System.out
.println(
"RedisMessageListenerContainer is running.."
);
}
Thread.sleep(
5000
);
}
}
|
由于这里是测试，通过一个死循环来保持程序一直运行，然后向Redis服务的指定频道发布消息，则符合订阅的频道的消息将被客户端连接接收到，并且MyMessageListener对象中的onMessage方法被调用。
下图是模拟过程：
发布了四条消息，分别是spring,springtv，hello， nono四个频道，根据定义bean这种配置中的Topic名称，只有springtv,hello符合模式匹配，并且也同样收到了这两条消息。
除了这种模拟外，实际在应用开发中是通过JedisConnection的Pub/Sub相关的方法来向Redis服务发布消息的或者RedisTemplate的convertAndSend方法。
4.最后
SDR正处于发展阶段的项目，更多特性阅读源代码，一步步挖掘。
如第一篇中提到了关于Sentinel的实现集成，多中客户端对Redis特性的实现差异和不支持的特性还在进一步的改善之中。
本文转自 secondriver 51CTO博客，原文链接：http://blog.51cto.com/aiilive/1627478，如需转载请自行联系原作者
