使用Spring Data Redis操作Redis(二)

 1. Redis的Pub/Sub命令

  Redis的订阅和发布服务有如下图6个命令,下面分别对每个命令做简单说明。

  

  使用Spring Data Redis操作Redis(二)

  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 Data Redis操作Redis(二)

   发布了四条消息,分别是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,如需转载请自行联系原作者

上一篇:Spring旅程(三) AOP--Spring AOP容器基础


下一篇:微信web开发ui框架制作网页实例教程