SpringBoot 集成 redis 发布订阅_springboot整合redis 发布订阅

 redis 配置类

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.jsontype.impl.LaissezFaireSubTypeValidator;
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
import org.springframework.cache.annotation.CachingConfigurerSupport;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
 
@Configuration
@EnableCaching
@AutoConfigureBefore(RedisAutoConfiguration.class)
public class RedisConfiguration extends CachingConfigurerSupport {
    /**
     *
     * 对 redis 操作做一些预处理
     *
     */
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisSerializer<Object> serializer = redisSerializer();
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisConnectionFactory);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(serializer);
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(serializer);
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }
 
    @Bean
    public RedisSerializer<Object> redisSerializer() {
        //创建JSON序列化器
        Jackson2JsonRedisSerializer<Object> serializer = new Jackson2JsonRedisSerializer<>(Object.class);
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        //必须设置，否则无法将JSON转化为对象，会转化成Map类型
        objectMapper.activateDefaultTyping(LaissezFaireSubTypeValidator.instance, ObjectMapper.DefaultTyping.NON_FINAL);
        serializer.setObjectMapper(objectMapper);
        return serializer;
    }
 
//    @Bean
//    public RedisCacheManager redisCacheManager(RedisConnectionFactory redisConnectionFactory) {
//        RedisCacheWriter redisCacheWriter = RedisCacheWriter.nonLockingRedisCacheWriter(redisConnectionFactory);
//        //设置Redis缓存有效期为1天
//        RedisCacheConfiguration redisCacheConfiguration = RedisCacheConfiguration.defaultCacheConfig()
//                .serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(redisSerializer())).entryTtl(Duration.ofDays(-1));
//        return new RedisCacheManager(redisCacheWriter, redisCacheConfiguration);
//    }
 
 
}

redis-topic配置

cloud:
  redis:
    topic-list:
      - name: Custom
        listener-class: com.huxiaosu.cloud.system.listener.CustomListener

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
 
import java.io.Serializable;
import java.util.List;
 
 
@Data
@Component
@ConfigurationProperties(ScanMenuConfig.PREFIX)
public class CloudRedisProperties implements Serializable {
    public static final String PREFIX = "cloud.redis";
    private List<Topic> topicList;
    @Data
    public static class Topic{
        private String name;
        private String listenerClass;
    }
}

import cn.hutool.core.collection.CollUtil;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.jsontype.impl.LaissezFaireSubTypeValidator;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskExecutor;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
import java.util.List;
 
 
@Slf4j
@Configuration
public class RedisPubSubConfig {
    @Autowired
    private CloudRedisProperties cloudRedisProperties;
    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory redisConnectionFactory) {
        RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();
        redisMessageListenerContainer.setConnectionFactory(redisConnectionFactory);
        // 订阅多个频道
        List<CloudRedisProperties.Topic> topicList = cloudRedisProperties.getTopicList();
        if(CollUtil.isEmpty(topicList)) {
            return redisMessageListenerContainer;
        }
        for(CloudRedisProperties.Topic topic: topicList){
            try {
                Class<?> aClass = Class.forName(topic.getListenerClass());
                // onMessage 为 listenerClass 中处理消息的方法
                MessageListenerAdapter getMessage = new MessageListenerAdapter(aClass.newInstance(), "onMessage");
                redisMessageListenerContainer.addMessageListener(getMessage, new PatternTopic(topic.getName()));
                log.info("add listener [{}] on topic [{}]",topic.getListenerClass(),topic.getName());
            }catch (Exception e){
                log.error("add redis listener error",e);
            }
 
        }
        //序列化对象（特别注意：发布的时候需要设置序列化；订阅方也需要设置序列化）
        Jackson2JsonRedisSerializer 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);
        redisMessageListenerContainer.setTopicSerializer(seria);
        redisMessageListenerContainer.setTaskExecutor(executor());
        return redisMessageListenerContainer;
    }
    @Bean
    public TaskExecutor executor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(Runtime.getRuntime().availableProcessors());
        executor.setMaxPoolSize(Runtime.getRuntime().availableProcessors() * 2);
        executor.setQueueCapacity(100);
        executor.initialize();
        return executor;
    }
}

import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
 
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
 
 
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
 
 
@Slf4j
public class CustomListener implements MessageListener {
   
    @Override
    public void onMessage(@NotNull Message message, byte[] topic) {
        
        log.info("CustomListener ... {}",message.toString);
       
        log.info("topic ... {}",new String(topic));
    }
 
   
 
}

需要注意的是 CustomListener 是通过newInstance创建的，不能直接注入 Bean 对象；如果想要使用 SpringBean 需要使用 SpringContextHolder 来获取 Bean 对象