Redis系列-JAVA与redis整合-Spring Data Redis实现一个订阅/发布系统

转载 /article/10222896.html

消费端：

<?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="192.168.1.234" 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="pubsub.dao.impl.RedisDAOImpl">
<property name="redisTemplate" ref="redisTemplate" />
</bean>

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

<bean id="messageDelegateListener" class="pubsub.message.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>

服务端

<?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="192.168.1.234" 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="pubsub.dao.impl.RedisDAOImpl">
<property name="redisTemplate" ref="redisTemplate" />
</bean>

<bean id="listener" class="pubsub.message.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>

package pubsub.dao;

import java.io.Serializable;

import org.springframework.data.redis.core.RedisTemplate;

public interface RedisDAO {

public abstract void sendMessage(String channel, Serializable message);

public abstract RedisTemplate getRedisTemplate();

public abstract void setRedisTemplate(RedisTemplate redisTemplate);

}

package pubsub.dao.impl;

import java.io.Serializable;

import org.springframework.data.redis.core.RedisTemplate;

import pubsub.dao.RedisDAO;

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;
}
}

package pubsub.message;

import java.io.Serializable;

public interface MessageDelegateListener {
void handleMessage(Serializable message);
}

package pubsub.message;

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

import org.apache.commons.lang.builder.ToStringBuilder;

/**
* 接收消息的Listener,用于接收订阅到的消息.
* @author Administrator
*
*/
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 Map<?, ?>) {
System.out.println(message);
} else {
System.out.println(ToStringBuilder.reflectionToString(message));
}
}
}

启动消费端

package pubsub.test;

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

import pubsub.message.MessageDelegateListenerImpl;

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 pubsub.test;

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

import pubsub.dao.impl.RedisDAOImpl;

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);  //发布一个数组消息
}
}

客户端返回：

package pubsub.test;

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

import pubsub.message.MessageDelegateListenerImpl;

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();
}
}
}

}

客户端返回：

消费者1

[21341, 123123, 12323]

java.lang.String@84ce7a[value={H,e,l,l,o,,, ,R,e,d,i,s,!},offset=0,count=13,hash=1345989452

项目打包下载

package pubsub.test;

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

import pubsub.message.MessageDelegateListenerImpl;

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();
}
}
}

}