09. SDR 发布订阅
Redis 使用channel实现pub/sub(发布订阅)功能,RedisTemplate对此也提供了相应的实现。一个消息会被推送到所有订阅了此消息到终端。1. SDR 实现channel 流程1.1 开发流程创建消息监听器实现类,对接受到的消息做相应的业务处理创建消息监听适配器,对消息监听器做不同的设置,如序列化方式等。将消息监听适配器注册到spring容器中创建消息监听容器,并...
·
Redis 使用channel实现pub/sub(发布订阅)功能,RedisTemplate对此也提供了相应的实现。一个消息会被推送到所有订阅了此消息到终端。
1. SDR 实现channel 流程
1.1 开发流程
- 创建消息监听器实现类,对接受到的消息做相应的业务处理
- 创建消息监听适配器,对消息监听器做不同的设置,如序列化方式等。
- 将消息监听适配器注册到spring容器中
- 创建消息监听容器,并将消息监听适配器添加到消息监听容器中,并指定监听适配器所监听的channel正则表达式。
- 使用RedisTemplate 向channel 发送消息。
1.2 流程猜测
- spring-boot 启动时,会将消息监听器容器中注册的消息监听器所监听的channel注册到redis中
- 当redis的channel接收到消息后,会向监听的此channel的所有消息监听容器发送消息
- 消息监听容器接收到redis服务器推送的消息之后,会根据channel名称,分发给相应的MessageListener 处理
2. 测试用例
2.1 创建消息监听器
创建两个消息监听器,一个监听用string序列化的消息,一个监听用json序列化的消息。
public class StringRedisChannelListener implements MessageListener {
@Override
public void onMessage(Message message, byte[] pattern) {
String channel = this.converte(message.getChannel());
String content = this.converte(message.getBody());
System.out.println("[StringRedisChannelListener] channel:" + channel + ", content:" + content);
}
private String converte(byte[] bytes){
String str = null;
try {
str = new String(bytes,"UTF-8");
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
return str;
}
}
public class JsonRedisChannelListener implements MessageListener {
@Override
public void onMessage(Message message, byte[] pattern) {
String channel = this.converte(message.getChannel());
String content = this.converte(message.getBody());
System.out.println("[JsonRedisChannelListener] channel:" + channel + ", content:" + content);
}
private String converte(byte[] bytes){
String str = null;
try {
str = new String(bytes,"UTF-8");
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
return str;
}
}
2.2 spring容器中注册消息监听容器和消息监听器
消息监听容器接收到消息之后,会根据channel名称将消息分发到相应的消息将监听器上。但是需要注意的是,发送消息的RedisTemplate用的值序列化方式和消息监听适配器用的序列化方式要一致,否则即使分发到相应的监听器上,也无法正确地反序列化。
@Configuration
public class RedisConfig {
/**
* 自定义redis操作模板: key 为String,value为json
* @param redisConnectionFactory
* @return RedisTemplate
* @throws UnknownHostException
*/
@Bean("strKeyRedisTemplate")
public RedisTemplate<Object,Object> strKeyRedisTemplate(RedisConnectionFactory redisConnectionFactory) throws UnknownHostException{
RedisTemplate<Object,Object> template = new RedisTemplate<>();
template.setConnectionFactory(redisConnectionFactory);
template.setKeySerializer(new StringRedisSerializer());
template.setValueSerializer(new GenericJackson2JsonRedisSerializer());
return template;
}
/**
* 自定义redis 操作工具类: key 为String, value为压缩后的json
* @param redisConnectionFactory
* @return RedisTemplate
* @throws UnknownHostException
*/
@Bean("gzipRedisTemplate")
public RedisTemplate<Object, Object> gzipRedisTemplate(RedisConnectionFactory redisConnectionFactory) throws UnknownHostException{
RedisTemplate<Object,Object> template = new RedisTemplate<>();
template.setConnectionFactory(redisConnectionFactory);
template.setKeySerializer(new StringRedisSerializer());
template.setValueSerializer(new GzipJsonSerializer());
return template;
}
/**
* spring 容器中注册用于监听string序列化的Bean
* @return MessageListenerAdapter
*/
@Bean("stringRedisListenerAdapter")
MessageListenerAdapter stringRedisListenerAdapter(){
// 默认为string 序列化方式
return new MessageListenerAdapter(new StringRedisChannelListener());
}
/**
* spring 容器中注册用于监听json序列化消息的Bean
* @return MessageListenerAdapter
*/
@Bean("jsonRedisListenerAdapter")
MessageListenerAdapter jsonRedisListenerAdapter(){
// 设置json序列化方式
MessageListenerAdapter adapter = new MessageListenerAdapter(new JsonRedisChannelListener());
adapter.setSerializer(new GenericJackson2JsonRedisSerializer());
return adapter;
}
/**
* 定义Redis 消息监听容器。需要设置监听规则。redis将相应规则的消息,推送给容器。然后容器将相应规则的消息分发给相应的MessageListenerAdapter 处理
* @param redisConnectionFactory 从spring容器中获取
* @param stringRedisListenerAdapter 从spring 容器中获取自定义bean
* @param jsonRedisListenerAdapter 从spring容器中获取自定义bean
* @return
*/
@Bean
RedisMessageListenerContainer container(RedisConnectionFactory redisConnectionFactory,
MessageListenerAdapter stringRedisListenerAdapter,
MessageListenerAdapter jsonRedisListenerAdapter){
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(redisConnectionFactory);
//绑定监听适配器匹配规则,channel正则匹配string.*,则转发给stringRedisListenerAdapter 处理, 正则匹配 json.*,则转发给 jsonRedisListenerAdapter 处理
container.addMessageListener(stringRedisListenerAdapter, new PatternTopic("string.*"));
container.addMessageListener(jsonRedisListenerAdapter, new PatternTopic("json.*"));
return container;
}
}
2.3 测试用例
一定要注意的,发送消息的ReidsTemplate和消息监听适配器的序列化方式要一致。否则将消息将无法进行解析。通常而讲,发布的消息都是字符串。
@RunWith(SpringRunner.class)
@SpringBootTest
public class RedisChannelTest {
@Autowired
private StringRedisTemplate redisTemplate;
@Autowired
private RedisTemplate strKeyRedisTemplate;
@Test
public void send_message1(){
redisTemplate.convertAndSend("string.info","hello, string.info");
redisTemplate.convertAndSend("string.warn","hello, string.warn");
strKeyRedisTemplate.convertAndSend("json.user", new UserPO(1001, "zhangshan",20, "man"));
strKeyRedisTemplate.convertAndSend("json.user", new UserPO(1002, "lisi",22, "man"));
}
/** redisTemplate 只需将valueSerializer 设置为MessageListenerAdapter中设置的序列方式即可 */
@Test
public void send_message2(){
redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());
redisTemplate.convertAndSend("json.user", new UserPO(1001, "zhangshan",20, "man"));
redisTemplate.convertAndSend("json.user", new UserPO(1002, "lisi",22, "man"));
}
}
2.4 测试输出
[StringRedisChannelListener] channel:string.info, content:hello, string.info
[StringRedisChannelListener] channel:string.warn, content:hello, string.warn
[JsonRedisChannelListener] channel:json.user, content:{"@class":"org.zongf.sboot.redis.po.UserPO","id":1001,"name":"zhangshan","age":20,"sex":"man"}
[JsonRedisChannelListener] channel:json.user, content:{"@class":"org.zongf.sboot.redis.po.UserPO","id":1002,"name":"lisi","age":22,"sex":"man"}
更多推荐
已为社区贡献4条内容
所有评论(0)