为主题动态创建消息列表

Question

1. 主题
2. GroupID
3. 分区
4. 筛选器（如果有）

我能想到的唯一方法是，我们可以在解析xml配置后创建messagelisteners并且每个主题都有自己的concurrentmessagelistenercontainer。

使用spring是否有其他更好的方法可用？

附注：我对Spring和Kafka并不陌生。如果在解释要求时有混淆，请告诉我

谢谢,Rajasekhar

Answer 1

也许你可以用话题模式。看看消费者属性。例如。听众

@KafkaListener(topicPattern = "topic1|topic2")

将侦听topic1和topic2。

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

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");