Redis常用技术-----发布订阅

刺骨的言语ヽ痛彻心扉 2022-05-16 10:36 166阅读 0赞

银行卡消费的时候，银行往往会通过微信、短信或邮件通知用户交易的详细信息，这便是发布订阅模式。

![交易信息发布订阅机制][5099bffe691ae9a54ece00baa62715a4.png]

我们可以使用下面的命令进行模拟

先打开一个客户端1，输入命令

SUBSCRIBE chat

表示客户端1订阅一个名为chat的渠道，之后打开客户端2,输入命令

publish chat "Hello World!"

这个时候客户端2就向渠道chat发送了一条消息，再看客户端1,可以看到消息已经输出。  
![Redis的发布订阅过程][Redis]

在Spring中，我们自定义接收消息的类，需要实现MessageListener接口，并实现接口定义的方法onMessage，代码如下。

public class RedisMessageListener implements MessageListener {
    
    	@SuppressWarnings("rawtypes")
    	private RedisTemplate rt;
    	
    	@SuppressWarnings("rawtypes")
    	public RedisTemplate getRt() {
    		return rt;
    	}
    
    	@SuppressWarnings("rawtypes")
    	public void setRt(RedisTemplate rt) {
    		this.rt = rt;
    	}
    
    
    	@Override
    	public void onMessage(Message message, byte[] pattern) {
    		// 获取消息
    		byte[] body = message.getBody();
    		// 使用值序列化器转换
    		String msgBody = (String)getRt().getValueSerializer().deserialize(body);
    		System.out.println(msgBody);
    		// 获取channel
    		byte[] channel = message.getChannel();
    		// 使用字符串序列化器转换
    		String channelStr = (String)getRt().getStringSerializer().deserialize(channel);
    		System.out.println(channelStr);
    		// 渠道名称转换
    		String bytesStr = new String(pattern);
    		System.out.println(bytesStr);
    	}
    
    }

这里我们没有使用注解，所以要去spring.xml中注册bean

<bean id="redisMsgListener" class="com.codeliu.transaction.RedisMessageListener">
        <property name="rt" ref="redisTemplate"></property>
    </bean>

实例化监听类之后还不行，还得给一个监听容器，在Spring中对应的类为RedisMessageListenerContainer，它可以用来监听Redis的发布订阅消息。同样在Spring.xml中进行配置

<bean id="topicContainer" class="org.springframework.data.redis.listener.RedisMessageListenerContainer" destroy-method="destroy">
    	
    	<property name="connectionFactory" ref="connectionFactory"></property>      
    	
    	<property name="taskExecutor">
    		<bean class="org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler">
    			<property name="poolSize" value="3"></property>
    		</bean>
    	</property>  	
    	
    	<property name="messageListeners">
    		<map>
    			
    			<entry key-ref="redisMsgListener">
    				
    				<bean class="org.springframework.data.redis.listener.ChannelTopic">
    					<constructor-arg value="chat"></constructor-arg>
    				</bean>
    			</entry>
    		</map>
    	</property>
    </bean>

这里配置了线程池，这个线程池将会持续的生存以等待消息传入，而这里配置了容器用id为redisMsgListener的Bean进行渠道chat的监听。当消息通过渠道chat方法的时候，就会使用id为redisMsgListener的Bean处理消息。

下面进行测试

public static void main(String[] args) {
    	ApplicationContext applicationContext = new ClassPathXmlApplicationContext("spring.xml");
    	RedisTemplate rt = applicationContext.getBean(RedisTemplate.class);
    	String channel = "chat";
    	// 向渠道chat发送消息
    	rt.convertAndSend(channel, "Hello World!");
    }

convertAndSend方法就是向指定的渠道发送消息。

输出

Hello World!
    chat
    chat

[5099bffe691ae9a54ece00baa62715a4.png]: /images/20220516/2d381783db694868b688aa6bb078925c.png
[Redis]: /images/20220516/de2d165c6c7f4dbc9d6c707f23c994fa.png