在 Spring Cloud Kafka 中，可以通过配置 KafkaListenerEndpointRegistrar 和 KafkaMessageConverter 来实现消息保留策略。以下是一个简单的示例：
application.yml 或 application.properties 文件中配置 Kafka 相关的属性：spring: cloud: kafka: bootstrap-servers: localhost:9092 consumer: group-id: my-group auto-offset-reset: earliest key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer producer: key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.apache.kafka.common.serialization.StringSerializer KafkaListenerEndpointRegistrar 和 KafkaMessageConverter：import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.common.serialization.StringDeserializer; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.annotation.KafkaListenerConfigurer; import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; import org.springframework.kafka.config.KafkaListenerEndpointRegistrar; import org.springframework.kafka.config.MethodKafkaListenerEndpoint; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.listener.ConcurrentMessageListenerContainer; import org.springframework.kafka.listener.config.MethodKafkaListenerEndpointRegistrar; import org.springframework.kafka.listener.config.MethodKafkaListenerEndpointRegistry; import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer; import org.springframework.kafka.support.serializer.JsonDeserializer; import org.springframework.kafka.support.serializer.StringDeserializer; import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer; import org.springframework.kafka.support.serializer.JsonDeserializer; import java.util.HashMap; import java.util.Map; @Configuration public class KafkaConfig implements KafkaListenerConfigurer { @Value("${spring.kafka.bootstrap-servers}") private String bootstrapServers; @Bean public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() { ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(consumerFactory()); factory.setProducerFactory(producerFactory()); return factory; } @Bean public ConsumerFactory<String, String> consumerFactory() { Map<String, Object> props = new HashMap<>(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group"); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); return new DefaultKafkaConsumerFactory<>(props); } @Bean public ProducerFactory<String, String> producerFactory() { Map<String, Object> props = new HashMap<>(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); return new DefaultKafkaProducerFactory<>(props); } @Bean public KafkaTemplate<String, String> kafkaTemplate() { return new KafkaTemplate<>(producerFactory()); } @Override public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) { MethodKafkaListenerEndpointRegistry registry = new MethodKafkaListenerEndpointRegistry(); registrar.setEndpointRegistries(registry); registry.register(messageListenerEndpoint()); } @Bean public MethodKafkaListenerEndpoint<String, String> messageListenerEndpoint() { MethodKafkaListenerEndpoint<String, String> endpoint = new MethodKafkaListenerEndpoint<>(); endpoint.setId("myMessageListener"); endpoint.setTopics("my-topic"); endpoint.setMessageHandlerMethodFactory(kafkaListenerContainerFactory().getMessageHandlerMethodFactory()); return endpoint; } } @KafkaListener 注解来监听特定的主题：import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Service; @Service public class MyKafkaConsumer { @KafkaListener(id = "myMessageListener", topics = "my-topic") public void listen(String message) { System.out.println("Received message: " + message); } } 在这个示例中，我们配置了一个简单的消费者，它会监听名为 “my-topic” 的主题。Kafka 会根据消息保留策略将消息存储在本地磁盘上，直到它们被消费或过期。具体的保留策略取决于你的 Kafka 配置，例如 log.retention.hours、log.retention.bytes 或 log.retention.ms 等属性。
