SpringBoot整合Redis实现消息发布与订阅的示例代码
当我们在多个集群应用中使用到本地缓存时,在数据库数据得到更新后,为保持各个副本当前被修改的数据与数据库数据保持同步,在数据被操作后向其他集群应用发出被更新数据的通知,使其删除;下次当其他应用请求该被更新的数据时,应用会到数据库去取,也就是最新的数据,从而使得被更新数据与数据库保持同步!
能实现发送与接收信息的中间介有很多,比如: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;
}
@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));
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;
}
@Bean
public MessageListenerAdapter listenerAdapter(RedisMessageReceiver redisMessageReceiver) {
MessageListenerAdapter receiveMessage = new MessageListenerAdapter(printMessageReceiver, "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 data,String title,T content 等属性)
log.info("接收消息内容:{}", messageDto);
//根据消息内容做具体业务逻辑。。。。。。。。。
//。。。。。。。。。。。。。。。。。。。。。。
}
}
//方式二
@Slf4j
@Component
public class RedisMessageReceiver {
@Autowired
private RedisTemplate<String,Object> redisTemplate;
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(data=null, title=null, content=null)
以上就是SpringBoot整合Redis实现消息发布与订阅的示例代码的详细内容,更多关于SpringBoot Redis消息发布 订阅的资料请关注编程网其它相关文章!
免责声明:
① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。
② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341