许久没发博客，转载一篇Spring Data Redis订阅/发布系统 -

ilrxx

浏览: 65118 次
性别:
来自: 北京

最近访客更多访客>>

zoulei

winnerq

zwd_18776156173

quhaobo11

博主相关

博客

微博

相册

留言

关于我

文章分类

全部博客 (34)

社区版块

存档分类

许久没发博客，转载一篇Spring Data Redis订阅/发布系统

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

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

下载,安装和配置Redis,见: http://my.oschina.net/u/729474/blog/79128 和http://www.php100.com/html/webkaifa/PHP/PHPyingyong/2011/0406/7873.html

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

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

01

public class RedisDAOImpl implements RedisDAO {

02

03

    private RedisTemplate<String, Object> redisTemplate = null;

04

05

    public RedisDAOImpl() {

06

07

    }

08

09

    @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,用于接收订阅到的消息.
代码如:

01

public class MessageDelegateListenerImpl implements MessageDelegateListener {

02

03

    @Override

04

    public void handleMessage(Serializable message) {

05

        //什么都不做,只输出

06

        if(message == null){

07

            System.out.println("null");

08

        } else if(message.getClass().isArray()){

09

            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:

01

<?xml version="1.0" encoding="UTF-8"?>

02

<beans xmlns="http://www.springframework.org/schema/beans"

03

       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

04

       xmlns:context="http://www.springframework.org/schema/context"

05

       xmlns:redis="http://www.springframework.org/schema/redis"

06

       xmlns:p="http://www.springframework.org/schema/p"

07

08

       xsi:schemaLocation="http://www.springframework.org/schema/beans

09

       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的消息才会被接收.

测试类:

01

public static void main(String[] args) {

02

        new ClassPathXmlApplicationContext("pubsubAppContext1.xml");;

03

        while (true) { //这里是一个死循环,目的就是让进程不退出,用于接收发布的消息

04

            try {

05

                System.out.println("current time: " + new Date());

06

07

                Thread.sleep(3000);

08

            } catch (InterruptedException e) {

09

                e.printStackTrace();

10

            }

11

        }

12

    }

OK,启动了订阅系统后,我们就可以发布消息,测试类如:

01
@Test

02

    public void testPublishMessage() throws Exception {

03

        String msg = "Hello, Redis!";

04

        redisDAO.sendMessage("java", msg); //发布字符串消息

05

06

07

        RedisTestBean bean = new RedisTestBean("123456");

08

        bean.setName("Redis");

09

        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还有传统风格的配置方式,实现的功能和前面一模一样.

01

<?xml version="1.0" encoding="UTF-8"?>

02

<beans xmlns="http://www.springframework.org/schema/beans"

03

       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

04

       xmlns:context="http://www.springframework.org/schema/context"

05

       xmlns:p="http://www.springframework.org/schema/p"

06

07

       xsi:schemaLocation="http://www.springframework.org/schema/beans

08

       http://www.springframework.org/schema/beans/spring-beans-3.0.xsd

09

       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>