09. SDR 发布订阅

￡神魔★判官ぃ 2022-01-31 18:09 178阅读 0赞

Redis 使用channel实现pub/sub(发布订阅)功能，RedisTemplate对此也提供了相应的实现。一个消息会被推送到所有订阅了此消息到终端。

## 1. SDR 实现channel 流程 ##

### 1.1 开发流程 ###

1.  创建消息监听器实现类，对接受到的消息做相应的业务处理
2.  创建消息监听适配器，对消息监听器做不同的设置，如序列化方式等。
3.  将消息监听适配器注册到spring容器中
4.  创建消息监听容器，并将消息监听适配器添加到消息监听容器中，并指定监听适配器所监听的channel正则表达式。
5.  使用RedisTemplate 向channel 发送消息。

### 1.2 流程猜测 ###

1.  spring-boot 启动时，会将消息监听器容器中注册的消息监听器所监听的channel注册到redis中
2.  当redis的channel接收到消息后，会向监听的此channel的所有消息监听容器发送消息
3.  消息监听容器接收到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"}