spring-kafka使消费者动态订阅新增的topic问题

要解决spring-kafka消费者动态订阅新增的topic问题，可以通过以下步骤完成：

步骤一：配置动态topic管理器

动态topic管理器可以监听主题变化并动态更新topic列表。spring-kafka可以通过自定义Topic管理器实现：

@Component
public class DynamicTopicManager implements ApplicationListener<ContextRefreshedEvent> {

    private KafkaAdmin kafkaAdmin;
    private KafkaListenerEndpointRegistry registry;

    public DynamicTopicManager(KafkaAdmin kafkaAdmin, KafkaListenerEndpointRegistry registry) {
        this.kafkaAdmin = kafkaAdmin;
        this.registry = registry;
    }

    @Override
    public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
        kafkaAdmin.initialize();
        List<String> topicNames = new ArrayList<>(kafkaAdmin.listTopics().keySet());
        for (String topicName : topicNames) {
            KafkaMessageListenerContainer<String, String> container = (KafkaMessageListenerContainer<String, String>) registry
                    .getListenerContainer(topicName);
            if (container == null) {
                ContainerProperties containerProperties = new ContainerProperties(topicName);
                containerProperties.setMessageListener(new MessageListener() {
                    @Override
                    public void onMessage(Object o) { }
                });
                container = new KafkaMessageListenerContainer<>(kafkaAdmin.getConfigurationProperties(),
                        containerProperties);
                container.start();
                registry.registerListenerContainer(container);
            }
        }
    }
}

在该类中，我们使用Spring的ApplicationListener接口来监听ContextRefreshedEvent事件，一旦应用程序上下文启动，就调用onApplicationEvent方法。在该方法中，我们获取kafkaAdmin中的所有topic名字并为每个topic创建一个KafkaMessageListenerContainer并将其注册到KafkaListenerEndpointRegistry中。

步骤二：实现动态创建消费者监听器

在步骤一中，我们创建了KafkaMessageListenerContainer对象用于监听已存在的topic。现在我们需要动态创建消费者监听器。为此，我们定义了一个Topic监视器类：

@Component
public class TopicMonitor {

    private KafkaListenerEndpointRegistry registry;

    @Autowired
    public TopicMonitor(KafkaListenerEndpointRegistry registry) {
        this.registry = registry;
    }

    public void monitor(String topicName) {
        KafkaMessageListenerContainer<String, String> container = new KafkaMessageListenerContainer<>(kafkaConfig,
                new ContainerProperties(topicName));
        container.setupMessageListener(new MessageListener<String, String>() {
            @Override
            public void onMessage(ConsumerRecord<String, String> record) {
                System.out.println("Received Message: " + record.value());
            }
        });
        container.start();
        registry.registerListenerContainer(container);
    }
}

在动态监测器中，我们构建了一个KafkaMessageListenerContainer对象，并为其设置了消息监听器，其中定义了ConsumerRecord类型以处理消息。该方法还将container注册到KafkaListenerEndpointRegistry中，以便spring-kafka可以自动管理它。

现在，我们已经准备好动态添加消费者监听器。为了验证方案，我们来看两个示例：

示例一：基于REST API动态创建消费者

我们可以使用Spring提供的REST API来动态创建新的Consumer。这种动态创建消费者的方法是基于HTTP请求，因此不需要修改代码。这是非常实用的，特别是在生产环境中。

定义一个REST Controller：

@RestController
public class ConsumerRestController {

    @Autowired
    private TopicMonitor topicMonitor;

    @PostMapping("/consumer")
    public ResponseEntity<Void> register(@RequestParam("topic") String topic) {
        topicMonitor.monitor(topic);
        return ResponseEntity.ok().build();
    }

}

在该控制器中，我们使用TopicMonitor类中定义的monitor方法，并注入到控制器中。然后我们通过POST请求将topic名称传递给该方法。一旦topic名称被传递，我们的方法将通过该名称构建一个KafkaMessageListenerContainer对象。

示例二：基于JMS队列动态创建消费者

我们可以通过集成java消息服务（JMS）队列来动态创建新的消费者。在这个例子中，我们使用DynamicDestinationResolver类来转换topic名字为队列（Destination）名字。

@Component
public class JmsConsumer {

    @Autowired
    private JmsListenerContainerFactory jmsListenerContainerFactory;

    @Autowired
    private DynamicDestinationResolver destinationResolver;

    public void registry(String topic) throws NamingException {

        Destination destination = destinationResolver.resolveDestinationName(null, topic, false);
        JmsMessageListenerContainer container = jmsListenerContainerFactory.createListenerContainer(destination);
        container.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                System.out.println("Received Message: " + message.toString());
            }
        });
        container.start();
    }
}

在JmsMessageListenerContainer中，我们定义了一个消息监听器，并使用jmsListenerContainerFactory创建了一个新增的静态消费者。在该类中，我们还使用DynamicDestinationResolver类将topic名称转换为队列名称，并使用该名称为JMS消费者构建目标（Destination）。

在总体上，我们可以通过定义一个动态topic管理器来监听topic并自动更新Spring-Kafka容器的注册表。使用这种方法，我们可以轻松地创建新的消费者，并避免耦合到代码中。