您的位置：首页 > 数据库 > Redis

Spring Data Redis实现一个订阅/发布系统

2015-11-04 17:47 741 查看
Redis是一个key-value的存储系统,提供的key-value类似与Memcached而数据结构又多于memcached,而且性能优异.广泛用于缓存,临时存储等.而我今天 这个例子是使用Redis实现一个订阅/发布系统,而不是如何使用它存储key-value的数据. 

Redis是天生支持订阅/发布的,不是我牵强附会拼凑而实现这样的效果,如果真是这样性能没法保证,而且要实现订阅/发布这样的系统是有很多解决方案的. 

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




publish: 向指定的channel(频道)发送message(消息)

subscribe:订阅指定channel，可以一次订阅多个

psubscribe:订阅指定pattern(模式，具有频道名的模式匹配)的频道

unsubscribe:取消订阅channel,可以一次取消多个订阅

punsubscribe:取消指定pattern的订阅

pubsub: 是一个查看订阅与发布系统状态的内省命令,它由数个不同格式的子命令组成

在SDR（Spring Data Redis）中频道对应Topic类，Top类是一个接口有Channel和Pattern两个实现类，分别是指定名称的频道和模式匹配的频道。关于订阅信息由Subscription接口定义。

在SDR中可以用两种方式来实现消息监听容器的声明，一种是通过Redis的命名空间，一种是定义Bean。这里主要涉及到RedisMessageListenerContainer，MessageListenerAdapter，MessageListener几个类。


Spring一直秉承不发明轮子的,对于很多其他技术都是提供一个模板:Template,如JDBC-JdbcTemplate,JMSTemplate等,Redis他也提供RedisTemplate,有了这个RedisTemplate你可以做任何事,存取key-value,订阅,发布等都通过这个对象实现.

实现一个RedisDAO,接口我不贴了

点击(此处)折叠或打开

  1. public class RedisDAOImpl implements RedisDAO {
  2.  
  3.     private RedisTemplate redisTemplate = null;
  4.  
  5.     public RedisDAOImpl() {
  6.  
  7.     }
  8.  
  9.     @Override
  10.     public void sendMessage(String channel, Serializable message) {
  11.         redisTemplate.convertAndSend(channel, message);
  12.     }
  13.  
  14.  
  15.     public RedisTemplate getRedisTemplate() {
  16.         return redisTemplate;
  17.     }
  18.  
  19.     public void setRedisTemplate(RedisTemplate redisTemplate) {
  20.         this.redisTemplate = redisTemplate;
  21.     }
  22. }
可以看到,通过这个 sendMessage方法,我可以把一条可序列化的消息发送到channel频道,订阅者只要订阅了这个channel,他就会接收发布者发布的消息. 

当然有了发布消息的sendMessage也得有个接收消息的Listener,用于接收订阅到的消息. 
代码如: 

点击(此处)折叠或打开

  1. public class MessageDelegateListenerImpl implements MessageDelegateListener {
  2.  
  3.     @Override
  4.     public void handleMessage(Serializable message) {
  5.         //什么都不做,只输出
  6.         if(message == null){
  7.             System.out.println("null");
  8.         } else if(message.getClass().isArray()){
  9.             System.out.println(Arrays.toString((Object[])message));
  10.         } else if(message instanceof List) {
  11.             System.out.println(message);
  12.         } else if(message instanceof Map) {
  13.             System.out.println(message);
  14.         } else {
  15.             System.out.println(ToStringBuilder.reflectionToString(message));
  16.         }
  17.     }
  18. }

好了,有上面的两个类,加上Spring基本上就可以工作了.当然还得启动Redis. 
Spring Schema:

点击(此处)折叠或打开

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <beans xmlns="http://www.springframework.org/schema/beans"
  3.        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.        xmlns:context="http://www.springframework.org/schema/context"
  5.        xmlns:redis="http://www.springframework.org/schema/redis"
  6.        xmlns:p="http://www.springframework.org/schema/p"
  7.  
  8.        xsi:schemaLocation="http://www.springframework.org/schema/beans
  9.        http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
  10.        http://www.springframework.org/schema/context
  11.        http://www.springframework.org/schema/context/spring-context-3.0.xsd
  12.        http://www.springframework.org/schema/redis
  13.         http://www.springframework.org/schema/redis/spring-redis-1.0.xsd">
  14.  
  15.     <bean id="redisConnectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory"
  16.           p:hostName="localhost" p:port="6379" p:usePool="true">
  17.     </bean>
  18.  
  19.     <!-- redis template definition -->
  20.     <bean id="redisTemplate" class="org.springframework.data.redis.core.RedisTemplate"
  21.           p:connectionFactory-ref="redisConnectionFactory"/>
  22.  
  23.     <bean id="redisDAO" class="net.dredis.dao.impl.RedisDAOImpl">
  24.         <property name="redisTemplate" ref="redisTemplate" />
  25.     </bean>
  26.  
  27.     <bean id="listener" class="net.dredis.listener.impl.MessageDelegateListenerImpl"/>
  28.  
  29.     <!-- the default ConnectionFactory -->
  30.     <bean id="jdkSerializer" class="org.springframework.data.redis.serializer.JdkSerializationRedisSerializer" />
  31.  
  32.     <redis:listener-container>
  33.         <!-- the method attribute can be skipped as the default method name is "handleMessage" -->
  34.         <redis:listener ref="listener" serializer="jdkSerializer" method="handleMessage" topic="java" />
  35.     </redis:listener-container>
  36. </beans>
如上面的配置, jdkSerializer是jdk默认的序列化的实现,当然还有很多其他序列化Java对象的方法,这里使用jdk默认实现. 
Method属性是配置订阅系统接收消息的方法,默认也是"handleMessage" 
topic就是订阅的channel频道,是有发布到java这个channel的消息才会被接收。定义topic可以是具体的channel的名字也可以是Pattern（如： topic="spring*"），多个频道（主题）用空格隔开即可。

测试类: 

点击(此处)折叠或打开

  1. public static void main(String[] args) {
  2.         new ClassPathXmlApplicationContext("pubsubAppContext1.xml");;
  3.         while (true) { //这里是一个死循环,目的就是让进程不退出,用于接收发布的消息
  4.             try {
  5.                 System.out.println("current time: " + new Date());
  6.  
  7.                 Thread.sleep(3000);
  8.             } catch (InterruptedException e) {
  9.                 e.printStackTrace();
  10.             }
  11.         }
  12.     }
OK,启动了订阅系统后,我们就可以发布消息,测试类如: 

点击(此处)折叠或打开

  1. @Test
  2.     public void testPublishMessage() throws Exception {
  3.         String msg = "Hello, Redis!";
  4.         redisDAO.sendMessage("java", msg); //发布字符串消息
  5.  
  6.  
  7.         RedisTestBean bean = new RedisTestBean("123456");
  8.         bean.setName("Redis");
  9.         bean.setOld((byte)2);
  10.         bean.setSeliry((short)40);
  11.         bean.setManbers(new String[]{"234567", "3456789"});
  12.         redisDAO.sendMessage("java", bean); //发布一个普通的javabean消息
  13.  
  14.  
  15.         Integer[] values = new Integer[]{21341,123123,12323};
  16.         redisDAO.sendMessage("java", values); //发布一个数组消息
  17.     }

如测试,我连续发布了3条消息,都是不同的数据类型.订阅端输出如:

点击(此处)折叠或打开

  1. current time: Fri Oct 26 20:38:31 CST 2012
  2. [21341, 123123, 12323]
  3. java.lang.String@379faa8c[value={H,e,l,l,o,,, ,R,e,d,i,s,!},hash=1345989452]
  4. net.dredis.entity.RedisTestBean@7dee05dc[uid=123456,name=Redis,seliry=40,old=2,manbers={234567,3456789}]
  5. current time: Fri Oct 26 20:38:34 CST 2012
  6. current time: Fri Oct 26 20:38:37 CST 2012

OK他接收到了这3条消息,而且和预期一样. 
对于Spring还有传统风格的配置方式,实现的功能和前面一模一样. 

点击(此处)折叠或打开

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <beans xmlns="http://www.springframework.org/schema/beans"
  3.        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.        xmlns:context="http://www.springframework.org/schema/context"
  5.        xmlns:p="http://www.springframework.org/schema/p"
  6.  
  7.        xsi:schemaLocation="http://www.springframework.org/schema/beans
  8.        http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
  9.        http://www.springframework.org/schema/context
  10.        http://www.springframework.org/schema/context/spring-context-3.0.xsd">
  11.  
  12.     <bean id="jedisConnectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory"
  13.           p:hostName="localhost" p:port="6379" p:usePool="true">
  14.     </bean>
  15.  
  16.     <!-- redis template definition -->
  17.     <bean id="redisTemplate" class="org.springframework.data.redis.core.RedisTemplate"
  18.           p:connectionFactory-ref="jedisConnectionFactory"/>
  19.  
  20.     <bean id="redisDAO" class="net.dredis.dao.impl.RedisDAOImpl">
  21.         <property name="redisTemplate" ref="redisTemplate" />
  22.     </bean>
  23.  
  24.     <bean id="serialization" class="org.springframework.data.redis.serializer.JdkSerializationRedisSerializer" />
  25.  
  26.     <bean id="messageDelegateListener" class="net.dredis.listener.impl.MessageDelegateListenerImpl" />
  27.      
  28.     <bean id="messageListener" class="org.springframework.data.redis.listener.adapter.MessageListenerAdapter">
  29.         <property name="delegate" ref="messageDelegateListener" />
  30.         <property name="serializer" ref="serialization" />
  31.     </bean>
  32.  
  33.     <bean id="redisContainer" class="org.springframework.data.redis.listener.RedisMessageListenerContainer">
  34.         <property name="connectionFactory" ref="jedisConnectionFactory"/>
  35.         <property name="messageListeners">
  36.             <!-- map of listeners and their associated topics (channels or/and patterns) -->
  37.             <map>
  38.                 <entry key-ref="messageListener">
  39.                     <bean class="org.springframework.data.redis.listener.ChannelTopic">
  40.                         <constructor-arg value="java" />
  41.                     </bean>
  42.                 </entry>
  43.             </map>
  44.         </property>
  45.     </bean>
  46. </beans>

来自 “ ITPUB博客 ” ，链接：http://blog.itpub.net/9399028/viewspace-1822585/，如需转载，请注明出处，否则将追究法律责任。

转载于:http://blog.itpub.net/9399028/viewspace-1822585/

内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理 点击这里给我发消息
标签： 
相关文章推荐
新的分享
章节导航