@kafkaListener while(true)

`@KafkaListener`注解标注的方法会一直监听指定的Kafka Topic，一旦有新的消息到达Topic，就会自动调用该方法进行消费。

在方法内部，使用`while(true)`循环的方式可以保证方法一直处于监听状态，不会停止。也就是说，只要应用程序在运行，就会一直监听Kafka Topic，一旦有新的消息到达，就会立即进行消费。

但是需要注意的是，这种方式可能会导致应用程序一直处于运行状态，无法正常退出。因此，在实际使用中，需要根据具体需求来决定是否需要使用`while(true)`循环，以及如何优雅地停止应用程序。例如，可以使用一些优雅退出的方式，如设置一个标志位，当收到退出信号时，将标志位设置为true，然后跳出循环，执行必要的清理工作，最后正常退出应用程序。

相关问题

@kafkalistener监听topic 实现代码

下面是一个简单的 Kafka Consumer 监听 Topic 的实现代码，使用 Kafka 的 Java API：

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Collections;
import java.util.Properties;

public class KafkaListener {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("test-topic"));

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

这个代码会连接到本地的 Kafka Broker，使用 `test-group` 作为消费者组名，监听 `test-topic` 这个 Topic。每当有新的消息到达 Topic 时，就会通过 `poll()` 方法从 Consumer 中取出来并处理。在这个例子中，只是简单地打印消息的 offset、key 和 value。

不使用 KafkaListener 注解，实现对应功能

可以使用 KafkaConsumer API 来实现 Kafka 消费者的功能。以下是一个示例代码，可以实现从 Kafka 主题中消费消息：

import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

public class MyKafkaConsumer {

    public static void main(String[] args) {

        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092"); // Kafka 服务器地址
        props.put("group.id", "my-group"); // 消费者组 ID
        props.put("key.deserializer", StringDeserializer.class.getName()); // 键反序列化器
        props.put("value.deserializer", StringDeserializer.class.getName()); // 值反序列化器

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        consumer.subscribe(Arrays.asList("my-topic")); // 订阅主题

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(1000); // 拉取消息
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Received message: " + record.value()); // 处理消息
            }
        }

        consumer.close(); // 关闭消费者
    }
}

这个示例代码创建了一个 Kafka 消费者，使用订阅的方式从主题中拉取消息，并对每条消息进行处理。可以根据需要对代码进行修改，以满足具体的业务需求。