redis 发布订阅实现方式

maven引入

            org.springframework.boot
            spring-boot-starter-data-redis
        
        
            redis.clients
            jedis
        

redis 配置项 通过反射自定义注解来获取所有订阅者的监听

@Configuration
public class RedisMessageConfig {

    @Resource
    private RequestMappingHandlerMapping requestMappingHandlerMapping;


    
    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        //这个container 可以添加多个 messageListener
        //获取所有用RedisMessageListener自定义注解的类名
        String[] beanNameArr = SpringUtils.getApplicationContext().getBeanNamesForAnnotation(RedisMessageListener.class);
        for (String key : beanNameArr) {
            //获取对象实体类
            Class clazz = SpringUtils.getBean(key).getClass();
            // 根据对象获取注解属性值
            RedisMessageListener annotation = clazz.getAnnotation(RedisMessageListener.class);
            if(annotation.ignore()){
                String topicName = annotation.topic();
                MessageListener messageListener = (MessageListener) SpringUtils.getBean(clazz);
                container.addMessageListener(messageListener, new PatternTopic(topicName));
            }
        }
        return container;
    }

    
    @Bean
    StringRedisTemplate stringRedisTemplate(RedisConnectionFactory connectionFactory) {
        return new StringRedisTemplate(connectionFactory);
    }

    @Bean
    public RedisTemplate redisTemplate(RedisConnectionFactory factory) {
        StringRedisTemplate template = new StringRedisTemplate(factory);

        //定义value的序列化方式
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);

        template.setValueSerializer(jackson2JsonRedisSerializer);
        template.setHashKeySerializer(jackson2JsonRedisSerializer);
        template.afterPropertiesSet();
        return template;
    }
}

自定义注解

@Documented
@Inherited
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD, ElementType.TYPE})
public @interface RedisMessageListener {

    String topic() default "";

    boolean ignore() default false;
}

发布者

@ApiOperation(value = "推送消息",notes = "推送消息")
    @PostMapping("/publishMessage")
    public R publishMessage(@RequestParam("message") String message,@RequestParam("topicName") String topicName) {
        stringRedisTemplate.convertAndSend(topicName,message);
        return R.success(true);
    }

订阅者 1

@Slf4j
@Component
@RedisMessageListener(topic = RedisTopicConstant.ORDER_MESSAGE_TOPIC,ignore = true)
public class OrderMessageListener extends MessageListenerAdapter implements MessageListener {

    @Override
    public void onMessage(Message message, byte[] pattern){
        log.info("=====我是ORDER监听",message.toString());
    }
}

订阅者 2

@Slf4j
@Component
@RedisMessageListener(topic = RedisTopicConstant.DEVICE_MESSAGE_TOPIC,ignore = false)
public class DeviceMessageListener extends MessageListenerAdapter implements MessageListener {

    @Override
    public void onMessage(Message message, byte[] pattern){
        log.info("=====我是DEVICE监听",message.toString());
    }
}

工具类

public final class SpringUtils {
    private SpringUtils() {
    }

    
    public static SpringUtils getInstance() {
        return SpringUtilsHolder.INSTANCE;
    }

    private static ApplicationContext applicationContext;
    private static ApplicationContext parentApplicationContext;

    public static ApplicationContext getApplicationContext() {
        return applicationContext;
    }

    public static void setApplicationContext(ApplicationContext ctx) {
        Assert.notNull(ctx, "SpringUtil injection ApplicationContext is null");
        applicationContext = ctx;
        parentApplicationContext = ctx.getParent();
    }

    public static Object getBean(String name) {
        Assert.hasText(name, "SpringUtil name is null or empty");
        try {
            return applicationContext.getBean(name);
        } catch (Exception e) {
            return parentApplicationContext.getBean(name);
        }
    }

    public static  T getBean(String name, Class type) {
        Assert.hasText(name, "SpringUtil name is null or empty");
        Assert.notNull(type, "SpringUtil type is null");
        try {
            return applicationContext.getBean(name, type);
        } catch (Exception e) {
            return parentApplicationContext.getBean(name, type);
        }
    }

    public static  T getBean(Class type) {
        Assert.notNull(type, "SpringUtil type is null");
        try {
            return applicationContext.getBean(type);
        } catch (Exception e) {
            return parentApplicationContext.getBean(type);
        }
    }

    public static  Map getBeansOfType(Class type) {
        Assert.notNull(type, "SpringUtil type is null");
        try {
            return applicationContext.getBeansOfType(type);
        } catch (Exception e) {
            return parentApplicationContext.getBeansOfType(type);
        }
    }

    public static ApplicationContext publishEvent(Object event) {
        applicationContext.publishEvent(event);
        return applicationContext;
    }


    
    private static class SpringUtilsHolder {
        
        private static final SpringUtils INSTANCE = new SpringUtils();
    }

}