SpringBoot使用Redis进行消息的发布订阅原创

　　/**
　　* Redis 相关的配置，包含推送，以及对象编码的定义
　　*/
　　@Configuration
　　public class RedisConfig {
　　public @Bean
　　RedisTemplate redisTemplate(RedisConnectionFactory connectionFactory) {
　　RedisTemplate template = new RedisTemplate<>();
　　template.setConnectionFactory(connectionFactory);
　　template.setDefaultSerializer(new GenericJackson2JsonRedisSerializer());
　　template.setKeySerializer(new StringRedisSerializer());
　　template.setHashKeySerializer(new GenericJackson2JsonRedisSerializer());
　　template.setValueSerializer(new GenericJackson2JsonRedisSerializer());
　　return template;
　　}
　　@Bean
　　ChannelTopic topic() {
　　return new ChannelTopic(＂messageQueue＂);
　　}
　　@Bean
　　RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
　　MessageListenerAdapter listenerAdapter) {
　　RedisMessageListenerContainer container = new RedisMessageListenerContainer();
　　container.setConnectionFactory(connectionFactory);
　　container.addMessageListener(listenerAdapter, new PatternTopic(＂messageQueue＂));
　　return container;
　　}
　　@Bean
　　MessageListenerAdapter listenerAdapter(Receiver receiver) {
　　return new MessageListenerAdapter(receiver, ＂receiveMessage＂);
　　}
　　}
　　public enum BusinessTypeEnum {
　　//修改SystemConfig缓存
　　UPDATE_SYSTEM_CONFIG;
　　}
　　//发布者接口
　　public interface MessagePublisher {
　　void publish(String message);
　　}
　　@Component
　　public class RedisMessagePublisher implements MessagePublisher{
　　private final static Logger logger = LoggerFactory.getLogger(RedisMessagePublisher.class);
　　@Autowired
　　private RedisTemplate redisTemplate;
　　@Autowired
　　private ChannelTopic topic;
　　@Override
　　public void publish(String message) {
　　logger.info(＂推送信息：＂+message);
　　redisTemplate.convertAndSend(topic.getTopic(), message);
　　}
　　}
　　//订阅接口
　　public interface Receiver { void receiveMessage(String message); }
　　@Component
　　public class MessageReceiver implements Receiver {
　　private final static Logger logger = LoggerFactory.getLogger(MessageReceiver.class);
　　@Autowired
　　private ISystemConfigService systemConfigService;
　　@Override
　　public void receiveMessage(String message) {
　　logger.info(＂消息接收：＂+message);
　　JSONObject object = JSONObject.parseObject(message);
　　String businessType = object.getString(＂businessType＂);
　　logger.info(＂处理业务类型：businessType=＂+businessType);
　　if(BusinessTypeEnum.UPDATE_SYSTEM_CONFIG.toString().equals(businessType)){//处理SystemConfig缓存
　　String jsonStr = object.getString(＂jsonStr＂);
　　if(jsonStr!=null&&!＂＂.equals(jsonStr)){
　　JSONObject json = JSONObject.parseObject(object.getString(＂jsonStr＂));
　　for(String key:json.keySet()){
　　String value = json.getString(key);
　　SystemConfig systemConfig = SystemConfigContainer.instance().getSystemConfig(key);
　　if(systemConfig!=null){
　　systemConfig.setKeyValue(value);
　　SystemConfigContainer.instance().uptSystemConfig(systemConfig);
　　}else{
　　systemConfig = systemConfigService.selectByPrimaryJianMing(key);
　　systemConfig.setKeyValue(value);
　　SystemConfigContainer.instance().uptSystemConfig(systemConfig);
　　}
　　}
　　}
　　}
　　}
　　}