你见过不长毛的羊吗

Redis(九)-消息订阅与发布

2019.08.09

1. 定义

Redis 发布订阅(pub/sub):是一种进程间消息通信模式——发送者(pub)发送消息,订阅者(sub)接收消息。

2. 示意图

来源:Redis发布订阅|菜鸟教程

下图展示了频道 channel1 , 以及订阅这个频道的三个客户端 —— client2 、 client5 和 client1 之间的关系: 当有新消息通过 publish命令发送给频道 channel1 时, 这个消息就会被发送给订阅它的三个客户端:

3. 案例

先订阅后发布才能接收到消息。 Redis 客户端可以订阅任意数量的频道。

步骤:

  1. A-client 订阅渠道,一次可订阅多个。支持通配符:
subscribe channel*
subscribe channel1 channel2 channel3
  1. B-client 在channel2发布消息,发布后订阅者将会收到该条消息
publish channel2 hello-subscriber 

4. Springboot中使用Redis的sub/pub

  1. Redis的基础信息配置——application.properties
spring.redis.host=127.0.0.1 #  Redis主机ip
spring.redis.port=7989 # Redis服务端口
spring.redis.password= # Redis密码,无的话,留空
  1. 订阅监听的配置类——RedisPubSubConfig.java
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;

/**
 * Redis订阅监听
 */
@Configuration
public class RedisPubSubConfig {

	/**
	 * 1. 使用默认的工厂 初始化redis操作模板
	 */
	@Bean
	StringRedisTemplate template(RedisConnectionFactory connectionFactory) {
		return new StringRedisTemplate(connectionFactory);
	}

	/**
	 * 2. 创建消息监听器适配器,绑定消息处理器,利用反射技术调用消息处理器的业务方法
	 */
	@Bean
	MessageListenerAdapter listenerAdapter(RedisReceiverHandler redisReceiverHandler) {
		System.out.println("消息适配器进来了");
		return new MessageListenerAdapter(redisReceiverHandler, "handle");
	}

	/**
	 * 3. 创建redis消息监听器容器
	 * 可以添加多个监听不同话题的redis监听器,只需要把消息监听器和相应的消息订阅处理器绑定,该消息监听器
	 * 通过反射技术调用消息订阅处理器的相关方法进行一些业务处理
	 * @param connectionFactory 默认的连接工厂
	 * @param listenerAdapter 消息监听器适配器
	 * @return 消息监听容器
	 */
	@Bean
	RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,MessageListenerAdapter listenerAdapter) {
		// 创建消息监听容器
		RedisMessageListenerContainer container = new RedisMessageListenerContainer();
		// 设置连接工厂
		container.setConnectionFactory(connectionFactory);

		//可以添加多个 messageListener
		container.addMessageListener(listenerAdapter, new PatternTopic("index"));

		return container;
	}	
}
  1. 收到消息后的处理类——RedisReceiverHandler.java
import org.springframework.stereotype.Component;

/**
 * 具体的消息处理类
 */
@Component
public class RedisReceiverHandler {

	/**
	 * 消息处理方法
	 * @param message 发布者发布的消息
	 */
	public void handle(String message) {
		System.out.println("消息来了:"+message);
		//这里是收到通道的消息之后执行的方法
	}
}
  1. 消息发布类——RedisPublishMsg.java
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

/**
 * 定时器,用于定时发布消息
 */
@EnableScheduling
@Component
public class RedisPublishMsg {

	@Autowired
	private StringRedisTemplate stringRedisTemplate;

	/**
	 *  定时通过redis操作模板,向redis消息队列index通道发布消息
	 */
	@Scheduled(fixedRate = 3000)
	public void pubMsg(){
		stringRedisTemplate.convertAndSend("index",String.valueOf(Math.random()));
	}
}