赞
踩
当我们在多个集群应用中使用到本地缓存时,在数据库数据得到更新后,为保持各个副本当前被修改的数据与数据库数据保持同步,在数据被操作后向其他集群应用发出被更新数据的通知,使其删除;下次当其他应用请求该被更新的数据时,应用会到数据库去取,也就是最新的数据,从而使得被更新数据与数据库保持同步!
能实现发送与接收信息的中间介有很多,比如:RocketMQ、RabbitMQ、ActiveMQ、Kafka等,本次主要简单介绍Redis的推送与订阅功能并集成Spring Boot的实现。
1.添加SpringBoot 集成Redis maven依赖
- <dependency>
- <groupId>org.springframework.boot</groupId>
- <artifactId>spring-boot-starter-data-redis</artifactId>
- </dependency>
2.配置Redis配置 RedisConfig.java
- @Configuration
- public class RedisConfig {
- @Value("${redis.server.nodes}")
- private String redisServerNodes;
-
- @Value("${redis.server.password}")
- private String redisServerPassword;
-
- //Redis集群配置,单机的redis注释掉该方法,在application配置文件里面配置就可以了
- @Bean
- public RedisClusterConfiguration getRedisClusterConfiguration() {
-
- RedisClusterConfiguration redisClusterConfiguration = new RedisClusterConfiguration();
-
- String[] serverArray = redisServerNodes.split(",");
- Set<RedisNode> nodes = new HashSet<RedisNode>();
- for (String ipPort : serverArray) {
- String[] ipAndPort = ipPort.split(":");
- nodes.add(new RedisNode(ipAndPort[0].trim(), Integer.parseInt(ipAndPort[1])));
- }
- redisClusterConfiguration.setClusterNodes(nodes);
- RedisPassword pwd = RedisPassword.of(redisServerPassword);
- redisClusterConfiguration.setPassword(pwd);
- return redisClusterConfiguration;
- }
-
- //指定redisTemplate类型,如下为<String, Object>
- @Bean
- public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
- RedisTemplate<String, Object> template = new RedisTemplate();
- template.setConnectionFactory(redisConnectionFactory);
- // 使用JSON格式序列化对象,对缓存数据key和value进行转换
- Jackson2JsonRedisSerializer<Object> jacksonSeial = new Jackson2JsonRedisSerializer<>(Object.class);
- // 解决查询缓存转换异常的问题
- ObjectMapper objectMapper = new ObjectMapper();
- objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
- objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
- jacksonSeial.setObjectMapper(objectMapper);
-
- // 设置RedisTemplate模板API的序列化方式为JSON
- template.setDefaultSerializer(jacksonSeial);
- return template;
- }
-
- /**
- * Redis消息监听器容器
- * 这个容器加载了RedisConnectionFactory和消息监听器
- * 可添加多个不同话题的redis监听器,需要将消息监听器和消息频道绑定,
- * 通过反射调用消息订阅处理器的相关方法进行业务处理
- *
- * @param redisConnectionFactory 连接工厂
- * @param listener Redis消息监听器
- * @param MessageListenerAdapter Redis消息监听适配器
- * @return RedisMessageListenerContainer 消息监听容器
- */
- @Bean
- @SuppressWarnings("all")
- public RedisMessageListenerContainer container(RedisConnectionFactory redisConnectionFactory,
- RedisMessageListener listener,
- MessageListenerAdapter adapter) {
- RedisMessageListenerContainer container = new RedisMessageListenerContainer();
- container.setConnectionFactory(redisConnectionFactory);
- // 所有的订阅消息,都需要在这里进行注册绑定
- // new PatternTopic(TOPIC_NAME1) 表示发布信息的频道
- // 可以添加多个频道以及配置不同的频道
- container.addMessageListener(listener, new PatternTopic(SystemConstants.TOPIC_NAME1));
- container.addMessageListener(adapter, new PatternTopic(SystemConstants.TOPIC_NAME2));
-
- /**
- * 设置序列化对象
- * 特别注意:1. 发布的时候和订阅方都需要设置序列化
- * 2. 设置序列化对象必须放在 {加入消息监听器} 这步后面,不然接收器接收不到消息
- */
- Jackson2JsonRedisSerializer seria = new Jackson2JsonRedisSerializer(Object.class);
- ObjectMapper objectMapper = new ObjectMapper();
- objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
- objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
- seria.setObjectMapper(objectMapper);
- container.setTopicSerializer(seria);
- return container;
- }
-
- /**
- * 这个地方是给messageListenerAdapter 传入一个消息接受的处理器,利用反射的方法调用“receiveMessage”
- * 也有好几个重载方法,这边默认调用处理器的方法 叫OnMessage
- *
- * @param redisMessageReceiver 消息接收对象
- * @return 消息监听适配器
- */
- @Bean
- public MessageListenerAdapter listenerAdapter(RedisMessageReceiver redisMessageReceiver) {
- MessageListenerAdapter receiveMessage = new MessageListenerAdapter(redisMessageReceiver, "onMessage");
- Jackson2JsonRedisSerializer<Object> seria = new Jackson2JsonRedisSerializer<>(Object.class);
- ObjectMapper objectMapper = new ObjectMapper();
- objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
- objectMapper.activateDefaultTyping(LaissezFaireSubTypeValidator.instance, ObjectMapper.DefaultTyping.NON_FINAL);
- seria.setObjectMapper(objectMapper);
- receiveMessage.setSerializer(seria);
- return receiveMessage;
- }
- }
3.Redis的订阅主要有两种实现方式:
方式一:编写Redis监听类RedisMessageListener,实现Redis的监听接口MessageListener,并重写onMessage方法
方式二:编写Redis消息监听适配器类,并在RedisConfig.java中配置消息监听适配器bean
方式一 与 方式二 主要是实现订阅Redis推送的消息后的具体操作,这两种方式可以同时使用来订阅多个频道里的消息
- //方式一:
- @Slf4j
- @Component
- public class RedisMessageListener implements MessageListener {
-
- @Autowired
- private RedisTemplate<String, Object> redisTemplate;
- @Autowired
- private CacheManager cacheManager;
-
- @Override
- public void onMessage(Message message, byte[] pattern) {
- // 接收的topic
- log.info("接收消息频道:" + new String(pattern));
- //序列化对象(特别注意:发布的时候需要设置序列化;订阅方也需要设置序列化)
- MessageDto<?> messageDto = (MessageDto<?>) redisTemplate.getValueSerializer().deserialize(message.getBody());
- //MessageDto<T>为自己编写的一个消息对象类(如自定义有:String date,String title,T content 等属性)
- log.info("接收消息内容:{}", messageDto);
- //根据消息内容做具体业务逻辑。。。。。。。。。
- //。。。。。。。。。。。。。。。。。。。。。。
- }
- }
- //方式二
- @Slf4j
- @Component
- public class RedisMessageReceiver {
-
- @Autowired
- private RedisTemplate<String,Object> redisTemplate;
-
- /**
- * 方法命名规则必须为function(Object messageDto) / function(Object messageDto,String topic)
- * @param messageDto 消息对象
- * @param topic 消息频道名称
- */
- public void onMessage(Object messageDto , String topic) {
- // 接收消息频道
- log.info("接收消息频道:" + topic);
- //接收消息内容
- log.info("接收消息内容:{}",messageDto);
- }
- }
4.编写Redis消息的推送工具类,在需要推送消息的地方使用来向Redis推送消息(如:操作数据的地方)
- @Slf4j
- @Component
- public class RedisMessageUtils {
-
- @Autowired
- private RedisTemplate<String,Object> redisTemplate;
-
- /**
- * 向通道发布消息
- */
- public void sendMessage(String topic, Object message) {
- if (!StringUtils.hasText(topic)) {
- return;
- }
- try {
- redisTemplate.convertAndSend(topic, message);
- log.info("发送消息成功,topic:{},message:{}", topic, message);
- } catch (Exception e) {
- log.info("发送消息失败,topic:{},message:{}", topic, message);
- e.printStackTrace();
- }
- }
- }
5.使用:
- @RestController
- @RequestMapping("/user")
- public class UserController{
-
- @Autowired
- private UserService userService;
-
- @PostMapping("/getUsers")
- public List<User> queryUsers(@RequestBody UserDto userDto){
- List<User> users=userService.queryUsers(userDto);
- //发送测试消息
- RedisMessageUtils.sendMessage(SystemConstants.TOPIC_NAME2, new MessageDto());
- return users;
- }
- }
成功示例:
- 2099-12-31 23:59:59.999 [http-nio-8888-exec-1] INFO com.xxx.yyy.util.RedisMessageUtils : 发送消息成功,topic:TOPIC2,message:MessageDto(data=null, title=null, content=null)
- 2099-12-31 23:59:59.999 [container-2] INFO com.xxx.yyy.zzz.RedisMessageReceiver : 接收消息频道:TOPIC2
- 2099-12-31 23:59:59.999 [container-2] INFO com.xxx.yyy.zzz.RedisMessageReceiver : 接收消息内容:MessageDto(date=null, title=null, content=null)
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。