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

时间：2025-03-22 09:32:03

转载  /chengxuyuan20100425/article/details/9100569


消费端：

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

       xsi:schemaLocation="/schema/beans
       /schema/beans/spring-beans-3.
	   /schema/context
       /schema/context/spring-context-3.">

    <bean  class=""
          p:hostName="192.168.1.234" p:port="6379" p:usePool="true">
    </bean>

    <!-- redis template definition -->
    <bean  class=""
          p:connectionFactory-ref="jedisConnectionFactory"/>

    <bean  class="">
        <property name="redisTemplate" ref="redisTemplate" />
    </bean>

    <bean  class="" />

    <bean  class="" />

    <bean  class="">
        <property name="delegate" ref="messageDelegateListener" />
        <property name="serializer" ref="serialization" />
    </bean>

    <bean  class="">
        <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="">
                        <constructor-arg value="java" />   <!-- 这里配置消费端需要订阅的频道，可以是多个。该一例子订阅JAVA这个频道  -->
                    </bean>
                </entry>
            </map>
        </property>
    </bean>
</beans>
服务端 

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

       xsi:schemaLocation="/schema/beans
       /schema/beans/spring-beans-3.
	   /schema/context
       /schema/context/spring-context-3.
       /schema/redis
        /schema/redis/spring-redis-1.">


    <bean  class=""
          p:hostName="192.168.1.234" p:port="6379" p:usePool="true">
    </bean>

    <!-- redis template definition -->
    <bean  class=""
          p:connectionFactory-ref="redisConnectionFactory"/>

    <bean  class="">
        <property name="redisTemplate" ref="redisTemplate" />
    </bean>

    <bean  class=""/>

    <!-- the default ConnectionFactory -->
    <bean  class="" />

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

import ;

import ;

public interface RedisDAO {

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

	public abstract RedisTemplate getRedisTemplate();

	public abstract void setRedisTemplate(RedisTemplate redisTemplate);

}



package ;

import ;

import ;

import ;

public class RedisDAOImpl implements RedisDAO{

    private RedisTemplate<String, Object> redisTemplate = null;

    public RedisDAOImpl() {

    }

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


    public RedisTemplate getRedisTemplate() {
        return redisTemplate;
    }

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



package ;

import ;

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



package ;

import ;
import ;
import ;
import ;

import ;

/**
 * 接收消息的Listener,用于接收订阅到的消息. 
 * @author Administrator
 *
 */
public class MessageDelegateListenerImpl implements MessageDelegateListener {

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


启动消费端 
package ;


import ;
import ;
import ;

import ;


public class TestRedisConsumer {
    private MessageDelegateListenerImpl messageDelegateListener=null;

    @Before
    public void setUp() throws Exception {
        ApplicationContext applicationContext = new ClassPathXmlApplicationContext("");
        messageDelegateListener = (MessageDelegateListenerImpl) ("messageDelegateListener");
    }

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

}



启动生产端 发送信息

package ;


import ;
import ;
import ;
import ;

import ;

public class TestRedisProduce {
    private RedisDAOImpl redisDAO=null;

    @Before
    public void setUp() throws Exception {
        ApplicationContext applicationContext = new ClassPathXmlApplicationContext("");
        redisDAO = (RedisDAOImpl) ("redisDAO");
    }

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

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

客户端返回：

package ;


import ;
import ;
import ;

import ;


public class TestRedisConsumer {
    private MessageDelegateListenerImpl messageDelegateListener=null;

    @Before
    public void setUp() throws Exception {
        ApplicationContext applicationContext = new ClassPathXmlApplicationContext("");
        messageDelegateListener = (MessageDelegateListenerImpl) ("messageDelegateListener");
    }

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

}

客户端返回：
消费者1
[21341, 123123, 12323]
@84ce7a[value={H,e,l,l,o,,, ,R,e,d,i,s,!},offset=0,count=13,hash=1345989452


项目打包下载

package ;


import ;
import ;
import ;

import ;


public class TestRedisConsumer {
    private MessageDelegateListenerImpl messageDelegateListener=null;

    @Before
    public void setUp() throws Exception {
        ApplicationContext applicationContext = new ClassPathXmlApplicationContext("");
        messageDelegateListener = (MessageDelegateListenerImpl) ("messageDelegateListener");
    }

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

}

相关文章