编程宝库 - 技术改变世界

编程宝库

Java编程技术

Mybatis-Plus使用updateById()、update()将字段更新为null

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;
  }

   /**
   * 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(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;

  /**
   * 方法命名规则必须为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(data=null, title=null, content=null)

以上就是SpringBoot整合Redis实现消息发布与订阅的示例代码的详细内容，更多关于SpringBoot Redis消息发布订阅的资料请关注编程宝库其它相关文章！

下一节：Java多线程事务回滚@Transactional失效处理方案

Java编程技术

背景介绍1，最近有一个大数据量插入的操作入库的业务场景，需要先做一些其他修改操作，然后在执行插入操作，由于插入数据可能会很多，用到多线程去拆分数据并行处理来提高响应 ...

Mybatis-Plus使用updateById()、update()将字段更新为null

Java多线程事务回滚@Transactional失效处理方案