spring-redis实现订阅发布

导入包：
        <dependency>
            <groupId>org.springframework.data</groupId>
            <artifactId>spring-data-redis</artifactId>
            <version>1.0.3.RELEASE</version>
        </dependency>

生产端代码：接口省略

package jredis;

import java.io.Serializable;
import org.springframework.data.redis.core.RedisTemplate ;


public class RedisDAOImpl implements RedisDAO{
    private RedisTemplate<String, Object> redisTemplate = null;

    public RedisDAOImpl() {

    }


    @Override
    public void sendMessage(String channel, Serializable message) {
        redisTemplate.convertAndSend(channel, message);
    }


    public RedisTemplate getRedisTemplate() {
        return redisTemplate;
    }

    public void setRedisTemplate(RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }
}

客户端代码：用于监听redis

package jredis;

import com.sun.tools.javac.util.List;
import com.sun.xml.internal.xsom.impl.scd.Iterators;
import org.apache.commons.lang.builder.ToStringBuilder;

import java.io.Serializable;
import java.util.Arrays;

public class MessageDelegateListenerImpl implements MessageDelegateListener {

    @Override
    public void handleMessage(Serializable message) {
        //什么都不做,只输出
        if(message == null){
            System.out.println("null");
        } else if(message.getClass().isArray()){
            System.out.println(Arrays.toString((Object[]) message));
        } else if(message instanceof List<?>) {
            System.out.println(message);
        } else if(message instanceof Iterators.Map<? , ?>) {
            System.out.println(message);
        } else {
            System.out.println(ToStringBuilder.reflectionToString(message));
            System.out.println(message);
        }
    }
}

先启消费端,消费端测试类：

package jredis;

import org.junit.Before;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import java.util.Date;


public class TestRedisConsumer {
    private MessageDelegateListenerImpl messageDelegateListener=null;

    @Before
    public void setUp() throws Exception {
        ApplicationContext applicationContext = new ClassPathXmlApplicationContext("spring-consumer-test.xml");
        messageDelegateListener = (MessageDelegateListenerImpl) applicationContext.getBean("messageDelegateListener");
    }

    public static void main(String[] args) {
        new ClassPathXmlApplicationContext("spring-consumer-test.xml");
        System.out.println("消费者1");
        while (true) { //这里是一个死循环,目的就是让进程不退出,用于接收发布的消息
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}

生产端测试类：

package jredis;

import service.weibo.impl.TencentOauthV1BackgroundServiceImpl;
import service.weibo.impl.TencentOauthV1ForegroundServiceImpl;
import org.junit.Before;
import org.junit.Test;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class TestRedisProduce {
    private RedisDAOImpl redisDAO=null;

    @Before
    public void setUp() throws Exception {
        ApplicationContext applicationContext = new ClassPathXmlApplicationContext("spring-service-test.xml");
        redisDAO = (RedisDAOImpl) applicationContext.getBean("redisDAO");
        }


    @Test
    public void testPublishMessage() throws Exception {
        String msg = "Hello, Redis!";
        redisDAO.sendMessage("java", msg); //发布字符串消息


       Integer[] values = new Integer[]{21341,123123,12323};
        redisDAO.sendMessage("java", values);  //发布一个数组消息
    }
}

生产端xml配置：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:redis="http://www.springframework.org/schema/redis"
       xmlns:p="http://www.springframework.org/schema/p"

       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
	   http://www.springframework.org/schema/context
       http://www.springframework.org/schema/context/spring-context-3.0.xsd
       http://www.springframework.org/schema/redis
        http://www.springframework.org/schema/redis/spring-redis-1.0.xsd">


    <bean id="redisConnectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory"
          p:hostName="10.25.172.174" p:port="6379" p:usePool="true">
    </bean>

    <!-- redis template definition -->
    <bean id="redisTemplate" class="org.springframework.data.redis.core.RedisTemplate"
          p:connectionFactory-ref="redisConnectionFactory"/>

    <bean id="redisDAO" class="jredis.RedisDAOImpl">
        <property name="redisTemplate" ref="redisTemplate" />
    </bean>

    <bean id="listener" class="jredis.MessageDelegateListenerImpl"/>

    <!-- the default ConnectionFactory -->
    <bean id="jdkSerializer" class="org.springframework.data.redis.serializer.JdkSerializationRedisSerializer" />

    <redis:listener-container>
        <!-- the method attribute can be skipped as the default method name is "handleMessage" -->
        <redis:listener ref="listener" serializer="jdkSerializer" method="handleMessage" topic="java" />  <!--  发布频道的名称-->
    </redis:listener-container>
</beans>

消费端XML配置：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:p="http://www.springframework.org/schema/p"

       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
	   http://www.springframework.org/schema/context
       http://www.springframework.org/schema/context/spring-context-3.0.xsd">

    <bean id="jedisConnectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory"
          p:hostName="10.25.172.174" p:port="6379" p:usePool="true">
    </bean>

    <!-- redis template definition -->
    <bean id="redisTemplate" class="org.springframework.data.redis.core.RedisTemplate"
          p:connectionFactory-ref="jedisConnectionFactory"/>

    <bean id="redisDAO" class="jredis.RedisDAOImpl">
        <property name="redisTemplate" ref="redisTemplate" />
    </bean>

    <bean id="serialization" class="org.springframework.data.redis.serializer.JdkSerializationRedisSerializer" />

    <bean id="messageDelegateListener" class="jredis.MessageDelegateListenerImpl" />

    <bean id="messageListener" class="org.springframework.data.redis.listener.adapter.MessageListenerAdapter">
        <property name="delegate" ref="messageDelegateListener" />
        <property name="serializer" ref="serialization" />
    </bean>

    <bean id="redisContainer" class="org.springframework.data.redis.listener.RedisMessageListenerContainer">
        <property name="connectionFactory" ref="jedisConnectionFactory"/>
        <property name="messageListeners">
            <!-- map of listeners and their associated topics (channels or/and patterns) -->
            <map>
                <entry key-ref="messageListener">
                    <bean class="org.springframework.data.redis.listener.ChannelTopic">
                        <constructor-arg value="java" />   <!-- 这里配置消费端需要订阅的频道，可以是多个。该一例子订阅JAVA这个频道  -->
                    </bean>
                </entry>
            </map>
        </property>
    </bean>
</beans>

缺点：

因为spring-redis是基于客户端的，所以如果将客户端进行分布式部署，部署几个实例就有几个实例收到相同的消息。大多数情况我们并不需要如此。我们只需要其中的一个实例处理该条消息即可。这种情况，redis有没有提供处理办法呢？本人还没有找到，欢迎大家提供方案。

原文链接：https://blog.csdn.net/chengxuyuan20100425/article/details/9100569