kafka topic中消息消费offset保存在哪里

Kafka 中，消费者消费进度（offset）的保存方式取决于消费者的消费模式。

在旧版的消费者 API 中，消费进度是由消费者自己保存在 ZooKeeper 中的。消费者会定期将消费进度提交到 ZooKeeper 中，并在消费者启动时从 ZooKeeper 中读取消费进度。这种方式需要额外的 ZooKeeper 支持，并且对 ZooKeeper 的压力较大，因此在新版的消费者 API 中已经不再使用。

在新版的消费者 API 中，消费进度的保存方式由消费者自行控制。消费者可以选择将消费进度保存在 Kafka 的 Topic 中，也可以选择将消费进度保存在外部存储系统中，如 Redis、MySQL 等。如果选择将消费进度保存在 Kafka 的 Topic 中，Kafka 会为每个消费者组维护一个特殊的内置 Topic，专门用来保存消费进度。消费者在消费消息时，会从该内置 Topic 中读取消费进度，并将消费进度提交回该 Topic 中。该方式不需要额外的存储支持，并且能够保证消费进度的高可用性。

相关问题

kafka topic中消息如何判断被消费了

在 Kafka 中，消息被消费是根据消费者的消费进度来判断的。每个消费者在消费消息时，都会保存一个消费进度（offset），用于标记已经消费的消息的位置。当消费者从 Kafka 中拉取消息时，Kafka 会返回最新的消息和消费者上一次拉取的位置之间的所有消息，消费者消费这些消息后，将消费进度更新为最新拉取的位置，以便下一次拉取时不会重复消费已经消费过的消息。

在 Kafka 中，通过消费者组的方式来协调多个消费者消费同一个 Topic 中的消息。消费者组中的每个消费者都会负责消费 Topic 中的一部分消息，并且消费者组会协调每个消费者的消费进度，确保每个消息只会被消费一次。因此，只要消息被消费者成功消费，消费进度就会被更新，Kafka 就会认为该消息已经被消费了。

在不同方法中kafka消费不同topic

可以使用多线程的方式，在每个线程中消费不同的topic。具体做法如下：

1. 创建多个消费者对象，每个消费者对象消费不同的topic。
2. 将每个消费者对象放入一个线程中，并启动线程。
3. 在每个线程中，使用消费者对象消费对应的topic。

示例代码：

public class KafkaConsumerDemo {
    private static final String TOPIC_1 = "topic1";
    private static final String TOPIC_2 = "topic2";

    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(2);
        executor.submit(new ConsumerThread(TOPIC_1));
        executor.submit(new ConsumerThread(TOPIC_2));
    }

    private static class ConsumerThread implements Runnable {
        private final String topic;

        public ConsumerThread(String topic) {
            this.topic = topic;
        }

        @Override
        public void run() {
            Properties props = new Properties();
            props.put("bootstrap.servers", "localhost:9092");
            props.put("group.id", "test-group");
            props.put("enable.auto.commit", "true");
            props.put("auto.commit.interval.ms", "1000");
            props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
            consumer.subscribe(Arrays.asList(topic));

            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("topic = %s, partition = %d, offset = %d, key = %s, value = %s\n",
                            record.topic(), record.partition(), record.offset(), record.key(), record.value());
                }
            }
        }
    }
}

上面的代码中，创建了两个消费者线程，一个消费topic1，一个消费topic2。在每个线程中，创建了一个KafkaConsumer对象，用于消费对应的topic。在循环中，使用poll方法获取消息，并处理消息。