为主题动态创建消息监听器

Question

我正在分析一个关于创建通用消费者库的问题，该库可以部署在多个微服务中（所有微服务都基于 spring）。要求是有大约 15-20 个主题可以监听。如果我们使用基于注解的 kafka 监听器，我们需要为每个微服务添加更多代码。有什么方法可以让我们根据一些 xml 文件动态创建消费者，每个消费者都可以注入这些数据

主题
groupid
分区
过滤器（如果有）

有注解，设计很死板。我唯一能想到的方法是，我们可以在解析xml配置后创建消息监听器，每个主题都有自己的concurrentmessagelistenercontainer。

是否有更好的替代方法可以使用 spring？

P.S：我对 spring 和 kafka 有点陌生。如果在解释要求时出现混淆，请告诉我

谢谢，拉贾塞卡

Answer 1

也许您可以使用主题模式。看看consumer properties。例如。听众

@KafkaListener(topicPattern = "topic1|topic2")

会收听topic1和topic2。

如果您需要动态创建侦听器，则必须格外小心，因为您必须关闭它。

我会使用与 spring 的 KafkaListenerAnnotationBeanPostProcessor 类似的方法。这个 post 处理器负责处理 @KafkaListeners.

这是它如何工作的建议：

public class DynamicEndpointRegistrar {

    private BeanFactory beanFactory;
    private KafkaListenerContainerFactory<?> containerFactory;
    private KafkaListenerEndpointRegistry endpointRegistry;
    private MessageHandlerMethodFactory messageHandlerMethodFactory;

    public DynamicEndpointRegistrar(BeanFactory beanFactory,
            KafkaListenerContainerFactory<?> containerFactory,
            KafkaListenerEndpointRegistry endpointRegistry, MessageHandlerMethodFactory messageHandlerMethodFactory) {
        this.beanFactory = beanFactory;
        this.containerFactory = containerFactory;
        this.endpointRegistry = endpointRegistry;
        this.messageHandlerMethodFactory = messageHandlerMethodFactory;
    }

    public void registerMethodEndpoint(String endpointId, Object bean, Method method, Properties consumerProperties,
            String... topics) throws Exception {
        KafkaListenerEndpointRegistrar registrar = new KafkaListenerEndpointRegistrar();
        registrar.setBeanFactory(beanFactory);
        registrar.setContainerFactory(containerFactory);
        registrar.setEndpointRegistry(endpointRegistry);
        registrar.setMessageHandlerMethodFactory(messageHandlerMethodFactory);

        MethodKafkaListenerEndpoint<Integer, String> endpoint = new MethodKafkaListenerEndpoint<>();
        endpoint.setBeanFactory(beanFactory);
        endpoint.setMessageHandlerMethodFactory(messageHandlerMethodFactory);

        endpoint.setId(endpointId);
        endpoint.setGroupId(consumerProperties.getProperty(ConsumerConfig.GROUP_ID_CONFIG));
        endpoint.setBean(bean);
        endpoint.setMethod(method);

        endpoint.setConsumerProperties(consumerProperties);
        endpoint.setTopics(topics);

        registrar.registerEndpoint(endpoint);
        registrar.afterPropertiesSet();
    }

}

然后您应该能够动态注册侦听器。例如

DynamicEndpointRegistrar dynamicEndpointRegistrar = ...;
MyConsumer myConsumer = ...; // create an instance of your consumer
Properties properties = ...; // consumer properties

// the method that should be invoked
// (the method that's normally annotated with KafkaListener)
Method method = MyConsumer.class.getDeclaredMethod("consume", String.class);

dynamicEndpointRegistrar.registerMethodEndpoint("endpointId", myConsumer, method, properties, "topic");

为主题动态创建消息监听器

Create message liseners dynamically for the topics

apache-kafka

spring-kafka