kafka 动态添加监听 java

在 Kafka 中动态添加监听可以通过以下步骤来实现：

1. 创建 KafkaConsumer 实例，指定所需的配置参数。
2. 调用 KafkaConsumer 的 subscribe() 方法，注册要订阅的主题。
3. 创建一个新的线程，在该线程中启动 KafkaConsumer 实例的轮询方法 poll()。
4. 在轮询线程中，使用 KafkaConsumer 的 assignment() 方法获取当前已分配的分区列表。
5. 如果需要动态添加监听器，使用 KafkaConsumer 的 assign() 方法分配新的分区，并在分区分配完成后调用 seekToBeginning() 方法将消费者指针移动到该分区的开头。

下面是一个简单的 Java 代码示例：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

public class DynamicListenerExample {
    private KafkaConsumer<String, String> consumer;

    public DynamicListenerExample(Properties props) {
        this.consumer = new KafkaConsumer<>(props);
    }

    public void subscribe(List<String> topics) {
        consumer.subscribe(topics);
        new Thread(new PollingThread()).start();
    }

    private class PollingThread implements Runnable {
        @Override
        public void run() {
            while (true) {
                consumer.poll(100);
                List<TopicPartition> partitions = consumer.assignment();
                // do something with the partitions
            }
        }
    }

    public void addListener(String topic) {
        TopicPartition partition = new TopicPartition(topic, 0);
        consumer.assign(Arrays.asList(partition));
        consumer.seekToBeginning(Arrays.asList(partition));
    }
}

在上面的示例中，subscribe() 方法用于注册要订阅的主题，并启动消费者的轮询线程。addListener() 方法用于动态添加监听器，将消费者指针移动到指定分区的开头。在轮询线程中，可以使用 consumer.assignment() 方法获取当前已分配的分区列表，并在需要时调用 addListener() 方法。