redis做服务间通信工具的项目示例

public interface MessageHubService {
    /**
     * 生产消息
     */
    void producer(MessageForm messageForm);
    /**
     * 消费消息
     */
    void consumer(ConsumerAdapterForm adapterForm);
    /**
     * 检查topic、type合法性
     */
    void checkTopic(String topic, String type);
}

@Data
public class MessageForm {
    // 消息组件类型
    private String type;
    // 消息主题
    private String topic;
    private String message = "";
    // 消费者组
    private String group = "UPTOWN";
}

@Service
public class MessageHubServiceImpl implements MessageHubService, ApplicationContextAware {
    @Resource
    protected StringRedisTemplate stringRedisTemplate;
    public Map<String, MessageHubService> messageHubServiceMap = new ConcurrentHashMap<>();
    private ApplicationContext applicationContext;
    @PostConstruct
    public void init() {
        messageHubServiceMap.put(TopicTypeConstants.REDIS_PUBSUB_TYPE, applicationContext.getBean(RedisPubSubProcessor.class));
        messageHubServiceMap.put(TopicTypeConstants.REDIS_STREAM_TYPE, applicationContext.getBean(RedisQueueProcessor.class));
        messageHubServiceMap.put(TopicTypeConstants.REDIS_QUEUE_TYPE, applicationContext.getBean(RedisStreamProcessor.class));
    }
    public void checkTopic(String topic, String type) {
        if (!messageHubServiceMap.containsKey(type)) {
            throw new MatrixException("消息类型不支持");
        }
        List<String> whiteTopicList = stringRedisTemplate.opsForList().range(TopicTypeConstants.WHITE_TOPIC, 0, -1);
        if ((!ObjectUtils.isEmpty(whiteTopicList) && !whiteTopicList.contains(topic)) || ObjectUtils.isEmpty(whiteTopicList)) {
            throw new MatrixException("当前topic未配置");
        }
    }
    @Override
    public void producer(MessageForm messageForm) {
        this.checkTopic(messageForm.getTopic(), messageForm.getType());
        this.messageHubServiceMap.get(messageForm.getType()).producer(messageForm);
    }
    /**
     * 消费者创建通过注解，已校验topic合法性
     */
    @Override
    public void consumer(ConsumerAdapterForm messageForm) {
        this.messageHubServiceMap.get(messageForm.getType()).consumer(messageForm);
    }
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}

@Scope(proxyMode = ScopedProxyMode.TARGET_CLASS)
@Service("redisPubSubProcessor")
public class RedisPubSubProcessor extends MessageHubServiceImpl {
    @Override
    public void producer(MessageForm messageForm) {
        // 具体生产逻辑
    }
    @Override
    public void consumer(ConsumerAdapterForm messageForm) {
        // 具体消费逻辑
    }
}

@Slf4j
@Component
public class ConsumerConfig implements DisposableBean, SmartInstantiationAwareBeanPostProcessor {
    @Resource(name = "messageHubServiceImpl")
    MessageHubService messageHubService;
    @Bean(name = "redisPubSubConsumerMap")
    public Map<String, MessageListenerAdapter> redisPubSubConsumerMap() {
        return new ConcurrentHashMap<>();
    }
    @Override
    public void destroy() throws Exception {
    }
    @Override
    public Object getEarlyBeanReference(Object bean, String beanName) throws BeansException {
        Method[] methods = ReflectionUtils.getAllDeclaredMethods(bean.getClass());
        for (Method method : methods) {
            MessageHub annotation = AnnotationUtils.findAnnotation(method, MessageHub.class);
            if (annotation == null) {
                continue;
            }
            String resolveTopic = annotation.topic();
            try {
                messageHubService.checkTopic(resolveTopic, annotation.type());
            } catch (Exception e) {
                throw new Error(e.getMessage());
            }
            ConsumerAdapterForm adapterForm = new ConsumerAdapterForm();
            adapterForm.setBean(bean);
            adapterForm.setInvokeMethod(method);
            adapterForm.setTopic(resolveTopic);
            adapterForm.setType(annotation.type());
            adapterForm.setGroup(annotation.group());
            messageHubService.consumer(adapterForm);
        }
        return bean;
    }
}

@Slf4j
@Service
public class TopicReloadTask extends TimerTask {
    @Resource
    StringRedisTemplate stringRedisTemplate;
    @Resource
    EntityManager entityManager;
    public final String TOPIC_SQL = " select * from MESSAGEHUB_TOPIC ";
    public final String LUA_SCRIPT =
                "redis.call('del', 'MESSAGEHUB_TOPIC')" +
                "local topics = KEYS " +
                "for i, v in pairs(topics) do " +
                "  redis.call('lpush', 'MESSAGEHUB_TOPIC', v) " +
                "end";
    @Override
    public void run() {
        try {
            List<String> topics = this.getQueryResult(TOPIC_SQL, MessageHubTopicBean.class).stream().map(MessageHubTopicBean::getTopic).collect(Collectors.toList());
            if (!ObjectUtils.isEmpty(topics)) {
                DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>(LUA_SCRIPT, Long.class);
                Long result = stringRedisTemplate.execute(redisScript, topics);
                log.info("reload topic finish");
            }
        } catch (Throwable t) {
            log.error("messagehub topic reload error", t);
        }
    }
    private <T> List<T> getQueryResult(String sql, Class<T> clazz) {
        Query dataQuery = entityManager.createNativeQuery(sql, clazz);
        List<T> result = new ArrayList<>();
        List<Object> list = dataQuery.getResultList();
        for (Object o : list) {
            result.add((T) o);
        }
        return result;
    }
}