springboot+redis自定义注解实现发布订阅的实现代码

@Slf4j
@Service
public class RedisConsumerDemo {
    @MessageHub(topic = "${uptown.topic}", type = "REDIS_PUBSUB")
    public void consumer(Object message) {
        log.info("pubsub info {} ", message);
    }   
}

@Bean 
public MessageListenerAdapter smsExpirationListener(TestSubscriber messageListener) {
    return new MessageListenerAdapter(messageListener, "onMessage");
}

@Component
public class TestSubscriber implements MessageListener {
    @Override
    public void onMessage(Message message, byte[] pattern) {
        log.info("get data :{}", msg);
    }
}

@Configuration
public class RedisConfig {
    @Bean
    public RedisMessageListenerContainer container(
        RedisConnectionFactory redisConnectionFactory,
        MessageListenerAdapter smsExpirationListener) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(redisConnectionFactory);
        container.addMessageListener(smsExpirationListener, new PatternTopic("test"));
        return container;
    }
}

/**
 * 生产消息
 */
Integer producer(MessageForm messageForm);
/**
 * 消费消息
 */
void consumer(ConsumerAdapterForm adapterForm);

stringRedisTemplate.convertAndSend(messageForm.getTopic(), messageForm.getMessage());

/**
 * Create a new {@link MessageListenerAdapter} for the given delegate.
 *
 * @param delegate the delegate object
 * @param defaultListenerMethod method to call when a message comes
 * @see #getListenerMethodName
 */
public MessageListenerAdapter(Object delegate, String defaultListenerMethod) {
   this(delegate);
   setDefaultListenerMethod(defaultListenerMethod);
}

MessageHub annotation = AnnotationUtils.findAnnotation(method, MessageHub.class);
if (annotation == null) {
    continue;
}

@Scope(proxyMode = ScopedProxyMode.TARGET_CLASS)
@Service("redisPubSubProcessor")
public class RedisPubSubProcessor extends MessageHubServiceImpl {
    @Resource
    RedisMessageListenerContainer redisPubSubContainer;
    @Override
    public void produce(ProducerAdapterForm producerAdapterForm) {
        stringRedisTemplate.convertAndSend(producerAdapterForm.getTopic(), producerAdapterForm.getMessage());
    }
    @Override
    public void consume(ConsumerAdapterForm messageForm) {
        MessageListenerAdapter adapter = new MessageListenerAdapter(messageForm.getBean(), messageForm.getInvokeMethod().getName());
        adapter.afterPropertiesSet();
        redisPubSubContainer.addMessageListener(adapter, new PatternTopic(messageForm.getTopic()));
    }
    @Bean
    public RedisMessageListenerContainer redisPubSubContainer(RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        return container;
    }
}